pubsub

.gitlab-ci.yml

 image: "transit/kubectl-plus:v1.19.4"
 
 stages:
-    - prepare
-    - build
-    - deploy
-    - terminate
+- prepare
+- build
+- deploy
+- terminate
+
+secrets:
+  stage: prepare
+  script:
+    - kubectl delete secret gitlab-registry-credentials || true
+    - "kubectl create secret docker-registry gitlab-registry-credentials \
+            --docker-server=$CI_REGISTRY \
+            --docker-username=image-registry \ 
+            --docker-password=$CI_REGISTRY_TOKEN \ 
+        "
 
 volumes:
     stage: prepare
     script:
     - kubectl apply -f volumes/redis-pvc.yaml
 
+consumer:
+  stage: build
+  image:
+    name: gcr.io/kaniko-project/executor:debug
+    entrypoint: [""]
+  only:
+    changes:
+    - .gitlab-ci.yml
+    - consumer/*
+    - consumer/*/*
+  script:
+  - mkdir -p /kaniko/.docker
+  - echo "{\"auths\":{\"$CI_REGISTRY\":{\"username\":\"$CI_REGISTRY_USER\",\"password\":\"$CI_REGISTRY_PASSWORD\"}}}" > /kaniko/.docker/config.json
+  - cd consumer
+  - /kaniko/executor --context . --destination $CI_REGISTRY_IMAGE/consumer:latest
+
+producer:
+  stage: build
+  image:
+    name: gcr.io/kaniko-project/executor:debug
+    entrypoint: [""]
+  only:
+    changes:
+    - .gitlab-ci.yml
+    - producer/*
+    - producer/*/*
+  script:
+  - mkdir -p /kaniko/.docker
+  - echo "{\"auths\":{\"$CI_REGISTRY\":{\"username\":\"$CI_REGISTRY_USER\",\"password\":\"$CI_REGISTRY_PASSWORD\"}}}" > /kaniko/.docker/config.json
+  - cd producer
+  - /kaniko/executor --context . --destination $CI_REGISTRY_IMAGE/producer:latest
+
 redis:
    stage: deploy
    script:
    - kubectl delete -f deploy/redis-dep.yaml || true
    - kubectl apply -f deploy/redis-dep.yaml
    - kubectl apply -f deploy/redis-svc.yaml
+
+pubsub-consumer:
+    stage: deploy
+    script:
+    - mo deploy/pubsub-consumer-dep.yaml | kubectl delete -f - || true
+    - mo deploy/pubsub-consumer-dep.yaml | kubectl apply -f -
+
+pubsub-producer:
+    stage: deploy
+    script:
+    - mo deploy/pubsub-producer-dep.yaml | kubectl delete -f - || true
+    - mo deploy/pubsub-producer-dep.yaml | kubectl apply -f -

consumer/Dockerfile

+FROM python:3.9-slim
+
+COPY Requirements.txt /app/Requirements.txt
+RUN pip3 install -r /app/Requirements.txt
+
+COPY . /app/
+WORKDIR /app
+
+ENTRYPOINT ["python3", "-u", "pubsub.py"]

consumer/Requirements.txt

+redis
\ No newline at end of file

consumer/pubsub.py

+import os, time, json
+import redis
+
+host = os.environ.get("REDIS_SVC", "redis-svc")
+port = int(os.environ.get("REDIS_PORT", "6379"))
+channel = os.environ.get("CHANNEL", "xfan")
+
+r = redis.Redis(host=host, port=port, db=0)
+
+ch = r.pubsub()
+ch.subscribe(channel)
+
+for msg in ch.listen():
+    try:
+        r = time.time_ns()
+        data = json.loads(msg['data'].decode('utf8'))
+        latency = (r - data['timestamp']) / 1000 / 1000
+        print(f"{latency:.2f}ms | msg: { json.dumps(data) }")
+    except Exception as ex:
+        print(f"Could not process message '{ msg }' caused by { ex }")
+

deploy/pubsub-consumer-dep.yaml

+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: pubsub-consumer
+spec:
+  replica: 1
+  selector:
+    matchLabels:
+      app: pubsub-consumer
+  template:
+    metadata:
+      labels:
+        app: pubsub-consumer
+    spec:
+      imagePullSecrets:
+      - name: gitlab-registry-credentials
+      containers:
+      - name: pubsub-consumer
+        image: {{ CI_REGISTRY_IMAGE }}/consumer:latest

deploy/pubsub-producer-dep.yaml

+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: pubsub-producer
+spec:
+  replica: 1
+  selector:
+    matchLabels:
+      app: pubsub-producer
+  template:
+    metadata:
+      labels:
+        app: pubsub-producer
+    spec:
+      imagePullSecrets:
+      - name: gitlab-registry-credentials
+      containers:
+      - name: pubsub-producer
+        image: {{ CI_REGISTRY_IMAGE }}/producer:latest

producer/Dockerfile

+FROM python:3.9-slim
+
+COPY Requirements.txt /app/Requirements.txt
+RUN pip3 install -r /app/Requirements.txt
+
+COPY . /app/
+WORKDIR /app
+
+ENTRYPOINT ["python3", "-u", "pubsub.py"]

producer/Requirements.txt

+redis
\ No newline at end of file

producer/pubsub.py

+import os, time, random, json
+import redis
+
+host = os.environ.get("REDIS_SVC", "redis-svc")
+port = int(os.environ.get("REDIS_PORT", "6379"))
+channel = os.environ.get("CHANNEL", "xfan")
+
+stream = redis.Redis(host=host, port=port, db=0)
+
+i = 1
+while True:
+    msg = f"Message { i }"
+    i = i + 1 
+    n = random.randint(1, 3)
+    time.sleep(n)
+    print(f"Publishing message '{ msg }' to channel '{ channel }'")
+    try:
+        stream.publish(channel, json.dumps({
+            "timestamp": time.time_ns(),
+            "message": msg
+        }))
+    except Exception as ex:
+        print(f"Could not publish message '{msg}' caused by {ex}")