k3s-health-dashboard/app/k8s_client.py

import os
from datetime import datetime, timezone
from kubernetes import client, config

_CONFIG_LOADED = False

def _ensure_config():
    global _CONFIG_LOADED
    if _CONFIG_LOADED:
        return
    try:
        if os.getenv("IN_CLUSTER"):
            config.load_incluster_config()
        else:
            # Load local kubeconfig
            config_path = os.getenv("KUBECONFIG", os.path.expanduser("~/.kube/config"))
            config.load_kube_config(config_file=config_path)
    except Exception:
        # If we can't load config (e.g. in tests where we mock), we don't want to crash here
        # but let the client creation fail later or be mocked.
        pass
    _CONFIG_LOADED = True

def get_core():
    """
    Returns an initialized Kubernetes CoreV1Api client.
    Matches the name expected by test fixtures.
    """
    _ensure_config()
    return client.CoreV1Api()

def get_core_client() -> client.CoreV1Api:
    """Legacy alias for get_core() if needed."""
    return get_core()

def get_custom_client() -> client.CustomObjectsApi:
    _ensure_config()
    return client.CustomObjectsApi()

def fetch_nodes():
    v1 = get_core()
    nodes = v1.list_node().items
    result = []
    for n in nodes:
        conditions = {c.type: c.status for c in n.status.conditions} if n.status.conditions else {}
        
        # Role logic
        labels = n.metadata.labels or {}
        roles = []
        if "node-role.kubernetes.io/control-plane" in labels:
            roles.append("control-plane")
        if "node-role.kubernetes.io/master" in labels:
            roles.append("master")
        
        role_str = ", ".join(roles) if roles else "worker"

        result.append({
            "name": n.metadata.name,
            "ready": conditions.get("Ready") == "True",
            "roles": role_str,
            "age": _format_age(n.metadata.creation_timestamp),
            "cpu_cap": _parse_millicores(n.status.capacity.get("cpu", "0")),
            "mem_cap_mi": _parse_mebibytes(n.status.capacity.get("memory", "0")),
            "mem_alloc_mi": _parse_mebibytes(n.status.allocatable.get("memory", "0")),
            "os": n.status.node_info.os_image,
            "kubelet_version": n.status.node_info.kubelet_version,
            "mem_pressure": conditions.get("MemoryPressure", "False"),
            "disk_pressure": conditions.get("DiskPressure", "False"),
            "pid_pressure": conditions.get("PIDPressure", "False"),
        })
    return result

def fetch_pods(namespace="all"):
    v1 = get_core()
    if namespace == "all":
        pods = v1.list_pod_for_all_namespaces().items
    else:
        pods = v1.list_namespaced_pod(namespace=namespace).items
    
    result = []
    for p in pods:
        result.append({
            "namespace": p.metadata.namespace,
            "name": p.metadata.name,
            "ready": _is_pod_ready(p),
            "phase": p.status.phase or "Unknown",
            "restarts": _get_pod_restarts(p),
            "node": p.spec.node_name or "—",
            "age": _format_age(p.metadata.creation_timestamp),
        })
    return result

def _is_pod_ready(pod):
    if not pod.status.conditions:
        return False
    for c in pod.status.conditions:
        if c.type == "Ready":
            return c.status == "True"
    return False

def _get_pod_restarts(pod):
    if not pod.status.container_statuses:
        return 0
    return sum(c.restart_count for c in pod.status.container_statuses)

def _format_age(ts):
    if not ts:
        return "—"
    now = datetime.now(timezone.utc)
    delta = now - ts
    if delta.days > 0:
        return f"{delta.days}d"
    hours = delta.seconds // 3600
    if hours > 0:
        return f"{hours}h"
    minutes = delta.seconds // 60
    return f"{minutes}m"

def _parse_millicores(val):
    if not val:
        return 0.0
    if isinstance(val, (int, float)):
        return float(val)
    if val.endswith("m"):
        return float(val[:-1]) / 1000.0
    return float(val)

def _parse_mebibytes(val):
    if not val:
        return 0.0
    if isinstance(val, (int, float)):
        return float(val)
    
    # Simple parser for Ki, Mi, Gi, Ti
    suffix_map = {
        "Ki": 1/1024,
        "Mi": 1,
        "Gi": 1024,
        "Ti": 1024*1024,
    }
    
    for suffix, multiplier in suffix_map.items():
        if val.endswith(suffix):
            return float(val[:-len(suffix)]) * multiplier
    
    # Bytes to MiB if no suffix
    return float(val) / (1024 * 1024)