event_listener.py

from kubernetes import client, config, watch

import os
import requests

import logging

# Configure logging
logging.basicConfig(
    level=logging.DEBUG,  # Adjust as needed (DEBUG, ERROR, WARNING, INFO)
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    handlers=[
        # logging.FileHandler("event_listener.log"),  # Log to a file
        logging.StreamHandler()  # Also log to the console
    ]
)


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)

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}"

logging.debug(f"STUDIO_SERVICE_NAME: {STUDIO_SERVICE_NAME}")
logging.debug(f"STUDIO_SERVICE_PORT: {STUDIO_SERVICE_PORT}")
logging.debug(f"APP_STATUS_ENDPOINT: {APP_STATUS_ENDPOINT}")
logging.debug(f"APP_STATUSES_ENDPOINT: {APP_STATUSES_ENDPOINT}")

K8S_STATUS_MAP = {
    "CrashLoopBackOff": "Error",
    "Completed": "Retrying...",
    "ContainerCreating": "Created",
    "PodInitializing": "Pending",
}

config.incluster_config.load_incluster_config()

api = client.CoreV1Api()
w = watch.Watch()

label_selector = "type=app"
namespace = "default"

latest_status = {}


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"]

        values += f"{release}:{status},"

    data = {"values": values}

    logging.debug(f"DATA: {data}")
    logging.debug("Syncing all statuses...")

    post(APP_STATUSES_URL, data=data)


def init_event_listener():
    resource_version = ""
    while True:
        try:
            for event in w.stream(
                api.list_namespaced_pod,
                namespace=namespace,
                label_selector=label_selector,
                timeout_seconds=20,
                resource_version=resource_version,
            ):
                pod = event["object"]

                resource_version = pod.metadata.resource_version
                status = get_status(pod)

                logging.info(f"Synchronizing status: {status}")

                # status = pod.status.phase
                release = pod.metadata.labels["release"]

                event_type = event["type"]

                if latest_status.get(release) == status:
                    logging.info("Status not changed, skipping...")

                latest_status[release] = status

                if event_type != "DELETED":
                    logging.info(f"EVENT_TYPE: {event_type}")
                    logging.info(f"STATUS: {status}")

                    data = {
                        "release": release,
                        "status": status,
                    }

                    post(APP_STATUS_URL, data=data)
        except client.exceptions.ApiException as e:
            if e.status == 410:
                # If the resource version is too old, log the issue and reset resource_version
                logging.error("Resource version expired, restarting watch from latest version.")
                resource_version = None
            else:
                logging.error(f"An error occurred: {e}")
            continue
        


def get_status(pod):
    """
    Returns the status of a pod by looping through each container
    and checking the 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:
                    reason = terminated.reason
                    return mapped_status(reason)

                waiting = state.waiting

                if waiting is not None:
                    reason = waiting.reason
                    return mapped_status(reason)
        else:
            running = state.running
            ready = container_status.ready
            if running and ready:
                return "Running"
            else:
                return "Pending"

    return pod.status.phase


def mapped_status(reason: str) -> str:
    return K8S_STATUS_MAP.get(reason, reason)


def post(url: str, data: dict):
    try:
        response = requests.post(url, data=data, verify=False)

        logging.info(f"RESPONSE STATUS CODE: {response.status_code}")
        logging.info(f"RESPONSE TEXT: {response.text}")

    except requests.exceptions.RequestException as err:
        logging.error(err)
        logging.error("Service did not respond.")


if __name__ == "__main__":
    logging.info("Starting event listener...")

    sync_all_statuses()
    init_event_listener()