Coverage for scripts / live_release_validation / checks / inference_inventory.py: 100.00%
147 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Kubernetes inventory and stable-absence proof for live inference validation."""
3from __future__ import annotations
5import json
6import time
7from dataclasses import dataclass
8from typing import Any
10from .inference_common import ManagedInferenceValidationError
13@dataclass(frozen=True)
14class KubernetesInventoryKind:
15 """One namespaced resource kind that may be owned by an endpoint."""
17 summary_key: str
18 resource: str
19 optional_api: bool = False
22KUBERNETES_INVENTORY_KINDS: tuple[KubernetesInventoryKind, ...] = (
23 KubernetesInventoryKind("deployments", "deployments.apps"),
24 KubernetesInventoryKind("replica_sets", "replicasets.apps"),
25 KubernetesInventoryKind("pods", "pods"),
26 KubernetesInventoryKind("services", "services"),
27 KubernetesInventoryKind("endpoints", "endpoints"),
28 KubernetesInventoryKind("endpoint_slices", "endpointslices.discovery.k8s.io"),
29 KubernetesInventoryKind("hpas", "horizontalpodautoscalers.autoscaling"),
30 KubernetesInventoryKind("scaled_objects", "scaledobjects.keda.sh", optional_api=True),
31 KubernetesInventoryKind("config_maps", "configmaps"),
32 KubernetesInventoryKind("generated_admin_secrets", "secrets"),
33 KubernetesInventoryKind("legacy_ingresses", "ingresses.networking.k8s.io"),
34 KubernetesInventoryKind(
35 "legacy_http_routes", "httproutes.gateway.networking.k8s.io", optional_api=True
36 ),
37)
40class InferenceInventoryMixin:
41 """Mixin for bounded kubectl inventory and two-observation absence proof."""
43 settings: Any
44 kubectl: Any
46 def _persist(self) -> None: # pragma: no cover - implemented by lifecycle
47 raise NotImplementedError
49 def _record_failure(
50 self, record: dict[str, Any], stage: str, exc: BaseException
51 ) -> None: # pragma: no cover - implemented by lifecycle
52 raise NotImplementedError
54 def _strong_get(
55 self, record: dict[str, Any]
56 ) -> dict[str, Any] | None: # pragma: no cover - implemented by lifecycle
57 raise NotImplementedError
59 @staticmethod
60 def _truncated(value: str) -> str: # pragma: no cover - implemented by lifecycle
61 raise NotImplementedError
63 @staticmethod
64 def _optional_api_missing(stderr: str) -> bool:
65 lowered = stderr.casefold()
66 return (
67 "the server doesn't have a resource type" in lowered
68 or "could not find the requested resource" in lowered
69 )
71 @staticmethod
72 def _not_found(stderr: str) -> bool:
73 lowered = stderr.casefold()
74 return "notfound" in lowered or "not found" in lowered
76 def _kubectl_json(
77 self,
78 record: dict[str, Any],
79 *arguments: str,
80 optional_api: bool = False,
81 deadline: float | None = None,
82 ) -> Any | None:
83 timeout = float(self.settings.command_timeout_seconds)
84 if deadline is not None:
85 remaining = deadline - time.monotonic()
86 if remaining <= 0:
87 raise ManagedInferenceValidationError(
88 "managed inference Kubernetes read exceeded its phase deadline"
89 )
90 timeout = min(timeout, remaining)
91 try:
92 code, stdout, stderr = self.kubectl(*arguments, timeout=timeout)
93 except Exception as exc:
94 self._record_failure(record, "kubectl", exc)
95 raise ManagedInferenceValidationError(
96 "managed inference Kubernetes read failed; inspect the private checkpoint"
97 ) from None
98 if code != 0:
99 if optional_api and self._optional_api_missing(stderr):
100 return {"items": []}
101 if self._not_found(stderr):
102 return None
103 record["last_kubectl_error"] = {
104 "argv": list(arguments),
105 "returncode": code,
106 "stdout": self._truncated(stdout),
107 "stderr": self._truncated(stderr),
108 }
109 self._persist()
110 raise ManagedInferenceValidationError(
111 "managed inference Kubernetes read failed; inspect the private checkpoint"
112 )
113 try:
114 return json.loads(stdout)
115 except json.JSONDecodeError as exc:
116 self._record_failure(record, "kubectl-json", exc)
117 raise ManagedInferenceValidationError(
118 "managed inference Kubernetes read returned invalid JSON"
119 ) from None
121 @staticmethod
122 def _owned_inventory_item(
123 summary_key: str,
124 item: dict[str, Any],
125 endpoint_name: str,
126 owned_replica_sets: set[str],
127 ) -> bool:
128 """Classify one Kubernetes object using exact endpoint ownership."""
129 metadata = item.get("metadata") if isinstance(item, dict) else None
130 if not isinstance(metadata, dict):
131 return False
132 name = metadata.get("name")
133 labels = metadata.get("labels")
134 labels = labels if isinstance(labels, dict) else {}
135 deployment_names = {
136 endpoint_name,
137 f"{endpoint_name}-canary",
138 f"{endpoint_name}-prefill",
139 f"{endpoint_name}-decode",
140 f"{endpoint_name}-proxy",
141 }
142 service_names = set(deployment_names)
143 exact_names: dict[str, set[str]] = {
144 "deployments": deployment_names,
145 "services": service_names,
146 "hpas": {
147 endpoint_name,
148 f"{endpoint_name}-prefill",
149 f"{endpoint_name}-decode",
150 f"keda-hpa-{endpoint_name}",
151 f"keda-hpa-{endpoint_name}-prefill",
152 f"keda-hpa-{endpoint_name}-decode",
153 },
154 "scaled_objects": {
155 endpoint_name,
156 f"{endpoint_name}-prefill",
157 f"{endpoint_name}-decode",
158 },
159 "config_maps": {f"{endpoint_name}-mooncake", f"{endpoint_name}-pd-proxy"},
160 "generated_admin_secrets": {f"{endpoint_name}-admin"},
161 "legacy_ingresses": {
162 endpoint_name,
163 f"{endpoint_name}-canary",
164 f"{endpoint_name}-proxy",
165 },
166 "legacy_http_routes": {
167 endpoint_name,
168 f"{endpoint_name}-canary",
169 f"{endpoint_name}-proxy",
170 },
171 }
172 if summary_key in exact_names:
173 return isinstance(name, str) and name in exact_names[summary_key]
174 if summary_key == "endpoints":
175 return isinstance(name, str) and name in service_names
176 if summary_key == "endpoint_slices":
177 return labels.get("kubernetes.io/service-name") in service_names
178 if summary_key not in {"replica_sets", "pods"}:
179 return False
181 app_name = labels.get("app")
182 if app_name not in deployment_names:
183 return False
184 if labels.get("project") != "gco" or labels.get("gco.io/type") != "inference":
185 return False
186 owner_references = metadata.get("ownerReferences")
187 owners = owner_references if isinstance(owner_references, list) else []
188 if not owners:
189 return True
190 for owner in owners:
191 if not isinstance(owner, dict):
192 continue
193 owner_kind = owner.get("kind")
194 owner_name = owner.get("name")
195 if owner_kind == "Deployment" and owner_name in deployment_names:
196 return True
197 if (
198 summary_key == "pods"
199 and owner_kind == "ReplicaSet"
200 and owner_name in owned_replica_sets
201 ):
202 return True
203 return False
205 def kubernetes_inventory(
206 self,
207 record: dict[str, Any],
208 *,
209 deadline: float | None = None,
210 ) -> dict[str, list[str]]:
211 """Inventory endpoint-owned resources using exact deterministic identity."""
212 endpoint_name = str(record["name"])
213 inventory: dict[str, list[str]] = {}
214 owned_replica_sets: set[str] = set()
215 for kind in KUBERNETES_INVENTORY_KINDS:
216 payload = self._kubectl_json(
217 record,
218 "get",
219 kind.resource,
220 "--namespace",
221 self.settings.namespace,
222 "--output",
223 "json",
224 optional_api=kind.optional_api,
225 deadline=deadline,
226 )
227 items = payload.get("items", []) if isinstance(payload, dict) else []
228 if not isinstance(items, list):
229 raise ManagedInferenceValidationError(
230 "managed inference Kubernetes inventory is malformed"
231 )
232 names: list[str] = []
233 for item in items:
234 if not isinstance(item, dict) or not self._owned_inventory_item(
235 kind.summary_key,
236 item,
237 endpoint_name,
238 owned_replica_sets,
239 ):
240 continue
241 metadata = item.get("metadata")
242 name = metadata.get("name") if isinstance(metadata, dict) else None
243 if isinstance(name, str):
244 names.append(name)
245 if kind.summary_key == "replica_sets":
246 owned_replica_sets.add(name)
247 inventory[kind.summary_key] = sorted(set(names))
248 record["last_kubernetes_inventory"] = inventory
249 self._persist()
250 return inventory
252 def absence_snapshot(
253 self,
254 record: dict[str, Any],
255 *,
256 deadline: float | None = None,
257 ) -> tuple[bool, dict[str, Any]]:
258 """Read strong DDB state plus the complete Kubernetes ownership inventory."""
259 if deadline is not None and time.monotonic() >= deadline:
260 raise ManagedInferenceValidationError(
261 "managed inference endpoint absence was not proven before timeout"
262 )
263 item = self._strong_get(record)
264 inventory = self.kubernetes_inventory(record, deadline=deadline)
265 counts = {key: len(names) for key, names in inventory.items()}
266 absent = item is None and not any(counts.values())
267 record["last_absence_observation"] = {
268 "ddb_present": item is not None,
269 "kubernetes": inventory,
270 }
271 self._persist()
272 return absent, {"ddb_absent": item is None, "kubernetes_counts": counts}
274 def prove_absence(self, record: dict[str, Any]) -> dict[str, Any]:
275 """Require two full absent sweeps separated by one monitor interval."""
276 deadline = time.monotonic() + self.settings.deletion_timeout_seconds
277 consecutive = 0
278 observations = record.setdefault("absence_observations", [])
279 if not isinstance(observations, list):
280 observations = []
281 record["absence_observations"] = observations
282 while True:
283 absent, evidence = self.absence_snapshot(record, deadline=deadline)
284 evidence = {**evidence, "observed_at_monotonic": time.monotonic()}
285 observations.append(evidence)
286 if len(observations) > 20:
287 del observations[:-20]
288 consecutive = consecutive + 1 if absent else 0
289 record["consecutive_absent_observations"] = consecutive
290 self._persist()
291 if consecutive >= 2:
292 record["absence_proven"] = True
293 record["cleanup_phase"] = "absent"
294 stable = {
295 **evidence,
296 "stable_absence_observations": consecutive,
297 "separated_by_seconds": self.settings.monitor_interval_seconds,
298 }
299 record["stable_absence_evidence"] = stable
300 self._persist()
301 return stable
302 remaining = deadline - time.monotonic()
303 if remaining <= 0:
304 raise ManagedInferenceValidationError(
305 "managed inference endpoint absence was not proven before timeout"
306 )
307 delay = (
308 self.settings.monitor_interval_seconds
309 if absent
310 else self.settings.poll_interval_seconds
311 )
312 time.sleep(min(float(delay), remaining))