Integrate platform with Kubernetes cluster discovery

Introduction

When you build an internal developer platform that provisions namespaces, enforces quotas, and schedules agent workloads, stale cluster state is one of the most damaging failure modes you can ship. A platform acting on yesterday's node inventory will over-provision into drained nodes, accept quota requests the cluster cannot satisfy, and silently fail users who expect self-service to just work. By the end of this lesson, you'll be able to wire a live Kubernetes discovery worker into your platform's control plane — using watch streams, PostgreSQL upserts, and metrics snapshots — so every platform decision is grounded in current cluster reality.

Key Terminology

• Watch Stream — A push-based event delivery channel provided by the Kubernetes API server, opened in Python via watch.Watch().stream(), that yields ADDED, MODIFIED, and DELETED events for resources like namespaces in sub-second latency without requiring the worker to poll on a timer.
• Resource Version — A monotonically increasing cursor embedded in every Kubernetes object's metadata.resource_version that marks the exact cluster state at the moment of an event; passing it back into watch_namespaces via the resource_version kwarg lets the worker resume a broken stream from the disconnection point, avoiding both event loss and a full-list replay.
• Idempotent Upsert — A PostgreSQL INSERT … ON CONFLICT pattern used in sync_node_metrics and the namespace reconcile loop that makes repeated writes safe: duplicate metric scrapes are silently discarded with DO NOTHING, and existing namespace rows are refreshed in place with DO UPDATE SET, so the worker can restart without corrupting data.
• In-Cluster Config — The service-account token and CA bundle automatically mounted into every pod, loaded by config.load_incluster_config(), that grants the discovery worker access to the Kubernetes API server without embedded credentials; the associated RBAC role must grant get, list, and watch on namespaces, nodes, and pods.
• Metrics Server — The metrics.k8s.io/v1beta1 aggregated API queried via CustomObjectsApi in fetch_node_metrics, which exposes rolling per-node CPU and memory usage as Kubernetes quantity strings (e.g., 234m, 512Ki) that _parse_cpu and _parse_mem normalize into nanocores and bytes for storage.
• Discovery Worker — The long-running run_discovery_worker coroutine that owns the watch stream, tracks last_rv across reconnects, and upserts live namespace and node state into the platform's PostgreSQL tables, serving as the single source of cluster truth that the control plane consults for provisioning decisions.

Concepts

Why Platform Decisions Require Live Cluster State

An internal developer platform sits above the Kubernetes API server: it accepts self-service requests — provision a namespace, schedule an agent workload, enforce a quota — translates them into cluster operations, and reports outcomes back to users. Every one of those translations assumes the platform's internal view of the cluster is accurate. When that view is stale, provisioning decisions fail silently. The platform accepts a request it cannot fulfill, the downstream Kubernetes operation lands on a drained node or violates an unmet quota, and the user experiences a broken self-service guarantee.

Stale state is not a theoretical edge case. Auto-scalers add and remove nodes continuously. Teams hand-edit resources via kubectl. Operators install CRDs. A discovery layer that reconciles against the live cluster — rather than trusting its own database as authoritative — is the only way to keep those platform guarantees intact.

Watch Streams vs. Polling

Polling the Kubernetes API on a fixed timer forces a tradeoff between two failure modes: poll too aggressively and you hammer the API server; poll too slowly and you accumulate staleness. Kubernetes's watch mechanism eliminates this tradeoff by inverting the information flow. Instead of the worker asking "what changed?", the API server pushes ADDED, MODIFIED, and DELETED events to the worker as they happen — typically within sub-second latency.

The resource_version field is the API server's monotonically increasing cursor. Threading last_rv through the reconnect path in run_discovery_worker means the worker resumes from the exact disconnection point on every restart — no events replayed from the beginning of history, no gap in coverage (see Code Walkthrough).

Making the Worker Durable Through Idempotent Persistence

A discovery worker that runs continuously must survive network interruptions, mid-scrape restarts, and overlapping metrics polls without corrupting the platform's database. The code enforces this through two complementary idempotent write patterns.

