Feature/SK-486 | Implement kubernetes event listener (#1)

* Post to studio api

* Added login

* Only push to api if apps are not deleted

* tar file deleted

* Start using environment variables

* print reponse text

* Build and push image to github registry workflow added

* Deployment file altered for use with docker compose + Workflow altered to be run on main branch

.github/workflows/deploy-image.yml

@@ -0,0 +1,52 @@
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+# GitHub recommends pinning actions to a commit SHA.
+# To get a newer version, you will need to update the SHA.
+# You can also reference a tag or branch, but the action may change without warning.
+
+name: Create and publish a Docker image
+
+on:
+  push:
+    branches: ['main']
+  pull_request:
+    branches: ['main']
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  build-and-push-image:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@65b78e6e13532edd9afa3aa52ac7964289d1a9c1
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@9ec57ed1fcdbf14dcef7dfbe97b2010124a938b7
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@f2a1d5e99d037542a71f64918e516c093c6f3fc4
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.gitignore

@@ -158,3 +158,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+#Other
+*-local*

deploy/deployment.yaml

@@ -1,3 +1,4 @@
+# Intended to be used with docker-compose file in the studio or stackn repository.
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -18,10 +19,8 @@ spec:
         app: stackn-studio
         allow-api-access: "true"
     spec:
-      automountServiceAccountToken: true
-      serviceAccountName: studio
       containers:
-      - image: k3d-registry:35187/studio-event:latest
+      - image: ghcr.io/scaleoutsystems/studio-kube-controller:feature-sk-486
         imagePullPolicy: Always
         name: studio-event-listener
         resources:
@@ -31,4 +30,17 @@ spec:
           requests:
             cpu: 50m
             memory: 20Mi
+        env:
+        - name: EVENT_LISTENER_USERNAME
+          value: "<username-set-in-docker-compose-file>"
+        - name: EVENT_LISTENER_PASSWORD
+          value: "<password-set-in-docker-compose-file>"
+        - name: STUDIO_SERVICE_NAME
+          value: "<host-machine-ip>"
+        - name: STUDIO_SERVICE_PORT
+          value: "8080"
+        - name: APP_STATUS_ENDPOINT
+          value: "api/app/status"
+        - name: TOKEN_ENDPOINT
+          value: "api/token-auth/"
       restartPolicy: Always

event_listener.py

@@ -1,28 +1,180 @@
 from kubernetes import client, config, watch
 
+import os
+import requests
+import time
 
-def init_event_listener():
-    config.incluster_config.load_incluster_config()
+STUDIO_SERVICE_NAME = os.environ.get("STUDIO_SERVICE_NAME", None)
+STUDIO_SERVICE_PORT = os.environ.get("STUDIO_SERVICE_PORT", None)
+APP_STATUS_ENDPOINT = os.environ.get("APP_STATUS_ENDPOINT", None)
+APP_STATUSES_ENDPOINT = os.environ.get("APP_STATUSES_ENDPOINT", None)
+TOKEN_ENDPOINT = os.environ.get("TOKEN_ENDPOINT", None)
 
-    api = client.CoreV1Api()
-    w = watch.Watch()
+BASE_URL = f"http://{STUDIO_SERVICE_NAME}:{STUDIO_SERVICE_PORT}"
+APP_STATUS_URL = f"{BASE_URL}/{APP_STATUS_ENDPOINT}"
+APP_STATUSES_URL = f"{BASE_URL}/{APP_STATUSES_ENDPOINT}"
+TOKEN_URL = f"{BASE_URL}/{TOKEN_ENDPOINT}"
 
-    label_selector = "type=app"
-    namespace = "default"
+USERNAME = os.environ.get("EVENT_LISTENER_USERNAME", None)
+PASSWORD = os.environ.get("EVENT_LISTENER_PASSWORD", None)
 
-    for event in w.stream(api.list_namespaced_pod, namespace=namespace, label_selector=label_selector):
-        pod = event["object"]
+token = None
+
+# Number of retries
+max_retries = 10
+
+# Time to wait between retries (in seconds)
+retry_interval = 10
+
+config.incluster_config.load_incluster_config()
+
+api = client.CoreV1Api()
+w = watch.Watch()
+
+label_selector = "type=app"
+namespace = "default"
+
+latest_status = {}
+
+
+def get_token():
+    req = {
+        "username": USERNAME,
+        "password": PASSWORD,
+    }
+
+    for retry in range(max_retries):
+        print(f"retry: {retry}")
+
+        try:
+            res = requests.post(TOKEN_URL, json=req, verify=False)
+
+            if res.status_code == 200:
+                resp = res.json()
+
+                if "token" in resp:
+                    print("Token retrieved successfully.")
+                    global token
+                    token = resp["token"]
+                    return True
+                else:
+                    print("Failed to fetch token.")
+                    print(res.text)
+
+        except requests.exceptions.RequestException:
+            # An exception occurred, service is not responding
+            if retry == max_retries - 1:
+                # Maximum retries reached, handle the failure
+                print("Service did not respond.")
+                break
+
+            # Wait for the specified interval before retrying
+            time.sleep(retry_interval)
+
+    return False
+
+
+def sync_all_statuses():
+    values = ""
+
+    for pod in api.list_namespaced_pod(
+        namespace=namespace, label_selector=label_selector
+    ).items:
         status = pod.status.phase
         release = pod.metadata.labels["release"]
-        print(f"EVENT_TYPE: {event['type']}", flush=True)
-        print(f"STATUS: {status}", flush=True)
-        send_status_to_rest_api(release, status)
-
-def send_status_to_rest_api(release, status):
-    """
-    TODO: Send the updated status phase to studio REST-API endpoint
-    """
-    print(f"UPDATE: {release} to {status}", flush=True)
-
-if __name__ == '__main__':
-    init_event_listener()
+
+        values += f"{release}:{status},"
+
+    data = {"values": values}
+
+    print(f"DATA: {data}", flush=True)
+    print("Syncing all statuses...", flush=True)
+
+    post(APP_STATUSES_URL, data=data)
+
+
+def init_event_listener():
+    for event in w.stream(
+        api.list_namespaced_pod,
+        namespace=namespace,
+        label_selector=label_selector,
+    ):
+        pod = event["object"]
+
+        status = get_status(pod)
+
+        print(f"Synchronizing status: {status}", flush=True)
+
+        # status = pod.status.phase
+        release = pod.metadata.labels["release"]
+
+        event_type = event["type"]
+
+        if latest_status.get(release) == status:
+            print("Status not changed, skipping...")
+
+        latest_status[release] = status
+
+        if event_type != "DELETED":
+            print(f"EVENT_TYPE: {event_type}", flush=True)
+            print(f"STATUS: {status}", flush=True)
+
+            data = {
+                "release": release,
+                "status": status,
+            }
+
+            post(APP_STATUS_URL, data=data)
+
+
+def get_status(pod):
+    print("Getting status...")
+
+    container_statuses = pod.status.container_statuses
+
+    if container_statuses is not None:
+        for container_status in container_statuses:
+            state = container_status.state
+
+            if state is not None:
+                terminated = state.terminated
+
+                if terminated is not None:
+                    return terminated.reason
+
+                waiting = state.waiting
+
+                if waiting is not None:
+                    return waiting.reason
+
+                running = state.running
+
+                if running is not None:
+                    return "Running"
+
+            print("Last state not found.")
+    else:
+        print("Container statuses not found.")
+
+    return pod.status.phase
+
+
+def post(url, data):
+    try:
+        headers = {"Authorization": f"Token {token}"}
+
+        response = requests.post(url, data=data, headers=headers, verify=False)
+
+        print(f"RESPONSE STATUS CODE: {response.status_code}")
+        print(f"RESPONSE TEXT: {response.text}")
+
+    except requests.exceptions.RequestException:
+        print("Service did not respond.")
+
+
+if __name__ == "__main__":
+    success = get_token()
+
+    if success:
+        sync_all_statuses()
+        init_event_listener()

requirements.txt

@@ -1 +1,2 @@
-kubernetes
+kubernetes
+requests