ea0452b9a9
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
260 lines
11 KiB
Python
Executable file
260 lines
11 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Capture non-secret cluster capacity, utilization, and allocation evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
CPU_FACTORS = {"n": 0.000001, "u": 0.001, "m": 1.0, "": 1000.0}
|
|
BYTE_FACTORS = {
|
|
"": 1,
|
|
"Ki": 1024,
|
|
"Mi": 1024**2,
|
|
"Gi": 1024**3,
|
|
"Ti": 1024**4,
|
|
"K": 1000,
|
|
"M": 1000**2,
|
|
"G": 1000**3,
|
|
"T": 1000**4,
|
|
}
|
|
|
|
|
|
def cpu_millicores(value: str | None) -> int:
|
|
match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(n|u|m)?", str(value or "0"))
|
|
if not match:
|
|
raise ValueError(f"unsupported CPU quantity: {value}")
|
|
return round(float(match.group(1)) * CPU_FACTORS[match.group(2) or ""])
|
|
|
|
|
|
def bytes_value(value: str | None) -> int:
|
|
match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)(Ki|Mi|Gi|Ti|K|M|G|T)?", str(value or "0"))
|
|
if not match:
|
|
raise ValueError(f"unsupported byte quantity: {value}")
|
|
return round(float(match.group(1)) * BYTE_FACTORS[match.group(2) or ""])
|
|
|
|
|
|
def workload_name(pod: dict[str, Any]) -> str:
|
|
metadata = pod.get("metadata") or {}
|
|
labels = metadata.get("labels") or {}
|
|
for key in ("app.kubernetes.io/part-of", "app.kubernetes.io/name", "app"):
|
|
if labels.get(key):
|
|
return str(labels[key])
|
|
owners = metadata.get("ownerReferences") or []
|
|
if owners and owners[0].get("name"):
|
|
return str(owners[0]["name"])
|
|
return re.sub(r"-[a-f0-9]{6,}(?:-[a-z0-9]{5})?$", "", str(metadata.get("name", "unknown")))
|
|
|
|
|
|
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():
|
|
fields = line.split()
|
|
if pods and len(fields) >= 4:
|
|
result[(fields[0], fields[1])] = {
|
|
"cpu_millicores": cpu_millicores(fields[2]),
|
|
"memory_bytes": bytes_value(fields[3]),
|
|
}
|
|
elif not pods and len(fields) >= 4:
|
|
result[fields[0]] = {
|
|
"cpu_millicores": cpu_millicores(fields[1]),
|
|
# kubectl top nodes: NAME CPU CPU% MEMORY MEMORY%
|
|
"memory_bytes": bytes_value(fields[3]),
|
|
}
|
|
return result
|
|
|
|
|
|
def build_observation(snapshots: dict[str, Any], captured_at: str) -> dict[str, Any]:
|
|
gaps: list[str] = []
|
|
node_top = parse_top(snapshots.get("node_top", ""), pods=False)
|
|
pod_top = parse_top(snapshots.get("pod_top", ""), pods=True)
|
|
if snapshots.get("node_top") is None:
|
|
gaps.append("node metrics unavailable")
|
|
if snapshots.get("pod_top") is None:
|
|
gaps.append("pod metrics unavailable")
|
|
|
|
nodes = []
|
|
alloc_cpu = alloc_memory = alloc_pods = 0
|
|
for node in snapshots["nodes"].get("items") or []:
|
|
meta = node.get("metadata") or {}
|
|
status = node.get("status") or {}
|
|
alloc = status.get("allocatable") or {}
|
|
labels = meta.get("labels") or {}
|
|
name = str(meta.get("name"))
|
|
topology = {
|
|
"region": labels.get("topology.kubernetes.io/region"),
|
|
"zone": labels.get("topology.kubernetes.io/zone"),
|
|
"hostname": labels.get("kubernetes.io/hostname"),
|
|
"provider_id": (node.get("spec") or {}).get("providerID"),
|
|
}
|
|
if not topology["region"] or not topology["zone"]:
|
|
gaps.append(f"node {name} lacks independent region/zone labels")
|
|
row = {
|
|
"name": name,
|
|
"ready": any(c.get("type") == "Ready" and c.get("status") == "True" for c in status.get("conditions") or []),
|
|
"allocatable": {
|
|
"cpu_millicores": cpu_millicores(alloc.get("cpu")),
|
|
"memory_bytes": bytes_value(alloc.get("memory")),
|
|
"ephemeral_storage_bytes": bytes_value(alloc.get("ephemeral-storage")),
|
|
"pods": int(alloc.get("pods") or 0),
|
|
},
|
|
"topology": topology,
|
|
"observed": node_top.get(name),
|
|
}
|
|
nodes.append(row)
|
|
alloc_cpu += row["allocatable"]["cpu_millicores"]
|
|
alloc_memory += row["allocatable"]["memory_bytes"]
|
|
alloc_pods += row["allocatable"]["pods"]
|
|
|
|
workloads: dict[tuple[str, str], dict[str, Any]] = {}
|
|
for pod in snapshots["pods"].get("items") or []:
|
|
meta = pod.get("metadata") or {}
|
|
namespace = str(meta.get("namespace", "default"))
|
|
name = workload_name(pod)
|
|
key = (namespace, name)
|
|
row = workloads.setdefault(key, {
|
|
"namespace": namespace,
|
|
"workload": name,
|
|
"pods": 0,
|
|
"requests": {"cpu_millicores": 0, "memory_bytes": 0},
|
|
"limits": {"cpu_millicores": 0, "memory_bytes": 0},
|
|
"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["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"))))
|
|
if metric and row["observed"] is not None:
|
|
row["observed"]["cpu_millicores"] += metric["cpu_millicores"]
|
|
row["observed"]["memory_bytes"] += metric["memory_bytes"]
|
|
|
|
storage_classes = []
|
|
for item in snapshots["storageclasses"].get("items") or []:
|
|
storage_classes.append({
|
|
"name": (item.get("metadata") or {}).get("name"),
|
|
"provisioner": item.get("provisioner"),
|
|
"reclaim_policy": item.get("reclaimPolicy"),
|
|
"volume_binding_mode": item.get("volumeBindingMode"),
|
|
"default": (item.get("metadata") or {}).get("annotations", {}).get("storageclass.kubernetes.io/is-default-class") == "true",
|
|
})
|
|
pv_capacity = sum(bytes_value(((item.get("spec") or {}).get("capacity") or {}).get("storage")) for item in snapshots["pvs"].get("items") or [])
|
|
pvc_rows = []
|
|
pvc_requested = 0
|
|
for item in snapshots["pvcs"].get("items") or []:
|
|
meta = item.get("metadata") or {}
|
|
spec = item.get("spec") or {}
|
|
requested = bytes_value(((spec.get("resources") or {}).get("requests") or {}).get("storage"))
|
|
pvc_requested += requested
|
|
pvc_rows.append({
|
|
"namespace": meta.get("namespace"),
|
|
"name": meta.get("name"),
|
|
"storage_class": spec.get("storageClassName"),
|
|
"requested_bytes": requested,
|
|
"phase": (item.get("status") or {}).get("phase"),
|
|
})
|
|
|
|
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)
|
|
return {
|
|
"schema_version": "railiance.cluster-resource-observation.v1",
|
|
"record_type": "usage_observation",
|
|
"resource_id": "resource:railiance:reef-railiance:k3s",
|
|
"source": "railiance-cluster",
|
|
"workplan_id": "RCLUSTER-WP-0014",
|
|
"reef": "reef-railiance",
|
|
"captured_at": captured_at,
|
|
"capacity": {
|
|
"nodes": nodes,
|
|
"totals": {"cpu_millicores": alloc_cpu, "memory_bytes": alloc_memory, "pods": alloc_pods, "pv_bytes": pv_capacity},
|
|
},
|
|
"utilization": {"workloads": workload_rows},
|
|
"storage": {"classes": storage_classes, "claims": pvc_rows},
|
|
"failure_domain": {
|
|
"current_contract": "single Ready node on railiance01; local storage is correlated with node loss",
|
|
"independent_domains_claimed": 1 if nodes else 0,
|
|
"threephoenix_target_is_live": False,
|
|
},
|
|
"allocation": {
|
|
"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},
|
|
"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)},
|
|
"selection_owner": "resource-control",
|
|
},
|
|
"measurement_gaps": sorted(set(gaps)),
|
|
"provenance": {
|
|
"commands": [
|
|
"kubectl get nodes -o json", "kubectl get pods -A -o json",
|
|
"kubectl get pvc -A -o json", "kubectl get pv -o json",
|
|
"kubectl get storageclass -o json", "kubectl top nodes --no-headers",
|
|
"kubectl top pods -A --no-headers",
|
|
],
|
|
"secret_surfaces_read": False,
|
|
},
|
|
}
|
|
|
|
|
|
def run_json(command: list[str]) -> dict[str, Any]:
|
|
return json.loads(subprocess.check_output(command, text=True))
|
|
|
|
|
|
def run_optional(command: list[str]) -> str | None:
|
|
completed = subprocess.run(command, text=True, capture_output=True, check=False)
|
|
return completed.stdout if completed.returncode == 0 else None
|
|
|
|
|
|
def collect(kubectl: str, remote: str | None = None) -> dict[str, Any]:
|
|
base = ["ssh", "-o", "BatchMode=yes", remote, kubectl] if remote else [kubectl]
|
|
snapshots = {
|
|
"nodes": run_json(base + ["get", "nodes", "-o", "json"]),
|
|
"pods": run_json(base + ["get", "pods", "-A", "-o", "json"]),
|
|
"pvcs": run_json(base + ["get", "pvc", "-A", "-o", "json"]),
|
|
"pvs": run_json(base + ["get", "pv", "-o", "json"]),
|
|
"storageclasses": run_json(base + ["get", "storageclass", "-o", "json"]),
|
|
"node_top": run_optional(base + ["top", "nodes", "--no-headers"]),
|
|
"pod_top": run_optional(base + ["top", "pods", "-A", "--no-headers"]),
|
|
}
|
|
captured_at = datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
return build_observation(snapshots, captured_at)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--kubectl", default="kubectl")
|
|
parser.add_argument("--remote", help="Optional SSH host on which kubectl runs")
|
|
parser.add_argument("--output-dir", type=Path)
|
|
args = parser.parse_args()
|
|
try:
|
|
observation = collect(args.kubectl, args.remote)
|
|
except (OSError, subprocess.CalledProcessError, json.JSONDecodeError, ValueError) as exc:
|
|
print(f"cluster observation failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
rendered = json.dumps(observation, indent=2) + "\n"
|
|
if args.output_dir:
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
stamp = observation["captured_at"].replace(":", "").replace("-", "")
|
|
destination = args.output_dir / f"{stamp}.json"
|
|
destination.write_text(rendered, encoding="utf-8")
|
|
(args.output_dir / "latest.json").write_text(rendered, encoding="utf-8")
|
|
print(destination)
|
|
else:
|
|
sys.stdout.write(rendered)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|