Namespace events use ON CONFLICT (cluster_id, name) DO UPDATE SET — a row that already exists is refreshed in place, so re-processing the same ADDED event twice is harmless. Metrics snapshots use a tighter constraint: ON CONFLICT (cluster_id, node, observed_at) DO NOTHING, where the composite unique key includes the observed_at timestamp. A duplicate scrape of the same 30-second window is silently discarded rather than raising a constraint violation (see Code Walkthrough).

The run_discovery_worker supervisor wraps the watch loop in while True with a sleep(5) backoff, so the worker self-heals after a disconnect without requiring an external process manager to restart it. Combined, idempotent writes and supervisor-loop retry mean the worker can be interrupted, redeployed on top of itself, or reconnected mid-stream without producing duplicate rows or dropping events — which is the durability bar a self-service platform must clear before it can make reliable provisioning decisions.

Code Walkthrough

Now that you understand why live cluster state matters and which failure modes threaten a discovery worker, here is how those concepts translate into running Python code.

Loading Cluster Config and Opening a Watch Stream

The Python kubernetes package opens a push-based watch stream against the API server, so your worker receives namespace and node events in sub-second latency rather than polling on a timer.

Code snippetpython
1from kubernetes import client, config, watch
2from kubernetes.client import CustomObjectsApi
3from sqlalchemy.ext.asyncio import AsyncConnection
4from sqlalchemy import text
5from typing import Iterator
6
7def load_k8s_config() -> None:
8    """Load in-cluster service-account config; fall back to kubeconfig for local dev."""
9    try:
10        config.load_incluster_config()
11    except config.ConfigException:
12        config.load_kube_config()
13
14def watch_namespaces(resource_version: str | None = None) -> Iterator[dict]:
15    """Yield namespace change events. Resumes from resource_version on reconnect."""
16    v1 = client.CoreV1Api()
17    w = watch.Watch()
18    kwargs = {"timeout_seconds": 0}
19    if resource_version:
20        kwargs["resource_version"] = resource_version
21    for event in w.stream(v1.list_namespace, **kwargs):
22        obj = event["object"]
23        yield {
24            "type": event["type"],               # ADDED | MODIFIED | DELETED
25            "name": obj.metadata.name,
26            "labels": obj.metadata.labels or {},
27            "phase": obj.status.phase,
28            "resource_version": obj.metadata.resource_version,
29        }
30
31def _parse_cpu(value: str) -> int:
32    """Convert '234m' → nanocores (int)."""
33    if value.endswith("m"):
34        return int(value[:-1]) * 1_000_000
35    return int(value) * 1_000_000_000
36
37def _parse_mem(value: str) -> int:
38    """Convert '512Ki' → bytes (int)."""
39    suffixes = {"Ki": 1024, "Mi": 1024**2, "Gi": 1024**3}
40    for suffix, factor in suffixes.items():
41        if value.endswith(suffix):
42            return int(value[: -len(suffix)]) * factor
43    return int(value)
44
45def fetch_node_metrics() -> list[dict]:
46    api = CustomObjectsApi()
47    raw = api.list_cluster_custom_object(
48        group="metrics.k8s.io", version="v1beta1", plural="nodes"
49    )
50    return [
51        {
52            "node": item["metadata"]["name"],
53            "cpu_usage_nanocores": _parse_cpu(item["usage"]["cpu"]),
54            "memory_usage_bytes": _parse_mem(item["usage"]["memory"]),
55            "observed_at": item["timestamp"],
56        }
57        for item in raw["items"]
58    ]
59
60async def sync_node_metrics(db: AsyncConnection, cluster_id: str) -> int:
61    metrics = fetch_node_metrics()
62    await db.execute(
63        text("""
64            INSERT INTO cluster_node_metrics
65                (cluster_id, node, cpu_usage_nanocores, memory_usage_bytes, observed_at)
66            VALUES (:cluster_id, :node, :cpu_usage_nanocores, :memory_usage_bytes, :observed_at)
67            ON CONFLICT (cluster_id, node, observed_at) DO NOTHING
68        """),
69        [{"cluster_id": cluster_id, **m} for m in metrics],
70    )
71    return len(metrics)

