Capstone: a fleet report tool

# src/fleetreport/core.py
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from kubernetes import client, config

BAD_REASONS = {"CrashLoopBackOff", "ImagePullBackOff", "ErrImagePull"}

@dataclass
class ClusterResult:
    context: str
    version: str = "-"
    nodes: int = 0
    not_ready: list[str] = field(default_factory=list)
    bad_pods: list[str] = field(default_factory=list)
    error: str | None = None

    @property
    def healthy(self) -> bool:
        return self.error is None and not self.not_ready and not self.bad_pods

def check_cluster(context: str, timeout: int) -> ClusterResult:
    res = ClusterResult(context)
    try:
        api_client = config.new_client_from_config(context=context)
        v1 = client.CoreV1Api(api_client)
        res.version = client.VersionApi(api_client).get_code(_request_timeout=timeout).git_version
        nodes = v1.list_node(_request_timeout=timeout).items
        res.nodes = len(nodes)
        for n in nodes:
            ready = next((c.status for c in n.status.conditions if c.type == "Ready"), "Unknown")
            if ready != "True":
                res.not_ready.append(n.metadata.name)
        for p in v1.list_pod_for_all_namespaces(_request_timeout=timeout).items:
            for cs in p.status.container_statuses or []:
                reason = cs.state.waiting.reason if cs.state and cs.state.waiting else None
                if reason in BAD_REASONS:
                    res.bad_pods.append(f"{p.metadata.namespace}/{p.metadata.name} ({reason})")
        # TODO: pods Pending for more than 10 minutes (compare p.status.start_time or creation time)
    except Exception as exc:                      # a cluster failing must not stop the fleet report
        res.error = f"{type(exc).__name__}: {exc}"[:200]
    return res

def check_fleet(contexts: list[str], workers: int, timeout: int) -> list[ClusterResult]:
    results = []
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(check_cluster, c, timeout): c for c in contexts}
        for fut in as_completed(futures):
            results.append(fut.result())
    return sorted(results, key=lambda r: r.context)

# src/fleetreport/cli.py (outline)
# - argparse: --contexts, --workers, --timeout, -o/--output {table,json,csv}, -v
# - contexts: config.list_kube_config_contexts()[0] gives all contexts (names under ["name"])
# - print the report to stdout (table / json.dumps([asdict(r) ...]) / csv)
# - exit 3 if any r.error, else 1 if any not healthy, else 0

# tests/test_fleet.py
from fleetreport import core

def test_unreachable_cluster_is_reported(monkeypatch):
    def fake_check(context, timeout):
        if context == "broken":
            return core.ClusterResult(context, error="ConnectionError: refused")
        return core.ClusterResult(context, version="v1.31.2", nodes=3)
    monkeypatch.setattr(core, "check_cluster", fake_check)

    results = core.check_fleet(["prod", "broken"], workers=2, timeout=5)
    assert [r.context for r in results] == ["broken", "prod"]
    assert results[0].error and not results[0].healthy
    assert results[1].healthy

$ fleetreport
CONTEXT          VERSION   NODES  NOT-READY  BAD-PODS  STATUS
kind-canary      v1.31.0   3      0          0         OK
kind-dev         v1.30.4   3      1          2         FINDINGS
edge-site-042    -         -      -          -         UNREACHABLE (ConnectTimeoutError)
$ echo $?
3