Publish pending unscheduled pods in cluster observations.

RAIL-BS-WP-0014: residual millicores count scheduled pods only.
state_hub_preflight_observation matches the STATE-WP-0091 fixture shape.

Assistant: grok
Assistant-Session: 01a09dc1-b21e-77e1-919e-fcad2f82b267

tools/observe_cluster_resources.py

@@ -53,6 +53,22 @@ def workload_name(pod: dict[str, Any]) -> str:
     return re.sub(r"-[a-f0-9]{6,}(?:-[a-z0-9]{5})?$", "", str(metadata.get("name", "unknown")))
 
 
+def pod_effective_requests(pod: dict[str, Any]) -> tuple[int, int]:
+    """Pod request is max(sum(containers), max(init containers))."""
+    spec = pod.get("spec") or {}
+    containers = spec.get("containers") or []
+    inits = spec.get("initContainers") or []
+
+    def req(container: dict[str, Any], key: str) -> str | None:
+        return ((container.get("resources") or {}).get("requests") or {}).get(key)
+
+    cpu = sum(cpu_millicores(req(c, "cpu")) for c in containers)
+    mem = sum(bytes_value(req(c, "memory")) for c in containers)
+    init_cpu = max((cpu_millicores(req(c, "cpu")) for c in inits), default=0)
+    init_mem = max((bytes_value(req(c, "memory")) for c in inits), default=0)
+    return max(cpu, init_cpu), max(mem, init_mem)
+
+
 def parse_top(lines: str | None, *, pods: bool) -> dict[tuple[str, str] | str, dict[str, int]]:
     result: dict[tuple[str, str] | str, dict[str, int]] = {}
     for line in (lines or "").splitlines():
@@ -114,9 +130,35 @@ def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str,
         alloc_pods += row["allocatable"]["pods"]
 
     workloads: dict[tuple[str, str], dict[str, Any]] = {}
+    pending_unscheduled: list[dict[str, Any]] = []
+    node_allocated: dict[str, dict[str, int]] = {
+        node["name"]: {"cpu_millicores": 0, "memory_bytes": 0} for node in nodes
+    }
+    scheduled_cpu = scheduled_memory = 0
     for pod in snapshots["pods"].get("items") or []:
         meta = pod.get("metadata") or {}
+        spec = pod.get("spec") or {}
+        status = pod.get("status") or {}
+        phase = str(status.get("phase") or "")
+        if phase in {"Succeeded", "Failed"}:
+            continue
         namespace = str(meta.get("namespace", "default"))
+        pod_name = str(meta.get("name"))
+        cpu, mem = pod_effective_requests(pod)
+        node_name = spec.get("nodeName")
+        if phase == "Pending" and not node_name:
+            pending_unscheduled.append({
+                "namespace": namespace,
+                "name": pod_name,
+                "cpu_millicores": cpu,
+                "memory_bytes": mem,
+            })
+            continue
+        if node_name in node_allocated:
+            node_allocated[node_name]["cpu_millicores"] += cpu
+            node_allocated[node_name]["memory_bytes"] += mem
+        scheduled_cpu += cpu
+        scheduled_memory += mem
         name = workload_name(pod)
         key = (namespace, name)
         row = workloads.setdefault(key, {
@@ -128,15 +170,13 @@ def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str,
             "observed": {"cpu_millicores": 0, "memory_bytes": 0} if snapshots.get("pod_top") is not None else None,
         })
         row["pods"] += 1
-        for container in (pod.get("spec") or {}).get("containers") or []:
-            resources = container.get("resources") or {}
-            requests = resources.get("requests") or {}
-            limits = resources.get("limits") or {}
-            row["requests"]["cpu_millicores"] += cpu_millicores(requests.get("cpu"))
-            row["requests"]["memory_bytes"] += bytes_value(requests.get("memory"))
+        row["requests"]["cpu_millicores"] += cpu
+        row["requests"]["memory_bytes"] += mem
+        for container in spec.get("containers") or []:
+            limits = (container.get("resources") or {}).get("limits") or {}
             row["limits"]["cpu_millicores"] += cpu_millicores(limits.get("cpu"))
             row["limits"]["memory_bytes"] += bytes_value(limits.get("memory"))
-        metric = pod_top.get((namespace, str(meta.get("name"))))
+        metric = pod_top.get((namespace, pod_name))
         if metric and row["observed"] is not None:
             row["observed"]["cpu_millicores"] += metric["cpu_millicores"]
             row["observed"]["memory_bytes"] += metric["memory_bytes"]
@@ -167,8 +207,22 @@ def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str,
         })
 
     workload_rows = sorted(workloads.values(), key=lambda row: (row["namespace"], row["workload"]))
-    requested_cpu = sum(row["requests"]["cpu_millicores"] for row in workload_rows)
-    requested_memory = sum(row["requests"]["memory_bytes"] for row in workload_rows)
+    requested_cpu = scheduled_cpu
+    requested_memory = scheduled_memory
+    pending_cpu = sum(item["cpu_millicores"] for item in pending_unscheduled)
+    pending_memory = sum(item["memory_bytes"] for item in pending_unscheduled)
+    preflight_nodes = []
+    for node in nodes:
+        allocated = node_allocated.get(node["name"], {"cpu_millicores": 0, "memory_bytes": 0})
+        preflight_nodes.append({
+            "name": node["name"],
+            "ready": node["ready"],
+            "unschedulable": False,
+            "allocatable_cpu_m": node["allocatable"]["cpu_millicores"],
+            "allocatable_memory_bytes": node["allocatable"]["memory_bytes"],
+            "allocated_cpu_m": allocated["cpu_millicores"],
+            "allocated_memory_bytes": allocated["memory_bytes"],
+        })
     return {
         "schema_version": "railiance.cluster-resource-observation.v1",
         "record_type": "usage_observation",
@@ -192,9 +246,29 @@ def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str,
             "method_version": "cluster-raw-drivers-v1",
             "drivers": ["cpu_requests", "memory_requests", "pvc_requested_bytes", "observed_usage"],
             "denominators": {"cpu_requested_millicores": requested_cpu, "memory_requested_bytes": requested_memory, "pvc_requested_bytes": pvc_requested},
+            "pending_unscheduled": {
+                "cpu_millicores": pending_cpu,
+                "memory_bytes": pending_memory,
+                "pods": pending_unscheduled,
+            },
             "residual_capacity": {"cpu_millicores": max(alloc_cpu - requested_cpu, 0), "memory_bytes": max(alloc_memory - requested_memory, 0), "pv_bytes": max(pv_capacity - pvc_requested, 0)},
+            "not_a_scheduling_guarantee": True,
             "selection_owner": "resource-control",
         },
+        "state_hub_preflight_observation": {
+            "schema": "state-hub.release-headroom-preflight.v1-input",
+            "observed_at": captured_at,
+            "freshness_seconds": 0,
+            "nodes": preflight_nodes,
+            "pending_unrelated": [
+                {
+                    "name": f"{item['namespace']}/{item['name']}",
+                    "cpu_m": item["cpu_millicores"],
+                    "memory_bytes": item["memory_bytes"],
+                }
+                for item in pending_unscheduled
+            ],
+        },
         "measurement_gaps": sorted(set(gaps)),
         "provenance": {
             "commands": [