load_k8s_config tries the in-cluster service-account token first — the pod must carry an RBAC role granting get, list, and watch on namespaces, nodes, and pods. When running locally it falls back to your kubeconfig automatically.

watch_namespaces accepts an optional resource_version so the worker can resume mid-stream after a reconnect without replaying the full namespace list. Pass timeout_seconds=0 to keep the stream open indefinitely; the API server controls pacing and reconnects.

sync_node_metrics writes rolling 30-second snapshots into cluster_node_metrics. The ON CONFLICT DO NOTHING clause makes re-runs idempotent — the unique key is (cluster_id, node, observed_at), so duplicate scrapes are silently discarded rather than raising an error.

Running the Reconcile Loop

Wire the two pieces into a supervisor that persists the last-seen resource_version and retries on watch disconnect:

Code snippetpython
1import asyncio
2import logging
3
4log = logging.getLogger(__name__)
5
6async def run_discovery_worker(db: AsyncConnection, cluster_id: str) -> None:
7    load_k8s_config()
8    last_rv: str | None = None
9    while True:
10        try:
11            for event in watch_namespaces(resource_version=last_rv):
12                last_rv = event["resource_version"]
13                await db.execute(
14                    text("""
15                        INSERT INTO cluster_namespace (cluster_id, name, labels, phase)
16                        VALUES (:cluster_id, :name, :labels, :phase)
17                        ON CONFLICT (cluster_id, name)
18                        DO UPDATE SET labels = EXCLUDED.labels, phase = EXCLUDED.phase
19                    """),
20                    {"cluster_id": cluster_id, **{k: event[k] for k in ("name", "labels", "phase")}},
21                )
22        except Exception as exc:
23            log.warning("Watch stream interrupted (%s); reconnecting in 5s", exc)
24            await asyncio.sleep(5)

The outer while True with exponential-style backoff (sleep(5)) matches the failure-mode guidance in the Concepts section: wrap the watch loop in retry logic and resume from last_rv so no events are lost or duplicated on reconnect.

Confirm that run_discovery_worker is healthy by checking that your cluster_namespace table receives upserts within seconds of a kubectl create namespace smoke-test command against the same cluster.

Do's and Don'ts

Having walked through the watch-stream and idempotent-upsert code above, the following rules distill the operational guardrails that keep run_discovery_worker healthy in production.

Do's

1. ✓Do pass last_rv into watch_namespaces on every reconnect — persisting the last-seen resource_version across retries lets the watch stream resume mid-sequence rather than triggering a full re-list, so no namespace events are lost or duplicated during the asyncio.sleep(5) backoff window.
2. ✓Do declare ON CONFLICT DO NOTHING with a (cluster_id, node, observed_at) composite key in cluster_node_metrics — rolling 30-second scrapes from sync_node_metrics will collide on retries; without this clause a duplicate scrape raises a constraint error instead of being silently discarded, crashing the worker.
3. ✓Do grant the in-cluster service account get, list, and watch on namespaces, nodes, and pods via RBAC before deploying — load_k8s_config authenticates with the pod's service-account token, and a missing verb surfaces as a 403 that kills the watch stream at startup with no retry possible.

Don'ts

1. ✗Don't poll the Kubernetes API on a fixed timer instead of opening a w.stream(v1.list_namespace, timeout_seconds=0) watch — timer-based polling introduces latency windows where the platform acts on stale node inventory, causing quota acceptance for capacity that already moved to drained nodes and silently failing self-service requests.
2. ✗Don't reset last_rv to None inside the except branch of run_discovery_worker — the retry loop preserves last_rv across the reconnect precisely so the watch resumes from the last-seen event; clearing it forces a full re-list and replays every ADDED event as a duplicate upsert against cluster_namespace.
3. ✗Don't use a bare INSERT without ON CONFLICT DO UPDATE SET in the cluster_namespace upsert — a namespace cycling through Active and Terminating phases emits repeated MODIFIED events against the same (cluster_id, name) key; a raw insert fails with a unique-key violation on the second event, halting the reconcile loop.