k8s.py

from kubernetes import client, config
from kubernetes.client.rest import ApiException

def load_k8s_config():
    try:
        config.load_incluster_config()   
        print("Loaded in-cluster config.")
    except config.ConfigException:
        config.load_kube_config()   
        print("Loaded kubeconfig file.")

def send_job_to_k8s(job_json, namespace="appollo-scans"):
    load_k8s_config()
    batch_v1 = client.BatchV1Api()
    try:
        res = batch_v1.create_namespaced_job(namespace=namespace, body=job_json)
        return res
    except ApiException as e:
        print(f"Exception when calling BatchV1Api->create_namespaced_job: {e}\n")
        raise


def get_job_status(namespace, job_name):
    """
    Return job status: running, completed, failed, or not_found.

    Does not raise on Kubernetes client errors (403, timeouts, etc.) — returns
    running + message so the dashboard keeps polling instead of HTTP 500.
    """
    try:
        load_k8s_config()
    except Exception as e:
        print(f"[appollo-api] load_k8s_config failed: {e}\n")
        return {
            "status": "running",
            "message": "Kubernetes client configuration unavailable. Check in-cluster SA or kubeconfig.",
        }

    batch_v1 = client.BatchV1Api()
    try:
        job = batch_v1.read_namespaced_job(name=job_name, namespace=namespace)
    except ApiException as e:
        if e.status == 404:
            return {"status": "not_found", "message": "Job not found"}
        # 403 Forbidden, 401, 503, connection errors, etc.
        print(
            f"[appollo-api] read_namespaced_job {namespace}/{job_name}: "
            f"status={getattr(e, 'status', None)} reason={getattr(e, 'reason', e)}\n"
        )
        return {
            "status": "running",
            "message": "Could not read Job from Kubernetes (RBAC, network, or API error). Retrying is OK.",
        }
    except Exception as e:
        print(f"[appollo-api] read_namespaced_job unexpected {namespace}/{job_name}: {e}\n")
        return {
            "status": "running",
            "message": "Unexpected error talking to Kubernetes API.",
        }

    st = job.status
    if st is None:
        return {"status": "running", "active": 0}

    succeeded = st.succeeded if st.succeeded is not None else 0
    failed = st.failed if st.failed is not None else 0
    active = st.active if st.active is not None else 0

    if succeeded >= 1:
        return {"status": "completed", "succeeded": succeeded}
    if failed >= 1:
        return {"status": "failed", "failed": failed}
    return {"status": "running", "active": active}


def get_job_pod_logs(namespace, job_name, tail_lines=3000, max_chars=450_000):
    """
    Read container logs from the pod created for this Job (label job-name=<job_name>).
    Requires RBAC: get/list pods and get pods/log in the job namespace.
    """
    import os

    try:
        tail_lines = int(os.environ.get("SCAN_LOG_TAIL_LINES", str(tail_lines)))
    except ValueError:
        pass
    tail_lines = max(100, min(50_000, tail_lines))
    load_k8s_config()
    v1 = client.CoreV1Api()
    try:
        pods = v1.list_namespaced_pod(
            namespace=namespace,
            label_selector=f"job-name={job_name}",
        )
    except ApiException as e:
        print(f"[appollo-api] list pods for job {job_name}: {e}\n")
        return None
    if not pods.items:
        return None
    ordered = sorted(
        pods.items,
        key=lambda p: (p.metadata.creation_timestamp or ""),
        reverse=True,
    )
    for pod in ordered:
        phase = (pod.status.phase or "").strip()
        if phase not in ("Running", "Succeeded", "Failed", "Unknown"):
            continue
        pname = pod.metadata.name
        try:
            logs = v1.read_namespaced_pod_log(
                name=pname,
                namespace=namespace,
                tail_lines=tail_lines,
            )
        except ApiException as e:
            if e.status in (400, 404):
                continue
            print(f"[appollo-api] pod log {pname}: {e}\n")
            continue
        if not logs:
            continue
        if len(logs) > max_chars:
            logs = logs[-max_chars:]
        return logs
    return None


def list_pods(namespace):
    load_k8s_config()
    v1 = client.CoreV1Api()
    try:
        pods_res = v1.list_namespaced_pod(namespace=namespace)
        for pod in pods_res.items:
            print(f"Pod name: {pod.metadata.name}")
    except ApiException as e:
        print(f"Exception when calling CoreV1Api->list_namespaced_pod: {e}\n")
        raise
 
if __name__ == '__main__':
    job_json = {
         
    }
    response = send_job_to_k8s(job_json)
    # print(response)
    # send_job_to_k8s(job_json)