Coverage for gco / services / inference_monitor.py: 100.00%
2004 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Inference Monitor — reconciliation controller for inference endpoints.
4Runs in each regional EKS cluster and polls the global DynamoDB table
5(gco-inference-endpoints) to reconcile desired state with actual
6Kubernetes resources. Follows a GitOps-style reconciliation pattern:
8 DynamoDB (desired state) → inference_monitor → Kubernetes (actual state)
10The monitor:
11- Creates and reconciles Deployments, ClusterIP Services, and optional autoscalers
12- Leaves public routing on the shared ``gco-system/gco-gateway`` HTTPRoute:
13 ``/inference`` -> ``gco-system/inference-proxy``
14- Removes legacy endpoint-specific Ingresses so upgrades cannot retain a bypass
15- Updates existing deployments when spec changes
16- Scales deployments up/down
17- Tears down resources when endpoints are deleted
18- Reports per-region status back to DynamoDB
20Environment Variables:
21 CLUSTER_NAME: Name of the EKS cluster
22 REGION: AWS region this monitor runs in
23 INFERENCE_ENDPOINTS_TABLE_NAME: DynamoDB table name
24 RECONCILE_INTERVAL_SECONDS: Seconds between reconciliation loops (default: 15)
25 INFERENCE_NAMESPACE: Namespace for inference workloads (default: gco-inference)
26"""
28import asyncio
29import base64
30import contextlib
31import json
32import logging
33import os
34import re
35import secrets
36import signal
37import threading
38import time
39from collections.abc import Callable, Iterator
40from dataclasses import dataclass, field
41from datetime import UTC, datetime
42from functools import partial
43from pathlib import Path
44from typing import Any, NoReturn, TypedDict
45from urllib.parse import urlsplit
47from kubernetes import client, config
48from kubernetes.client.models import V1Deployment
49from kubernetes.client.rest import ApiException
51from gco.services.inference_store import InferenceEndpointStore
52from gco.services.structured_logging import configure_structured_logging
54# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
55# Generated at (UTC): 2026-09-13T13:44:22Z
56# Generated from Git commit: c49331669c66625fecfecf44ae6ab5f95afbfcb4
57# Flowchart(s) generated from this file:
58# * ``InferenceMonitor._reconcile_endpoint_authorized`` -> ``diagrams/code_diagrams/gco/services/inference_monitor.InferenceMonitor__reconcile_endpoint_authorized.html``
59# (PNG: ``diagrams/code_diagrams/gco/services/inference_monitor.InferenceMonitor__reconcile_endpoint_authorized.png``)
60# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
61# <pyflowchart-code-diagram> END
64logging.basicConfig(
65 level=logging.INFO,
66 format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
67)
68logger = logging.getLogger(__name__)
71class NetworkPolicyApplyError(Exception):
72 """An intra-namespace allow rule could not be applied.
74 Raised when materialization-time enforcement fails to create or verify one
75 of the allow rules that disaggregated inference depends on. The default-deny
76 posture is left intact — only the widening allow rule failed — and ``rule``
77 names the offending NetworkPolicy so callers can surface exactly which rule
78 could not be applied.
79 """
81 def __init__(self, rule: str, reason: str):
82 self.rule = rule
83 self.reason = reason
84 super().__init__(f"Network policy {rule!r} could not be applied: {reason}")
87class AdminApiKeySecretError(Exception):
88 """A user-named proxy admin API key Secret is missing or empty.
90 Raised before the prefill-decode proxy is materialized when a Secret named
91 by ``proxy.admin_api_key_secret`` is absent or carries no usable
92 ``ADMIN_API_KEY`` value. An endpoint that names no Secret takes the separate
93 auto-managed path and receives a generated ``{name}-admin`` Secret. The
94 proxy never starts without a usable key, so no proxy Deployment or Service
95 is created on this error. ``secret`` records the rejected Secret name.
96 """
98 def __init__(self, secret: str | None, reason: str):
99 self.secret = secret
100 self.reason = reason
101 named = repr(secret) if secret else "<unnamed>"
102 super().__init__(f"Admin API key Secret {named} is unusable: {reason}")
105# Official AWS CLI v2 multi-architecture image. Keep the readable release tag
106# and immutable manifest-list digest together: both amd64 and arm64 inference
107# nodes resolve through this single verified index.
108AWS_CLI_IMAGE = (
109 "public.ecr.aws/aws-cli/aws-cli:2.36.44@"
110 "sha256:e8467f2c319f9bc9a1471808a69949a76915e9c95eaf4a09ece9f9e85fd32747"
111)
113# Valid TCP port boundaries for KV-transfer bootstrap ports.
114MIN_BOOTSTRAP_PORT = 1024
115MAX_BOOTSTRAP_PORT = 65535
117# vLLM kv_role for each worker role: prefill produces KV, decode consumes it,
118# and a single-instance store node both produces and consumes.
119_KV_ROLE_BY_WORKER_ROLE = {
120 "prefill": "kv_producer",
121 "decode": "kv_consumer",
122 "single": "kv_both",
123}
125# The worker roles each mooncake mode supports. Disaggregated and both split
126# work across prefill/decode; store runs a single kv_both instance.
127_WORKER_ROLES_BY_MODE = {
128 "disaggregated": {"prefill", "decode"},
129 "store": {"single"},
130 "both": {"prefill", "decode"},
131}
133# The EFA RDMA fabric is advertised as a Kubernetes extended resource, gated by
134# a node taint, and selected through a node label. KV cache transfer over
135# RoCE only runs on nodes that carry all three.
136EFA_RESOURCE_NAME = "vpc.amazonaws.com/efa"
137EFA_NODE_SELECTOR_KEY = "efa"
138EFA_NODE_SELECTOR_VALUE = "true"
140# Mooncake KV-transfer role pods are pinned to a dedicated EFA NodePool
141# (mooncake-efa-pool, manifest 46-nodepool-mooncake-efa.yaml) that only offers
142# instance families with >=80GB of GPU memory and FP8-capable Hopper/Blackwell
143# GPUs. The shared training EFA pool (43-nodepool-efa.yaml) also offers p4d
144# (A100 40GB, Ampere, no FP8), which is too small for many disaggregated/store
145# models and can be selected by Karpenter whenever a pod asks only for efa=true.
146# Selecting this extra label keeps role pods off p4d without disturbing the
147# training pool. The value must match the label on the dedicated NodePool.
148MOONCAKE_EFA_NODE_SELECTOR_KEY = "mooncake-efa"
149MOONCAKE_EFA_NODE_SELECTOR_VALUE = "true"
151# The shared per-region Mooncake master exposes its RPC service and the
152# built-in HTTP metadata server on these fixed ports.
153MOONCAKE_MASTER_RPC_PORT = 50051
154MOONCAKE_METADATA_PORT = 8080
155MOONCAKE_MASTER_SERVICE = "mooncake-master"
157# GPU utilization is not a Kubernetes Resource metric, so a native
158# HorizontalPodAutoscaler cannot scale on it (Resource metrics are limited to
159# cpu and memory). The cluster's amazon-cloudwatch-observability agent publishes
160# per-pod GPU utilization to CloudWatch ContainerInsights, so any autoscaler
161# that requests a GPU metric is materialized as a KEDA ScaledObject with an
162# aws-cloudwatch trigger instead. KEDA generates the backing HPA under the hood,
163# and cpu/memory targets ride along as native cpu/memory triggers on the same
164# ScaledObject. KEDA is a mandatory cluster component, so this path is always
165# available.
166KEDA_API_GROUP = "keda.sh"
167KEDA_API_VERSION = "v1alpha1"
168KEDA_SCALEDOBJECT_PLURAL = "scaledobjects"
170# Metric types that can only be served via CloudWatch (KEDA), keyed to the
171# ContainerInsights metric the aws-cloudwatch trigger reads. PodName in the
172# ContainerInsights dimension set is the workload (Deployment) name, so the
173# dimension triple ClusterName/Namespace/PodName yields the average across a
174# Deployment's pods — exactly the signal autoscaling needs.
175GPU_METRIC_NAMESPACE = "ContainerInsights"
176_CLOUDWATCH_METRIC_BY_TYPE = {
177 "gpu": "pod_gpu_utilization",
178 "gpu_memory": "pod_gpu_memory_utilization",
179}
181# Default base port for the KV-transfer bootstrap handshake (VLLM_MOONCAKE_
182# BOOTSTRAP_PORT) and the span of per-worker ports derived from it. vLLM assigns
183# each worker base_port + dp_rank * tp_size + tp_rank, so the intra-namespace
184# allow rule opens a contiguous window starting at the base port. A spec may
185# override the base via mooncake.transfer.bootstrap_base_port.
186MOONCAKE_BOOTSTRAP_BASE_PORT = 8998
187MOONCAKE_BOOTSTRAP_PORT_SPAN = 100
189# Environment variable through which each role pod receives the KV-transfer
190# bootstrap base port. vLLM derives per-worker ports (base + dp_rank * tp_size +
191# tp_rank) from it, so prefill and decode agree on the handshake ports.
192VLLM_MOONCAKE_BOOTSTRAP_PORT_ENV = "VLLM_MOONCAKE_BOOTSTRAP_PORT"
194# Each role pod reads the shared transport settings (metadata-server address,
195# protocol, device) from the rendered mooncake.json. The per-endpoint
196# ``{name}-mooncake`` ConfigMap is mounted read-only at the directory below, and
197# the connector is pointed at the file through MOONCAKE_CONFIG_PATH.
198MOONCAKE_CONFIG_PATH_ENV = "MOONCAKE_CONFIG_PATH"
199MOONCAKE_CONFIG_MOUNT_DIR = "/etc/mooncake"
200MOONCAKE_CONFIG_FILE_PATH = f"{MOONCAKE_CONFIG_MOUNT_DIR}/mooncake.json"
202# Label selector identifying inference workload pods (prefill/decode/proxy and
203# legacy single-Deployment endpoints). Used as both the target and the peer of
204# the intra-namespace allow rules.
205INFERENCE_POD_SELECTOR = {"gco.io/type": "inference"}
208def _inference_service_selector(app: str, **extra: str) -> dict[str, str]:
209 """The ``spec.selector`` of every ClusterIP Service fronting inference pods.
211 The selector carries ``gco.io/type: inference`` next to the per-endpoint
212 ``app`` label, and that is load-bearing under enforced NetworkPolicies on
213 the VPC CNI: its egress probe sees ClusterIP traffic before kube-proxy's
214 DNAT, so the network policy controller admits a Service's ClusterIP only
215 when the Service's own selector matches the policy peer's ``podSelector``
216 (``gco.io/type: inference`` in ``allow-inference-proxy-to-inference`` and
217 ``allow-inference-internal``). With ``app`` alone the pods were admitted
218 and the Service in front of them was not, and the inference proxy's
219 requests timed out against a healthy endpoint.
220 """
221 return {"app": app, **INFERENCE_POD_SELECTOR, **extra}
224# Names of the intra-namespace allow rules the monitor maintains alongside the
225# default-deny posture in gco-inference. These mirror the manifest names in
226# 03-network-policies.yaml so a failure can point at the same object an operator
227# would inspect with kubectl.
228NETWORK_POLICY_INFERENCE_INTERNAL = "allow-inference-internal"
229NETWORK_POLICY_POD_TO_MASTER = "allow-pod-to-master"
230NETWORK_POLICY_POD_TO_METADATA = "allow-pod-to-metadata"
231NETWORK_POLICY_RDMA_BOOTSTRAP = "allow-rdma-bootstrap"
233# Regional configuration keys through which the in-region deployment supplies
234# the shared master's address. The store cannot be wired without an own-region
235# master address, so a store-bearing endpoint defers configuration when the
236# master address is absent or blank. The metadata server defaults to the master
237# host on the metadata port when not supplied explicitly.
238MOONCAKE_MASTER_ADDRESS_ENV = "MOONCAKE_MASTER_ADDRESS"
239MOONCAKE_METADATA_SERVER_ENV = "MOONCAKE_METADATA_SERVER"
241# Container image for the shared per-region master. Supplied by the in-region
242# deployment so the master tracks the same pinned build the manifests use.
243MOONCAKE_MASTER_IMAGE_ENV = "MOONCAKE_MASTER_IMAGE"
245# Maximum time a store-bearing endpoint keeps deferring role-pod creation while
246# the shared master has not reported a Ready replica. Past this window the
247# monitor keeps deferring and stays in the ``creating`` state, but also surfaces
248# an error so operators can see the master never came up. The master itself is
249# never deleted or modified on account of this timeout.
250MOONCAKE_MASTER_READY_TIMEOUT_SECONDS = 600
252# Object-key prefix under which cold-tier KV objects are written in the
253# general-purpose regional bucket. Mirrors the value the regional stack and the
254# `gco inference populate-kv` upload surface use; kept local so the monitor
255# needs no infrastructure (CDK) imports at runtime.
256MOONCAKE_COLD_TIER_KEY_PREFIX = "mooncake-kv"
258# SSM namespace suffix publishing the always-on general-purpose regional
259# bucket's discovery values for a region. The full namespace is
260# ``/<project_name>/regional-shared-bucket`` (see
261# ``constants.regional_shared_ssm_parameter_prefix``); the monitor builds it at
262# runtime from the injected ``PROJECT_NAME`` env var rather than importing the
263# CDK constant, so it needs no infrastructure imports at runtime. Kept as a
264# suffix constant so the ``/name``, ``/arn``, ``/region`` contract stays in one
265# place. See ``_regional_shared_ssm_parameter_prefix``.
266REGIONAL_SHARED_SSM_PARAMETER_SUFFIX = "regional-shared-bucket"
269def _regional_shared_ssm_parameter_prefix() -> str:
270 """Return this deployment's regional-shared-bucket SSM namespace.
272 Built from the ``PROJECT_NAME`` environment variable (default ``"gco"``)
273 so the monitor reads the same project-scoped path the regional stack
274 writes (``/<project_name>/regional-shared-bucket``). Mirrors
275 ``constants.regional_shared_ssm_parameter_prefix`` without importing CDK.
276 """
277 project_name = os.environ.get("PROJECT_NAME", "gco")
278 return f"/{project_name}/{REGIONAL_SHARED_SSM_PARAMETER_SUFFIX}"
281# Matches an AWS region identifier embedded in an address (host or URI), e.g.
282# ``us-east-1``, ``eu-west-2``, ``ap-southeast-1``, ``us-gov-west-1``. KV
283# transfer over RoCE is intra-region, so any address a topology wires to must
284# resolve to the monitor's own region; an embedded token naming a different
285# region marks the address as out-of-region. An address that carries no token
286# (a bare in-cluster Service name) is region-local by construction.
287_REGION_TOKEN_PATTERN = re.compile(r"\b[a-z]{2}-(?:gov-)?[a-z]+-\d+\b")
289# --- PD proxy behavior -------------------------------------------------------
290#
291# The prefill-decode proxy that fronts a disaggregated endpoint checks whether a
292# prompt's KV blocks already live in the shared store before it sends the prompt
293# to a prefill pod. That check is bounded: it is given this many seconds, and a
294# miss or a check that does not finish in time is treated as "not resident" so
295# the prompt goes to prefill without the request waiting any longer. Holding the
296# bound here keeps the proxy responsive even when the store is slow or
297# unreachable.
298PD_PROXY_RESIDENCY_TIMEOUT_SECONDS = 2
300# Default strategy for spreading requests across the backends of a single role.
301PD_PROXY_DEFAULT_SCHEDULING = "round_robin"
303# Where a residency miss or timed-out lookup is sent. The prompt always goes to
304# a prefill pod in that case; the proxy never stalls the request on the store.
305PD_PROXY_RESIDENCY_MISS_TARGET = "prefill"
307# The residency lookup never blocks the request: a slow or failed store check
308# falls through to prefill rather than holding the client.
309PD_PROXY_RESIDENCY_BLOCKING = "false"
311# Decode-phase requests only ever reach decode pods that report Ready; pods that
312# are still starting are skipped.
313PD_PROXY_DECODE_ROUTING_READY_ONLY = "ready_only"
315# When no decode pod reports Ready, the proxy refuses the request outright
316# instead of streaming a partial generation. The refusal carries a stable
317# status and message so clients can distinguish "no backend yet" from a model
318# error.
319PD_PROXY_NO_DECODE_BACKEND_ACTION_REJECT = "reject"
320PD_PROXY_NO_DECODE_BACKEND_STATUS = "503"
321PD_PROXY_NO_DECODE_BACKEND_MESSAGE = "no available decode backend"
323# Environment variable names the proxy container reads to pick up the behavior
324# above. Surfacing them here keeps the proxy's runtime contract in one place;
325# the reconcile path attaches the values produced by ``build_pd_proxy_config``.
326PD_PROXY_RESIDENCY_TIMEOUT_ENV = "PD_PROXY_RESIDENCY_TIMEOUT_SECONDS"
327PD_PROXY_RESIDENCY_BLOCKING_ENV = "PD_PROXY_RESIDENCY_CHECK_BLOCKING"
328PD_PROXY_RESIDENCY_MISS_TARGET_ENV = "PD_PROXY_RESIDENCY_MISS_TARGET"
329PD_PROXY_DECODE_ROUTING_ENV = "PD_PROXY_DECODE_ROUTING"
330PD_PROXY_NO_DECODE_BACKEND_ACTION_ENV = "PD_PROXY_NO_DECODE_BACKEND_ACTION"
331PD_PROXY_NO_DECODE_BACKEND_STATUS_ENV = "PD_PROXY_NO_DECODE_BACKEND_STATUS"
332PD_PROXY_NO_DECODE_BACKEND_MESSAGE_ENV = "PD_PROXY_NO_DECODE_BACKEND_MESSAGE"
333PD_PROXY_SCHEDULING_ENV = "PD_PROXY_SCHEDULING"
334PD_PROXY_STORE_ADDRESS_ENV = "PD_PROXY_STORE_ADDRESS"
336# Marker label carried by proxy pods so a Service can select the proxy alone,
337# distinct from the prefill/decode role pods (which carry their own role marker).
338PD_PROXY_ROLE_LABEL = "proxy"
340# TCP port the proxy container listens on for authenticated serving requests.
341PD_PROXY_PORT = 8000
343# Environment variable the proxy reads its admin key from, and the data key the
344# backing Kubernetes Secret stores it under. The key value is delivered to the
345# container through a Secret reference at pod start — it is never written to the
346# endpoint spec or passed as a command-line argument.
347PD_PROXY_ADMIN_API_KEY_ENV = "ADMIN_API_KEY"
348ADMIN_API_KEY_SECRET_DATA_KEY = "ADMIN_API_KEY"
350# The proxy program (gco/services/mooncake_pd_proxy.py) is shipped to the proxy
351# pod as a ConfigMap and run from this mount path. The prefill/decode backend
352# URLs and the listen port are passed to it through these env vars; it routes to
353# the role pods through their in-cluster Services so kube-proxy load-balances
354# across only the Ready endpoints of each role.
355PD_PROXY_SCRIPT_FILENAME = "mooncake_pd_proxy.py"
356PD_PROXY_CONFIG_MOUNT_DIR = "/etc/pd-proxy"
357PD_PROXY_SCRIPT_PATH = f"{PD_PROXY_CONFIG_MOUNT_DIR}/{PD_PROXY_SCRIPT_FILENAME}"
358PD_PROXY_PORT_ENV = "PD_PROXY_PORT"
359PD_PROXY_PREFILL_URL_ENV = "PD_PROXY_PREFILL_URL"
360PD_PROXY_DECODE_URL_ENV = "PD_PROXY_DECODE_URL"
363@dataclass
364class RegionServicesResolution:
365 """Outcome of resolving the in-region service addresses an endpoint needs.
367 ``render_mooncake_config`` consumes already-resolved values via a
368 ``region_services`` dict; this carries that dict together with the signals
369 a reconcile pass acts on:
371 - ``region_services`` is the resolved dict to render with, or ``None`` when
372 rendering must be skipped.
373 - ``render_skipped`` is set when the store is enabled but the own-region
374 master address is not configured: the existing endpoint configuration is
375 left untouched and ``store_master_unresolved`` records why.
376 - ``cold_tier_unresolved`` is set when the cold tier was requested but the
377 own-region general-purpose bucket could not be resolved; the cold tier is
378 dropped while the hot-path store keeps operating, and ``error`` explains
379 the condition.
380 """
382 region_services: dict[str, Any] | None = None
383 render_skipped: bool = False
384 store_master_unresolved: bool = False
385 cold_tier_unresolved: bool = False
386 error: str | None = None
389@dataclass
390class MasterReadinessGate:
391 """Outcome of gating dependent role-pod creation on the shared master.
393 A store-bearing endpoint must not materialize its role pods until the
394 single shared ``mooncake-master`` reports a Ready replica. This carries the
395 decision a reconcile pass acts on:
397 - ``proceed`` is ``True`` only when the master reports at least one Ready
398 replica; the caller may then create the dependent role pods and advance
399 out of ``creating``. While it is ``False`` the caller materializes no
400 dependent pods.
401 - ``state`` is the endpoint state to report. It is ``"creating"`` whenever
402 creation is deferred (master not ready, still within the wait window,
403 timed out, or could not be created) and ``None`` when the gate is open.
404 - ``error`` records why creation could not advance: the master did not
405 become Ready within the wait window, or the master could not be created.
406 It is ``None`` while the master is simply still coming up within the
407 window, and ``None`` once the gate is open.
408 """
410 proceed: bool = False
411 state: str | None = None
412 error: str | None = None
415@dataclass
416class RegionalScopeResolution:
417 """Outcome of confirming a disaggregated topology stays inside one region.
419 KV cache transfer over RoCE cannot cross a region boundary, so every
420 ``MooncakeConnector`` peer address and the ``master_server_address`` a
421 topology wires to must resolve to the monitor's own region. An endpoint
422 that targets several regions runs one independent topology per region; each
423 region's monitor reconciles only its own topology and confirms that
424 topology's addresses never escape the region.
426 - ``in_region`` is ``True`` only when every resolved address belongs to the
427 monitor's own region. While it is ``True`` the caller may materialize the
428 topology's role Deployments.
429 - ``peer_addresses`` lists the addresses that were resolved and checked, in
430 a stable order, so callers and logs can show exactly what was wired.
431 - ``state`` is the endpoint state to report. It is ``"failed"`` when an
432 out-of-region address is detected and ``None`` when the topology is
433 wholly in-region.
434 - ``error`` describes the cross-region boundary violation — which addresses
435 resolved to which other regions — when one is found, and is ``None``
436 otherwise. When a violation is reported the caller materializes no role
437 Deployments and leaves any previously materialized resources unchanged.
438 """
440 in_region: bool = True
441 peer_addresses: list[str] = field(default_factory=list)
442 state: str | None = None
443 error: str | None = None
446class RegionStatusConditions(TypedDict, total=False):
447 """Optional DynamoDB predicates shared by every regional status write."""
449 expected_lifecycle_id: str
450 expected_region_generation: str
451 expected_deletion_generation: str
454_LIFECYCLE_ANNOTATION = "gco.io/lifecycle-id"
455_REGION_GENERATION_ANNOTATION = "gco.io/region-generation"
456_LEADER_EPOCH_ANNOTATION = "gco.io/leader-epoch"
459class ReconcileFencedError(RuntimeError):
460 """The caller no longer owns the endpoint or leader epoch it read."""
463@dataclass(frozen=True)
464class ReconcileAuthority:
465 """Immutable authority carried by one endpoint reconciliation pass."""
467 endpoint_name: str
468 lifecycle_id: str
469 region_generation: str
470 leader_epoch: str
471 deletion_generation: str | None = None
472 deleting: bool = False
473 region_removed: bool = False
475 @property
476 def annotations(self) -> dict[str, str]:
477 return {
478 _LIFECYCLE_ANNOTATION: self.lifecycle_id,
479 _REGION_GENERATION_ANNOTATION: self.region_generation,
480 _LEADER_EPOCH_ANNOTATION: self.leader_epoch,
481 }
484@dataclass(frozen=True)
485class EndpointResourceInventory:
486 """Deterministic names of every top-level Kubernetes object owned by an endpoint."""
488 deployments: tuple[str, ...]
489 services: tuple[str, ...]
490 horizontal_pod_autoscalers: tuple[str, ...]
491 scaled_objects: tuple[str, ...]
492 config_maps: tuple[str, ...]
493 legacy_ingresses: tuple[str, ...]
494 legacy_http_routes: tuple[str, ...]
495 generated_admin_secret: str
498@dataclass(frozen=True)
499class ResourceCleanupResult:
500 """Observed result of one idempotent endpoint cleanup pass.
502 A successful Kubernetes delete only starts asynchronous deletion. ``pending``
503 therefore contains objects still returned by a read-after-delete, while
504 ``errors`` contains sanitized request failures. Cleanup is complete only
505 when every owned object is independently observed absent and no request
506 failed during the pass.
507 """
509 pending: tuple[str, ...] = ()
510 errors: tuple[str, ...] = ()
511 resources_found: bool = False
513 @property
514 def complete(self) -> bool:
515 return not self.pending and not self.errors
517 @property
518 def error_message(self) -> str | None:
519 if not self.errors:
520 return None
521 return "Endpoint cleanup could not be completed: " + "; ".join(self.errors)
524def _resolved_mooncake_transfer(mooncake: dict[str, Any]) -> tuple[str, str]:
525 """Resolve the persisted transfer intent and optional network device.
527 GCO's spec deliberately uses ``rdma`` as the portable high-performance
528 intent because the mounted Mooncake store configuration accepts
529 ``rdma|tcp``. On AWS, role pods with that intent are placed on EFA nodes,
530 so :func:`build_kv_transfer_config` translates it to vLLM's explicit
531 ``mooncake_protocol=efa`` at the point-to-point connector boundary.
532 """
533 transfer = mooncake.get("transfer", {})
534 if not isinstance(transfer, dict):
535 raise ValueError("mooncake.transfer must be a mapping")
537 protocol = transfer.get("protocol", "rdma")
538 if protocol not in {"rdma", "tcp"}:
539 raise ValueError(
540 f"mooncake.transfer.protocol must be one of {{rdma, tcp}}, got {protocol!r}"
541 )
542 device_name = transfer.get("device_name", "")
543 if not isinstance(device_name, str):
544 raise ValueError(f"mooncake.transfer.device_name must be a string, got {device_name!r}")
545 return protocol, device_name
548def build_kv_transfer_config(mooncake: dict[str, Any], role: str) -> str:
549 """Return the JSON string for vLLM's ``--kv-transfer-config``.
551 Translates a mooncake spec block plus a worker role into the connector
552 configuration vLLM expects:
554 - ``disaggregated`` emits a ``MooncakeConnector``.
555 - ``store`` emits a ``MooncakeStoreConnector``.
556 - ``both`` emits a ``MultiConnector`` wrapping a ``MooncakeConnector``
557 (index 0) followed by a ``MooncakeStoreConnector`` (index 1), both
558 sharing the role's ``kv_role``.
560 Every point-to-point ``MooncakeConnector`` receives explicit
561 ``kv_connector_extra_config``. GCO's default/high-performance ``rdma``
562 intent maps to Mooncake's AWS-specific ``efa`` protocol because the same
563 pod is pinned to the EFA node pool; ``tcp`` remains an explicit fallback.
564 ``device_name`` is forwarded verbatim, with an empty string requesting
565 Mooncake/libfabric auto-detection.
567 The emitted ``kv_role`` is ``kv_producer`` for prefill, ``kv_consumer``
568 for decode, and ``kv_both`` for a single store instance.
570 Args:
571 mooncake: The ``spec["mooncake"]`` block; its ``mode`` selects the
572 connector shape and its optional ``transfer`` block selects the
573 protocol/device.
574 role: One of ``"prefill"``, ``"decode"``, or ``"single"``.
576 Returns:
577 A JSON object string parseable by vLLM.
579 Raises:
580 ValueError: If the ``(mode, role)`` combination or transfer settings
581 are unsupported. No configuration is emitted in that case.
582 """
583 mode = mooncake.get("mode")
584 supported_roles = _WORKER_ROLES_BY_MODE.get(mode) if isinstance(mode, str) else None
585 if supported_roles is None or role not in supported_roles:
586 raise ValueError(f"Unsupported (mode, role) pair: ({mode!r}, {role!r})")
588 kv_role = _KV_ROLE_BY_WORKER_ROLE[role]
590 if mode == "store":
591 return json.dumps({"kv_connector": "MooncakeStoreConnector", "kv_role": kv_role})
593 protocol, device_name = _resolved_mooncake_transfer(mooncake)
594 connector = {
595 "kv_connector": "MooncakeConnector",
596 "kv_role": kv_role,
597 "kv_connector_extra_config": {
598 "mooncake_protocol": "efa" if protocol == "rdma" else "tcp",
599 "device_name": device_name,
600 },
601 }
602 if mode == "disaggregated":
603 return json.dumps(connector)
605 # mode == "both": MultiConnector chains transfer then store.
606 return json.dumps(
607 {
608 "kv_connector": "MultiConnector",
609 "kv_role": kv_role,
610 "kv_connector_extra_config": {
611 "connectors": [
612 connector,
613 {"kv_connector": "MooncakeStoreConnector", "kv_role": kv_role},
614 ]
615 },
616 }
617 )
620def bootstrap_port_for_worker(base_port: int, dp_rank: int, tp_size: int, tp_rank: int) -> int:
621 """Compute the bootstrap port for a ``(dp_rank, tp_rank)`` worker.
623 The port is ``base_port + dp_rank * tp_size + tp_rank``. For a fixed
624 ``base_port`` and ``tp_size`` distinct ``(dp_rank, tp_rank)`` pairs map to
625 distinct ports.
627 Args:
628 base_port: The base bootstrap port for the endpoint.
629 dp_rank: The data-parallel rank of the worker (``>= 0``).
630 tp_size: The tensor-parallel world size (``>= 1``).
631 tp_rank: The tensor-parallel rank within the worker (``0 <= tp_rank < tp_size``).
633 Returns:
634 The TCP port assigned to the worker.
636 Raises:
637 ValueError: If the computed port falls outside the valid range
638 ``1024..65535``. No port is assigned in that case.
639 """
640 port = base_port + dp_rank * tp_size + tp_rank
641 if port < MIN_BOOTSTRAP_PORT or port > MAX_BOOTSTRAP_PORT:
642 raise ValueError(
643 f"Computed bootstrap port {port} is outside the valid range "
644 f"{MIN_BOOTSTRAP_PORT}..{MAX_BOOTSTRAP_PORT}"
645 )
646 return port
649def render_mooncake_config(
650 mooncake: dict[str, Any], region_services: dict[str, Any]
651) -> dict[str, Any]:
652 """Render the ``mooncake.json`` contents mounted into each vLLM pod.
654 The returned dict is written verbatim to a ConfigMap and mounted at the
655 path named by ``MOONCAKE_CONFIG_PATH``. It always carries the metadata
656 server and the RDMA/TCP transport settings (``protocol`` and
657 ``device_name``); the key-value store and its optional cold tier are layered
658 on only when requested.
660 The transport block (``protocol``/``device_name``) describes the hot
661 RDMA/RoCE path. The cold tier is an asynchronous object-store backend keyed
662 separately as ``cold_tier_s3_uri``; it is never wired into the transport
663 block, so cold-tier reads and writes stay off the RDMA hot path.
665 Resolution of in-region addresses is the caller's responsibility: this
666 function consumes already-resolved values from ``region_services`` and
667 performs no lookups of its own. In particular, the cold-tier URI is the
668 monitor-resolved general-purpose regional bucket for the monitor's own
669 region; any cold-tier bucket URI in the user spec is ignored.
671 Args:
672 mooncake: The ``spec["mooncake"]`` block.
673 region_services: Resolved in-region addresses, e.g.::
675 {
676 "metadata_server": "http://mooncake-master:8080/metadata",
677 "master_server_address": "mooncake-master:50051",
678 "cold_tier_s3_uri": "s3://gco-regional-shared-<acct>-<region>/...",
679 }
681 ``master_server_address`` is required when the store is enabled and
682 ``cold_tier_s3_uri`` is required when the cold tier is enabled.
684 Returns:
685 The ``mooncake.json`` contents as a dict, where:
687 - ``protocol`` and ``device_name`` are always present.
688 - ``master_server_address`` is present only when the store is enabled.
689 - ``cold_tier_s3_uri`` is present only when the cold tier is enabled,
690 which requires ``cold_tier_enabled`` to be the boolean ``True``; any
691 other value (absent, null, truthy non-bool) leaves the cold tier off.
692 """
693 protocol, device_name = _resolved_mooncake_transfer(mooncake)
694 store = mooncake.get("store", {})
695 cfg: dict[str, Any] = {
696 "metadata_server": region_services["metadata_server"],
697 "protocol": protocol,
698 "device_name": device_name,
699 }
700 if store.get("enabled"):
701 cfg["master_server_address"] = region_services["master_server_address"]
702 # The store runs embedded in each vLLM pod (every rank contributes
703 # `global_segment_size` to the shared pool; GCO's per-region
704 # mooncake-master is only the metadata/master coordinator, not a
705 # standalone store that owns the pool). Embedded mode rejects a zero
706 # segment, so default to 4 GiB (the upstream default) when the spec
707 # does not set one; an operator can tune it via configure-store.
708 cfg["global_segment_size"] = store.get("global_segment_size", "4294967296")
709 cfg["local_buffer_size"] = store.get("local_buffer_size", "2147483648")
710 # Only the boolean True enables the cold tier; any other value leaves it
711 # off. The URI is resolved by the caller for the monitor's own region —
712 # never authored by the user — and is an object-store backend kept off
713 # the RDMA transport block above.
714 if store.get("cold_tier_enabled") is True:
715 cfg["cold_tier_s3_uri"] = region_services["cold_tier_s3_uri"]
716 return cfg
719def apply_efa_scheduling(mooncake: dict[str, Any], pod_spec: client.V1PodSpec) -> None:
720 """Place a role pod on the EFA RDMA fabric when transfer runs over RDMA.
722 KV cache transfer over RoCE only runs on EFA-enabled nodes, which carry a
723 ``vpc.amazonaws.com/efa`` taint, advertise the ``vpc.amazonaws.com/efa``
724 extended resource, and are labelled ``efa=true``. When the transfer
725 protocol is ``rdma`` this mutates ``pod_spec`` in place to:
727 - add a ``vpc.amazonaws.com/efa`` toleration (in addition to any existing
728 tolerations such as the GPU one),
729 - add an ``efa=true`` node selector plus a ``mooncake-efa=true`` node
730 selector (merged with any existing selectors), and
731 - request at least one ``vpc.amazonaws.com/efa`` device on the pod's
732 containers, leaving every existing resource request and limit — including
733 GPU asks — untouched.
735 The ``mooncake-efa=true`` selector pins the pod to the dedicated
736 ``mooncake-efa-pool`` NodePool, which only offers instance families with
737 >=80GB of GPU memory and FP8-capable Hopper/Blackwell GPUs. This keeps role
738 pods off the A100-40GB ``p4d`` family that the shared training EFA pool
739 still offers — that family OOMs on many models and cannot run FP8 KV-cache
740 configs, so Karpenter selecting it for a mooncake pod is a latent failure.
742 When the transfer protocol is explicitly set to anything other than
743 ``rdma`` (for example ``tcp``) the pod is left exactly as it was: no
744 toleration, no node selector, and no device request are added. An unset
745 protocol defaults to ``rdma`` — matching the rest of the Mooncake path — so
746 a disaggregated endpoint lands on EFA by default.
748 Tolerations, selectors, and device requests are applied idempotently, so
749 re-running over an already-scheduled pod produces no duplicates.
751 Args:
752 mooncake: The ``spec["mooncake"]`` block; ``transfer.protocol``
753 (defaulting to ``rdma`` when unset) decides whether EFA scheduling
754 applies.
755 pod_spec: The pod specification to mutate in place.
756 """
757 protocol, _device_name = _resolved_mooncake_transfer(mooncake)
758 if protocol != "rdma":
759 return
761 # Tolerate the EFA taint without disturbing existing tolerations.
762 tolerations = list(pod_spec.tolerations or [])
763 if not any(t.key == EFA_RESOURCE_NAME for t in tolerations):
764 tolerations.append(
765 client.V1Toleration(
766 key=EFA_RESOURCE_NAME,
767 operator="Equal",
768 value="true",
769 effect="NoSchedule",
770 )
771 )
772 pod_spec.tolerations = tolerations
774 # Merge the EFA node selectors with any selectors already in place. The
775 # generic efa=true selector lands the pod on EFA fabric; mooncake-efa=true
776 # narrows that to the dedicated mooncake-efa-pool, which excludes the
777 # A100-40GB p4d family that the shared training EFA pool still offers.
778 node_selector = dict(pod_spec.node_selector or {})
779 node_selector[EFA_NODE_SELECTOR_KEY] = EFA_NODE_SELECTOR_VALUE
780 node_selector[MOONCAKE_EFA_NODE_SELECTOR_KEY] = MOONCAKE_EFA_NODE_SELECTOR_VALUE
781 pod_spec.node_selector = node_selector
783 # Request at least one EFA device, preserving existing requests and limits
784 # (notably the GPU asks). Apply to containers that already request an
785 # accelerator; if none do, apply to every container so the pod still asks
786 # for the fabric it needs.
787 containers = pod_spec.containers or []
788 accelerator_keys = ("nvidia.com/gpu", "aws.amazon.com/neuron")
790 def _requests_accelerator(container: client.V1Container) -> bool:
791 reqs = container.resources
792 if reqs is None:
793 return False
794 for table in (reqs.requests, reqs.limits):
795 if table and any(key in table for key in accelerator_keys):
796 return True
797 return False
799 targets = [c for c in containers if _requests_accelerator(c)] or list(containers)
800 for container in targets:
801 if container.resources is None:
802 container.resources = client.V1ResourceRequirements()
803 if container.resources.requests is None:
804 container.resources.requests = {}
805 if container.resources.limits is None:
806 container.resources.limits = {}
807 container.resources.requests.setdefault(EFA_RESOURCE_NAME, "1")
808 container.resources.limits.setdefault(EFA_RESOURCE_NAME, "1")
811def build_pd_proxy_config(mooncake: dict[str, Any]) -> dict[str, str]:
812 """Return the environment the prefill-decode proxy runs with.
814 The proxy fronts a disaggregated endpoint and decides, per request, whether
815 to consult the shared store and which backends to dispatch to. Its behavior
816 is fixed by the values returned here so every disaggregated endpoint front
817 behaves identically:
819 - It looks up whether the prompt's KV blocks already reside in the store
820 before sending the prompt to prefill, and that lookup is bounded to
821 ``PD_PROXY_RESIDENCY_TIMEOUT_SECONDS`` seconds.
822 - A miss, or a lookup that does not finish in time, is treated as "not
823 resident": the prompt is sent to a prefill pod and the request is never
824 held waiting on the store.
825 - Decode-phase requests are routed only to decode pods reporting Ready, so a
826 pod that is still starting is skipped.
827 - When no decode pod reports Ready, the proxy refuses the request with a
828 stable status and message rather than streaming any partial output.
830 The residency bound is held constant rather than read from the spec so the
831 responsiveness guarantee cannot be weakened per endpoint. The store address
832 points at the shared in-region master, and the same-role dispatch strategy
833 falls back to round-robin when the spec names none.
835 Args:
836 mooncake: The ``spec["mooncake"]`` block; its optional ``proxy`` section
837 supplies the same-role scheduling strategy.
839 Returns:
840 A mapping of environment variable name to value, ready to attach to the
841 proxy container.
842 """
843 proxy = mooncake.get("proxy", {}) or {}
844 scheduling = proxy.get("scheduling") or PD_PROXY_DEFAULT_SCHEDULING
845 store_address = f"{MOONCAKE_MASTER_SERVICE}:{MOONCAKE_MASTER_RPC_PORT}"
846 return {
847 PD_PROXY_RESIDENCY_TIMEOUT_ENV: str(PD_PROXY_RESIDENCY_TIMEOUT_SECONDS),
848 PD_PROXY_RESIDENCY_BLOCKING_ENV: PD_PROXY_RESIDENCY_BLOCKING,
849 PD_PROXY_RESIDENCY_MISS_TARGET_ENV: PD_PROXY_RESIDENCY_MISS_TARGET,
850 PD_PROXY_DECODE_ROUTING_ENV: PD_PROXY_DECODE_ROUTING_READY_ONLY,
851 PD_PROXY_NO_DECODE_BACKEND_ACTION_ENV: PD_PROXY_NO_DECODE_BACKEND_ACTION_REJECT,
852 PD_PROXY_NO_DECODE_BACKEND_STATUS_ENV: PD_PROXY_NO_DECODE_BACKEND_STATUS,
853 PD_PROXY_NO_DECODE_BACKEND_MESSAGE_ENV: PD_PROXY_NO_DECODE_BACKEND_MESSAGE,
854 PD_PROXY_SCHEDULING_ENV: scheduling,
855 PD_PROXY_STORE_ADDRESS_ENV: store_address,
856 }
859class InferenceMonitor:
860 """
861 Reconciliation controller for inference endpoints.
863 Polls DynamoDB for desired endpoint state and reconciles with
864 the actual Kubernetes resources in the local cluster.
865 """
867 def __init__(
868 self,
869 cluster_id: str,
870 region: str,
871 store: InferenceEndpointStore,
872 namespace: str = "gco-inference",
873 reconcile_interval: int = 15,
874 ):
875 self.cluster_id = cluster_id
876 self.region = region
877 self.store = store
878 self.namespace = namespace
879 self.reconcile_interval = reconcile_interval
880 self._running = False
882 # Initialize Kubernetes clients
883 try:
884 config.load_incluster_config()
885 logger.info("Loaded in-cluster Kubernetes configuration")
886 except config.ConfigException:
887 try:
888 config.load_kube_config()
889 logger.info("Loaded local Kubernetes configuration")
890 except config.ConfigException as e:
891 logger.error("Failed to load Kubernetes configuration: %s", e)
892 raise
894 self.apps_v1 = client.AppsV1Api()
895 self.core_v1 = client.CoreV1Api()
896 self.networking_v1 = client.NetworkingV1Api()
897 self.discovery_v1 = client.DiscoveryV1Api()
899 # Timeout for Kubernetes API calls (seconds)
900 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
902 # Health watchdog: tracks when each endpoint first became unready.
903 # Inference traffic enters through the shared ``gco-system/gco-gateway``
904 # HTTPRoute at ``/inference`` and then ``gco-system/inference-proxy``, so
905 # model readiness never mutates shared Gateway API resources. Once this
906 # threshold is exceeded, reconciliation emits an explicit degraded-state
907 # warning while the proxy continues returning 503 until a replica is ready.
908 self._unready_since: dict[str, datetime] = {}
909 self._unhealthy_threshold_seconds = int(
910 os.environ.get("INFERENCE_UNHEALTHY_THRESHOLD_SECONDS", "300")
911 ) # 5 minutes default
913 # Master-readiness gate: tracks when each store-bearing endpoint first
914 # deferred its role-pod creation because the shared master was not yet
915 # Ready. The entry is cleared once the master reports a Ready replica so
916 # a later restart of the master restarts the clock cleanly.
917 self._master_deferral_since: dict[str, datetime] = {}
919 # Leader authority is renewed from a dedicated thread while reconcile
920 # performs synchronous Kubernetes calls. The per-acquisition epoch is
921 # also stamped on endpoint-owned objects; object UID/resourceVersion
922 # preconditions remain the hard fence if a process resumes after losing
923 # the Lease.
924 self._lease_name: str | None = None
925 self._lease_holder: str | None = None
926 self._leader_epoch: str | None = None
927 self._leadership_lost = threading.Event()
928 self._active_authority: ReconcileAuthority | None = None
930 # Metrics
931 self._reconcile_count = 0
932 self._errors_count = 0
933 # Monotonic stamp of the last completed loop iteration (leader pass or
934 # standby lease check). Exported as seconds_since_last_pass so a loop
935 # that is wedged — not crashed, which the probes would catch — is
936 # visible; initialised at construction so the gauge exists before the
937 # first iteration and grows if the loop never starts.
938 self._last_loop_completed_at = time.monotonic()
940 # ------------------------------------------------------------------
941 # Reconciliation loop
942 # ------------------------------------------------------------------
944 async def start(self) -> None:
945 """Start reconciliation under an independently renewed leader Lease."""
946 if self._running:
947 logger.warning("Inference monitor already running")
948 return
949 self._running = True
950 logger.info(
951 "Starting inference monitor for %s in %s (interval=%ds)",
952 self.cluster_id,
953 self.region,
954 self.reconcile_interval,
955 )
957 pod_name = os.environ.get("HOSTNAME", f"monitor-{id(self)}")
958 lease_name = "inference-monitor-leader"
959 self._lease_name = lease_name
960 self._lease_holder = pod_name
962 while self._running:
963 try:
964 if self._try_acquire_lease(lease_name, pod_name):
965 with self._renewing_leadership():
966 await self.reconcile()
967 else:
968 logger.debug("Not the leader, waiting...")
969 except ReconcileFencedError as error:
970 logger.warning("Reconciliation stopped after authority loss: %s", error)
971 except Exception as error:
972 logger.error("Reconciliation error: %s", error, exc_info=True)
973 self._errors_count += 1
974 self._last_loop_completed_at = time.monotonic()
975 try:
976 await asyncio.sleep(self.reconcile_interval)
977 except Exception as error:
978 logger.error("Sleep interrupted: %s", error)
979 break
981 @staticmethod
982 def _lease_annotations(lease: Any) -> dict[str, str]:
983 metadata = getattr(lease, "metadata", None)
984 annotations = getattr(metadata, "annotations", None)
985 return dict(annotations) if isinstance(annotations, dict) else {}
987 def _lease_is_expired(self, lease: Any, now: datetime) -> bool:
988 renew_time = getattr(getattr(lease, "spec", None), "renew_time", None)
989 if renew_time is None:
990 return True
991 if renew_time.tzinfo is None:
992 renew_time = renew_time.replace(tzinfo=UTC)
993 duration = getattr(lease.spec, "lease_duration_seconds", None)
994 if not isinstance(duration, int) or isinstance(duration, bool) or duration <= 0:
995 duration = self.reconcile_interval * 3
996 return bool((now - renew_time).total_seconds() >= int(duration))
998 def _stamp_lease_epoch(self, lease: Any, epoch: str) -> None:
999 metadata = getattr(lease, "metadata", None)
1000 if metadata is None:
1001 lease.metadata = client.V1ObjectMeta()
1002 metadata = lease.metadata
1003 annotations = self._lease_annotations(lease)
1004 annotations[_LEADER_EPOCH_ANNOTATION] = epoch
1005 metadata.annotations = annotations
1007 def _try_acquire_lease(self, lease_name: str, holder: str) -> bool:
1008 """Acquire/renew a Lease and establish one stable acquisition epoch."""
1009 coordination_v1 = client.CoordinationV1Api()
1010 now = datetime.now(UTC)
1011 lease_duration = self.reconcile_interval * 3
1012 try:
1013 lease = coordination_v1.read_namespaced_lease(lease_name, self.namespace)
1014 current_holder = lease.spec.holder_identity
1015 expired = self._lease_is_expired(lease, now)
1016 annotations = self._lease_annotations(lease)
1017 observed_epoch = annotations.get(_LEADER_EPOCH_ANNOTATION)
1019 if current_holder == holder and not expired:
1020 # Adopt the persisted epoch after a harmless local restart of
1021 # the loop, but never renew a same-name holder with a different
1022 # in-memory epoch.
1023 if self._leader_epoch is None:
1024 self._leader_epoch = observed_epoch or secrets.token_hex(32)
1025 elif observed_epoch and observed_epoch != self._leader_epoch:
1026 self._leadership_lost.set()
1027 return False
1028 epoch = self._leader_epoch
1029 elif current_holder in (None, "") or expired:
1030 epoch = secrets.token_hex(32)
1031 self._leader_epoch = epoch
1032 lease.spec.holder_identity = holder
1033 lease.spec.acquire_time = now
1034 transitions = getattr(lease.spec, "lease_transitions", None)
1035 lease.spec.lease_transitions = int(transitions or 0) + 1
1036 logger.info("Acquiring leader lease as %s with a new epoch", holder)
1037 else:
1038 return False
1040 lease.spec.renew_time = now
1041 lease.spec.lease_duration_seconds = lease_duration
1042 self._stamp_lease_epoch(lease, epoch)
1043 coordination_v1.replace_namespaced_lease(lease_name, self.namespace, lease)
1044 self._lease_name = lease_name
1045 self._lease_holder = holder
1046 self._leadership_lost.clear()
1047 return True
1048 except ApiException as error:
1049 if error.status == 404:
1050 epoch = secrets.token_hex(32)
1051 lease = client.V1Lease(
1052 metadata=client.V1ObjectMeta(
1053 name=lease_name,
1054 namespace=self.namespace,
1055 annotations={_LEADER_EPOCH_ANNOTATION: epoch},
1056 ),
1057 spec=client.V1LeaseSpec(
1058 holder_identity=holder,
1059 lease_duration_seconds=lease_duration,
1060 acquire_time=now,
1061 renew_time=now,
1062 lease_transitions=0,
1063 ),
1064 )
1065 try:
1066 coordination_v1.create_namespaced_lease(self.namespace, lease)
1067 except ApiException:
1068 return False
1069 self._leader_epoch = epoch
1070 self._lease_name = lease_name
1071 self._lease_holder = holder
1072 self._leadership_lost.clear()
1073 logger.info("Created leader lease as %s", holder)
1074 return True
1075 if error.status == 409:
1076 logger.info("Lost leader Lease optimistic-concurrency race")
1077 self._leadership_lost.set()
1078 return False
1079 logger.warning("Lease check failed: %s", error.reason)
1080 self._leadership_lost.set()
1081 return False
1083 def _renew_current_lease(self) -> bool:
1084 """Renew only the exact holder/epoch currently owned by this process."""
1085 lease_name = self._lease_name
1086 holder = self._lease_holder
1087 epoch = self._leader_epoch
1088 if not all(isinstance(value, str) and value for value in (lease_name, holder, epoch)):
1089 return False
1090 coordination_v1 = client.CoordinationV1Api()
1091 now = datetime.now(UTC)
1092 try:
1093 lease = coordination_v1.read_namespaced_lease(lease_name, self.namespace)
1094 if (
1095 lease.spec.holder_identity != holder
1096 or self._lease_annotations(lease).get(_LEADER_EPOCH_ANNOTATION) != epoch
1097 or self._lease_is_expired(lease, now)
1098 ):
1099 return False
1100 lease.spec.renew_time = now
1101 coordination_v1.replace_namespaced_lease(lease_name, self.namespace, lease)
1102 return True
1103 except Exception:
1104 logger.warning("Leader Lease renewal failed", exc_info=True)
1105 return False
1107 def _lease_renewal_loop(self, stop_event: threading.Event) -> None:
1108 interval = max(1.0, float(self.reconcile_interval))
1109 while not stop_event.wait(interval):
1110 if not self._renew_current_lease():
1111 self._leadership_lost.set()
1112 return
1114 @contextlib.contextmanager
1115 def _renewing_leadership(self) -> Iterator[None]:
1116 """Renew the Lease outside the asyncio loop while one pass is running."""
1117 stop_event = threading.Event()
1118 renewal = threading.Thread(
1119 target=self._lease_renewal_loop,
1120 args=(stop_event,),
1121 name="inference-monitor-lease-renewer",
1122 daemon=True,
1123 )
1124 renewal.start()
1125 try:
1126 yield
1127 finally:
1128 stop_event.set()
1129 renewal.join(timeout=max(1.0, float(self.reconcile_interval)))
1131 def _assert_current_leadership(self) -> None:
1132 """Fail closed before mutation when the persisted Lease epoch changed."""
1133 if self._leadership_lost.is_set():
1134 raise ReconcileFencedError("leader Lease was lost")
1135 # Direct method-level unit tests do not enter start(); production does.
1136 if self._lease_name is None or self._lease_holder is None or self._leader_epoch is None:
1137 return
1138 coordination_v1 = client.CoordinationV1Api()
1139 try:
1140 lease = coordination_v1.read_namespaced_lease(self._lease_name, self.namespace)
1141 except Exception as error:
1142 self._leadership_lost.set()
1143 raise ReconcileFencedError("leader Lease could not be verified") from error
1144 if (
1145 lease.spec.holder_identity != self._lease_holder
1146 or self._lease_annotations(lease).get(_LEADER_EPOCH_ANNOTATION) != self._leader_epoch
1147 or self._lease_is_expired(lease, datetime.now(UTC))
1148 ):
1149 self._leadership_lost.set()
1150 raise ReconcileFencedError("leader Lease holder or epoch changed")
1152 def stop(self) -> None:
1153 """Stop the reconciliation loop and prevent additional mutations."""
1154 self._running = False
1155 self._leadership_lost.set()
1156 logger.info("Inference monitor stopped")
1158 @staticmethod
1159 def _lifecycle_metadata_complete(endpoint: dict[str, Any]) -> bool:
1160 """Return whether an endpoint has complete immutable cleanup metadata."""
1161 lifecycle_id = endpoint.get("lifecycle_id")
1162 cleanup_regions = endpoint.get("cleanup_regions")
1163 region_generations = endpoint.get("region_generations")
1164 return (
1165 isinstance(lifecycle_id, str)
1166 and bool(lifecycle_id)
1167 and isinstance(cleanup_regions, list)
1168 and isinstance(region_generations, dict)
1169 and all(
1170 isinstance(region, str)
1171 and bool(region)
1172 and isinstance(region_generations.get(region), str)
1173 and bool(region_generations[region])
1174 for region in cleanup_regions
1175 )
1176 )
1178 def _status_write_conditions(
1179 self,
1180 endpoint: dict[str, Any],
1181 *,
1182 deleting: bool = False,
1183 ) -> RegionStatusConditions:
1184 """Return lifecycle/generation predicates for one regional status write."""
1185 lifecycle_id = endpoint.get("lifecycle_id")
1186 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1187 return {}
1188 conditions: RegionStatusConditions = {"expected_lifecycle_id": lifecycle_id}
1189 if deleting:
1190 generation = endpoint.get("deletion_generation")
1191 if isinstance(generation, str) and generation:
1192 conditions["expected_deletion_generation"] = generation
1193 return conditions
1194 raw_generations = endpoint.get("region_generations")
1195 generations = raw_generations if isinstance(raw_generations, dict) else {}
1196 region_generation = generations.get(self.region)
1197 if isinstance(region_generation, str) and region_generation:
1198 conditions["expected_region_generation"] = region_generation
1199 return conditions
1201 def _authority_from_endpoint(self, endpoint: dict[str, Any]) -> ReconcileAuthority | None:
1202 """Build object-level provenance for a production endpoint snapshot."""
1203 lifecycle_id = endpoint.get("lifecycle_id")
1204 generations = endpoint.get("region_generations")
1205 region_generation = generations.get(self.region) if isinstance(generations, dict) else None
1206 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1207 if isinstance(endpoint.get("updated_at"), str):
1208 raise ReconcileFencedError("endpoint snapshot has no lifecycle authority")
1209 return None # Compatibility for isolated method-level fixtures only.
1210 if not isinstance(region_generation, str) or not region_generation:
1211 if isinstance(endpoint.get("updated_at"), str):
1212 raise ReconcileFencedError("endpoint snapshot has no Region generation")
1213 return None
1214 desired_state = endpoint.get("desired_state")
1215 targets = endpoint.get("target_regions")
1216 target_regions = targets if isinstance(targets, list) else []
1217 deletion_generation = endpoint.get("deletion_generation")
1218 return ReconcileAuthority(
1219 endpoint_name=str(endpoint.get("endpoint_name", "")),
1220 lifecycle_id=lifecycle_id,
1221 region_generation=region_generation,
1222 leader_epoch=self._leader_epoch or f"direct-{lifecycle_id[:16]}",
1223 deletion_generation=(
1224 deletion_generation
1225 if isinstance(deletion_generation, str) and deletion_generation
1226 else None
1227 ),
1228 deleting=desired_state == "deleted",
1229 region_removed=desired_state != "deleted" and self.region not in target_regions,
1230 )
1232 def _strong_authority_matches(self, authority: ReconcileAuthority) -> bool:
1233 """Re-read DynamoDB before claiming legacy/previous-epoch objects."""
1234 try:
1235 latest = self.store.get_endpoint(authority.endpoint_name, consistent_read=True)
1236 except AttributeError, TypeError:
1237 return getattr(self, "_lease_name", None) is None
1238 if not isinstance(latest, dict):
1239 return getattr(self, "_lease_name", None) is None
1240 if latest.get("lifecycle_id") != authority.lifecycle_id:
1241 return False
1242 if authority.deleting:
1243 return (
1244 latest.get("desired_state") == "deleted"
1245 and latest.get("deletion_generation") == authority.deletion_generation
1246 )
1247 generations = latest.get("region_generations")
1248 return (
1249 latest.get("desired_state") != "deleted"
1250 and isinstance(generations, dict)
1251 and generations.get(self.region) == authority.region_generation
1252 )
1254 @staticmethod
1255 def _object_metadata(resource: Any) -> tuple[Any, dict[str, str], str | None, str | None]:
1256 """Return metadata, annotations, UID, and resourceVersion for typed/dict objects."""
1257 if isinstance(resource, dict):
1258 metadata = resource.get("metadata")
1259 metadata = metadata if isinstance(metadata, dict) else {}
1260 annotations = metadata.get("annotations")
1261 return (
1262 metadata,
1263 dict(annotations) if isinstance(annotations, dict) else {},
1264 metadata.get("uid") if isinstance(metadata.get("uid"), str) else None,
1265 (
1266 metadata.get("resourceVersion")
1267 if isinstance(metadata.get("resourceVersion"), str)
1268 else None
1269 ),
1270 )
1271 metadata = getattr(resource, "metadata", None)
1272 annotations = getattr(metadata, "annotations", None)
1273 uid = getattr(metadata, "uid", None)
1274 resource_version = getattr(metadata, "resource_version", None)
1275 return (
1276 metadata,
1277 dict(annotations) if isinstance(annotations, dict) else {},
1278 uid if isinstance(uid, str) else None,
1279 resource_version if isinstance(resource_version, str) else None,
1280 )
1282 @staticmethod
1283 def _object_labels(resource: Any) -> dict[str, str]:
1284 if isinstance(resource, dict):
1285 metadata = resource.get("metadata")
1286 labels = metadata.get("labels") if isinstance(metadata, dict) else None
1287 else:
1288 metadata = getattr(resource, "metadata", None)
1289 labels = getattr(metadata, "labels", None)
1290 return dict(labels) if isinstance(labels, dict) else {}
1292 @classmethod
1293 def _has_monitor_provenance(cls, resource: Any) -> bool:
1294 labels = cls._object_labels(resource)
1295 return labels.get("project") == "gco" and labels.get("gco.io/type") == "inference"
1297 def _handoff_stale_resource(
1298 self,
1299 resource: Any,
1300 *,
1301 kind: str,
1302 resource_name: str,
1303 delete_resource: Callable[..., Any] | None,
1304 reason: str,
1305 ) -> NoReturn:
1306 """UID-delete a proven stale monitor object only for current DDB authority."""
1307 authority = getattr(self, "_active_authority", None)
1308 if (
1309 authority is None
1310 or delete_resource is None
1311 or not self._has_monitor_provenance(resource)
1312 or not self._strong_authority_matches(authority)
1313 ):
1314 raise ReconcileFencedError(f"{kind} {resource_name} {reason}")
1315 self._assert_current_leadership()
1316 try:
1317 delete_resource(
1318 body=self._delete_options_for(
1319 resource,
1320 kind=kind,
1321 resource_name=resource_name,
1322 )
1323 )
1324 except ApiException as error:
1325 if error.status not in {404, 409}:
1326 raise
1327 raise ReconcileFencedError(
1328 f"{kind} {resource_name} stale authority handoff deletion requested"
1329 )
1331 def _authorize_resource(
1332 self,
1333 resource: Any,
1334 *,
1335 kind: str,
1336 resource_name: str,
1337 patch_metadata: Callable[..., Any] | None = None,
1338 read_resource: Callable[[], Any] | None = None,
1339 delete_resource: Callable[..., Any] | None = None,
1340 allow_region_mismatch: bool = False,
1341 ) -> Any:
1342 """Verify lifecycle provenance and CAS-claim the current leader epoch."""
1343 authority = getattr(self, "_active_authority", None)
1344 if authority is None:
1345 return resource
1346 self._assert_current_leadership()
1347 metadata, annotations, _uid, resource_version = self._object_metadata(resource)
1348 if getattr(self, "_lease_name", None) is None and resource_version is None:
1349 # Historical method-level fixtures use metadata-less MagicMocks.
1350 # Production reconciliation always has a Lease and real metadata.
1351 return resource
1352 observed_lifecycle = annotations.get(_LIFECYCLE_ANNOTATION)
1353 if observed_lifecycle is None and not self._has_monitor_provenance(resource):
1354 raise ReconcileFencedError(f"{kind} {resource_name} has ambiguous legacy ownership")
1355 if observed_lifecycle not in (None, authority.lifecycle_id):
1356 self._handoff_stale_resource(
1357 resource,
1358 kind=kind,
1359 resource_name=resource_name,
1360 delete_resource=delete_resource,
1361 reason="belongs to another endpoint lifecycle",
1362 )
1363 observed_region = annotations.get(_REGION_GENERATION_ANNOTATION)
1364 if not allow_region_mismatch and observed_region not in (None, authority.region_generation):
1365 self._handoff_stale_resource(
1366 resource,
1367 kind=kind,
1368 resource_name=resource_name,
1369 delete_resource=delete_resource,
1370 reason="belongs to another Region generation",
1371 )
1372 expected = authority.annotations
1373 if all(annotations.get(key) == value for key, value in expected.items()):
1374 return resource
1375 if patch_metadata is None or read_resource is None:
1376 raise ReconcileFencedError(f"{kind} {resource_name} lacks current immutable provenance")
1377 if not isinstance(resource_version, str) or not resource_version:
1378 raise ReconcileFencedError(
1379 f"{kind} {resource_name} has no resourceVersion for authority claim"
1380 )
1381 if not self._strong_authority_matches(authority):
1382 raise ReconcileFencedError("endpoint authority changed before Kubernetes mutation")
1383 self._assert_current_leadership()
1384 merged = dict(annotations)
1385 merged.update(expected)
1386 try:
1387 patch_metadata(
1388 body={
1389 "metadata": {
1390 "resourceVersion": resource_version,
1391 "annotations": merged,
1392 }
1393 }
1394 )
1395 except Exception as error:
1396 raise ReconcileFencedError(
1397 f"{kind} {resource_name} changed during authority claim"
1398 ) from error
1399 claimed = read_resource()
1400 _metadata, claimed_annotations, _claimed_uid, _claimed_version = self._object_metadata(
1401 claimed
1402 )
1403 if not all(claimed_annotations.get(key) == value for key, value in expected.items()):
1404 raise ReconcileFencedError(
1405 f"{kind} {resource_name} authority claim could not be verified"
1406 )
1407 return claimed
1409 def _provenance_annotations(self) -> dict[str, str] | None:
1410 authority = getattr(self, "_active_authority", None)
1411 return dict(authority.annotations) if authority is not None else None
1413 def _assert_mutation_authority(self) -> None:
1414 authority = getattr(self, "_active_authority", None)
1415 if authority is None:
1416 return
1417 self._assert_current_leadership()
1418 if not self._strong_authority_matches(authority):
1419 raise ReconcileFencedError("endpoint authority changed before Kubernetes mutation")
1421 def _confirm_created_resource(
1422 self,
1423 *,
1424 kind: str,
1425 resource_name: str,
1426 read_resource: Callable[[], Any],
1427 delete_resource: Callable[..., Any],
1428 ) -> Any:
1429 """Compensate an exact just-created object if Lease/DDB authority changed."""
1430 authority = getattr(self, "_active_authority", None)
1431 if authority is None:
1432 return None
1433 try:
1434 created = read_resource()
1435 except ApiException as error:
1436 if getattr(self, "_lease_name", None) is None and error.status == 404:
1437 return None
1438 raise
1439 _metadata, annotations, _uid, resource_version = self._object_metadata(created)
1440 if getattr(self, "_lease_name", None) is None and resource_version is None:
1441 return created
1442 if not all(annotations.get(key) == value for key, value in authority.annotations.items()):
1443 raise ReconcileFencedError(f"{kind} {resource_name} post-create provenance changed")
1444 try:
1445 self._assert_mutation_authority()
1446 except ReconcileFencedError:
1447 # This process created the exact annotated UID. Remove only that UID;
1448 # a replacement racing into the name makes the precondition fail and
1449 # is never touched.
1450 try:
1451 delete_resource(
1452 body=self._delete_options_for(
1453 created,
1454 kind=kind,
1455 resource_name=resource_name,
1456 )
1457 )
1458 except ApiException as error:
1459 if error.status not in {404, 409}:
1460 logger.warning(
1461 "Post-create compensation failed for %s/%s: status %s",
1462 kind,
1463 resource_name,
1464 error.status,
1465 )
1466 raise
1467 return created
1469 def _delete_options_for(self, resource: Any, *, kind: str, resource_name: str) -> Any:
1470 """Build UID/resourceVersion delete preconditions for one authorized object."""
1471 _metadata, _annotations, uid, resource_version = self._object_metadata(resource)
1472 if uid and resource_version:
1473 return client.V1DeleteOptions(
1474 propagation_policy="Foreground",
1475 preconditions=client.V1Preconditions(
1476 uid=uid,
1477 resource_version=resource_version,
1478 ),
1479 )
1480 if (
1481 getattr(self, "_active_authority", None) is None
1482 or getattr(self, "_lease_name", None) is None
1483 ):
1484 return client.V1DeleteOptions(
1485 propagation_policy="Foreground",
1486 preconditions=client.V1Preconditions(uid=uid) if uid else None,
1487 )
1488 raise ReconcileFencedError(
1489 f"{kind} {resource_name} lacks UID/resourceVersion delete authority"
1490 )
1492 async def reconcile(self) -> list[dict[str, Any]]:
1493 """Run one reconciliation cycle with generation-fenced deletion."""
1494 self._reconcile_count += 1
1495 actions: list[dict[str, Any]] = []
1496 try:
1497 endpoints = self.store.list_endpoints()
1498 except Exception as e:
1499 logger.error("Failed to list endpoints from DynamoDB: %s", e)
1500 return actions
1502 # Records written before lifecycle fencing are upgraded from their
1503 # DynamoDB snapshot before any Kubernetes mutation. The update is
1504 # conditional on ``updated_at``; a concurrent writer wins and this pass
1505 # simply retries from the next scan. Direct method-level test fixtures
1506 # without a persistence timestamp remain outside this production path.
1507 normalized_endpoints: list[dict[str, Any]] = []
1508 for endpoint in endpoints:
1509 if not self._lifecycle_metadata_complete(endpoint) and isinstance(
1510 endpoint.get("updated_at"), str
1511 ):
1512 upgraded = self.store.ensure_lifecycle_metadata(endpoint)
1513 if not isinstance(upgraded, dict):
1514 continue
1515 endpoint = upgraded
1516 actions.append(
1517 {
1518 "action": "initialize_lifecycle",
1519 "endpoint": endpoint.get("endpoint_name", "unknown"),
1520 }
1521 )
1522 normalized_endpoints.append(endpoint)
1523 endpoints = normalized_endpoints
1525 for endpoint in endpoints:
1526 name = endpoint.get("endpoint_name", "unknown")
1527 try:
1528 action = await self._reconcile_endpoint(endpoint)
1529 if action:
1530 actions.append(action)
1531 except ReconcileFencedError as error:
1532 # Authority loss is not endpoint health. Another leader may
1533 # already have written the terminal acknowledgement; a stale
1534 # error write must never regress that quorum to ``error``.
1535 logger.warning("Stopped stale reconciliation for %s: %s", name, error)
1536 continue
1537 except Exception as e:
1538 logger.error("Failed to reconcile endpoint %s: %s", name, e)
1539 self._errors_count += 1
1540 status_conditions = self._status_write_conditions(
1541 endpoint,
1542 deleting=endpoint.get("desired_state") == "deleted",
1543 )
1544 self.store.update_region_status(
1545 name,
1546 self.region,
1547 "error",
1548 error=str(e),
1549 **status_conditions,
1550 )
1552 # Any monitor may purge, but only from a strong snapshot containing a
1553 # fresh terminal acknowledgement for every immutable deletion member.
1554 # Each acknowledgement itself represents two child-complete inventory
1555 # sweeps, and the delete is conditioned on the same lifecycle,
1556 # generation, and updated_at snapshot.
1557 for endpoint in endpoints:
1558 if endpoint.get("desired_state") != "deleted":
1559 continue
1560 ep_name = endpoint.get("endpoint_name")
1561 if not isinstance(ep_name, str):
1562 continue
1563 try:
1564 latest = self.store.get_endpoint(ep_name, consistent_read=True)
1565 except Exception as e:
1566 logger.warning("Failed to refresh deleted endpoint %s: %s", ep_name, e)
1567 continue
1568 if not isinstance(latest, dict) or latest.get("desired_state") != "deleted":
1569 continue
1570 lifecycle_id = latest.get("lifecycle_id")
1571 generation = latest.get("deletion_generation")
1572 deletion_regions = latest.get("deletion_regions")
1573 updated_at = latest.get("updated_at")
1574 if not all(
1575 isinstance(value, str) and value for value in (lifecycle_id, generation, updated_at)
1576 ):
1577 continue
1578 if not isinstance(deletion_regions, list) or not deletion_regions:
1579 continue
1580 cleanup_regions = {
1581 region for region in deletion_regions if isinstance(region, str) and region
1582 }
1583 if len(cleanup_regions) != len(deletion_regions):
1584 continue
1585 raw_status = latest.get("region_status")
1586 region_status = raw_status if isinstance(raw_status, dict) else {}
1587 if not all(
1588 isinstance(region_status.get(region), dict)
1589 and region_status[region].get("state") == "deleted"
1590 and region_status[region].get("lifecycle_id") == lifecycle_id
1591 and region_status[region].get("deletion_generation") == generation
1592 and region_status[region].get("absence_observations", 0) >= 2
1593 for region in cleanup_regions
1594 ):
1595 continue
1596 try:
1597 deleted = self.store.delete_endpoint(
1598 ep_name,
1599 expected_updated_at=updated_at,
1600 expected_lifecycle_id=lifecycle_id,
1601 expected_deletion_generation=generation,
1602 )
1603 if not deleted:
1604 continue
1605 logger.info(
1606 "Purged endpoint %s lifecycle %s generation %s",
1607 ep_name,
1608 lifecycle_id,
1609 generation,
1610 )
1611 actions.append({"action": "purge", "endpoint": ep_name})
1612 except Exception as e:
1613 logger.warning("Failed to purge endpoint %s: %s", ep_name, e)
1614 return actions
1616 @staticmethod
1617 def _cleanup_ack_is_terminal(endpoint: dict[str, Any], region: str) -> bool:
1618 """Return whether this region already durably acknowledged this generation."""
1619 raw_statuses = endpoint.get("region_status")
1620 statuses = raw_statuses if isinstance(raw_statuses, dict) else {}
1621 status = statuses.get(region)
1622 if not isinstance(status, dict) or status.get("state") != "deleted":
1623 return False
1624 lifecycle_id = endpoint.get("lifecycle_id")
1625 if status.get("lifecycle_id") != lifecycle_id:
1626 return False
1627 if endpoint.get("desired_state") == "deleted":
1628 return (
1629 status.get("deletion_generation") == endpoint.get("deletion_generation")
1630 and status.get("absence_observations", 0) >= 2
1631 )
1632 raw_generations = endpoint.get("region_generations")
1633 generations = raw_generations if isinstance(raw_generations, dict) else {}
1634 return (
1635 status.get("region_generation") == generations.get(region)
1636 and status.get("absence_observations", 0) >= 2
1637 )
1639 def _record_cleanup_observation(
1640 self,
1641 endpoint: dict[str, Any],
1642 cleanup: ResourceCleanupResult,
1643 ) -> tuple[str, bool]:
1644 """Persist one stable-absence observation for the active lifecycle/generation."""
1645 lifecycle_id = endpoint.get("lifecycle_id")
1646 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1647 raise RuntimeError("Endpoint cleanup requires an immutable lifecycle id")
1648 deleting = endpoint.get("desired_state") == "deleted"
1649 generation = endpoint.get("deletion_generation") if deleting else None
1650 raw_region_generations = endpoint.get("region_generations")
1651 region_generations = (
1652 raw_region_generations if isinstance(raw_region_generations, dict) else {}
1653 )
1654 region_generation = region_generations.get(self.region) if not deleting else None
1655 if deleting and (not isinstance(generation, str) or not generation):
1656 raise RuntimeError("Endpoint deletion requires an immutable deletion generation")
1657 if not deleting and (not isinstance(region_generation, str) or not region_generation):
1658 raise RuntimeError("Endpoint cleanup requires a current Region generation")
1660 raw_statuses = endpoint.get("region_status")
1661 statuses = raw_statuses if isinstance(raw_statuses, dict) else {}
1662 previous = statuses.get(self.region)
1663 same_generation = (
1664 isinstance(previous, dict)
1665 and previous.get("lifecycle_id") == lifecycle_id
1666 and (
1667 previous.get("deletion_generation") == generation
1668 if deleting
1669 else previous.get("region_generation") == region_generation
1670 )
1671 )
1672 previous_observations = (
1673 int(previous.get("absence_observations", 0))
1674 if same_generation and isinstance(previous, dict)
1675 else 0
1676 )
1677 observations = previous_observations + 1 if cleanup.complete else 0
1678 state = "deleted" if observations >= 2 else "deleting"
1679 status_kwargs: dict[str, Any] = {
1680 "extra": {"absence_observations": observations},
1681 "expected_lifecycle_id": lifecycle_id,
1682 }
1683 if deleting:
1684 status_kwargs["expected_deletion_generation"] = generation
1685 else:
1686 status_kwargs["expected_region_generation"] = region_generation
1687 if cleanup.error_message:
1688 status_kwargs["error"] = cleanup.error_message
1689 written = self.store.update_region_status(
1690 endpoint["endpoint_name"],
1691 self.region,
1692 state,
1693 **status_kwargs,
1694 )
1695 return state, bool(written)
1697 async def _reconcile_endpoint(self, endpoint: dict[str, Any]) -> dict[str, Any] | None:
1698 """Reconcile one endpoint under immutable Kubernetes provenance."""
1699 authority = self._authority_from_endpoint(endpoint)
1700 previous = self._active_authority
1701 self._active_authority = authority
1702 try:
1703 return await self._reconcile_endpoint_authorized(endpoint)
1704 finally:
1705 self._active_authority = previous
1707 async def _reconcile_endpoint_authorized(
1708 self, endpoint: dict[str, Any]
1709 ) -> dict[str, Any] | None:
1710 """Implementation for one endpoint after authority has been installed."""
1711 name = endpoint["endpoint_name"]
1712 desired_state = endpoint.get("desired_state", "deploying")
1713 target_regions = endpoint.get("target_regions", [])
1714 spec = endpoint.get("spec", {})
1715 ns = endpoint.get("namespace", self.namespace)
1717 lifecycle_value = endpoint.get("lifecycle_id")
1718 lifecycle_id = (
1719 lifecycle_value if isinstance(lifecycle_value, str) and lifecycle_value else None
1720 )
1722 if desired_state == "deleted":
1723 generation = endpoint.get("deletion_generation")
1724 deletion_regions = endpoint.get("deletion_regions")
1725 if not isinstance(generation, str) or not isinstance(deletion_regions, list):
1726 # Upgrade a legacy or interrupted delete transition atomically;
1727 # the next pass sees the persisted immutable snapshot.
1728 self.store.update_desired_state(
1729 name,
1730 "deleted",
1731 expected_lifecycle_id=lifecycle_id,
1732 )
1733 return {"action": "initialize_deletion", "endpoint": name}
1734 if self.region not in deletion_regions:
1735 return None
1736 if self._cleanup_ack_is_terminal(endpoint, self.region):
1737 # Durable current-generation acknowledgement makes completed
1738 # non-target work quiescent: no Kubernetes reads and no DDB write.
1739 return None
1740 return self._reconcile_deleted(endpoint, ns, spec if isinstance(spec, dict) else None)
1742 if self.region not in target_regions:
1743 cleanup_regions = endpoint.get("cleanup_regions")
1744 if not isinstance(cleanup_regions, list) or self.region not in cleanup_regions:
1745 return None
1746 if self._cleanup_ack_is_terminal(endpoint, self.region):
1747 return None
1748 cleanup = self._delete_resources(
1749 name,
1750 ns,
1751 spec if isinstance(spec, dict) else None,
1752 expected_lifecycle_id=lifecycle_id,
1753 )
1754 state, written = self._record_cleanup_observation(endpoint, cleanup)
1755 return {
1756 "action": "cleanup",
1757 "endpoint": name,
1758 "reason": "region_removed",
1759 "cleanup_complete": state == "deleted" and written,
1760 }
1762 if desired_state in ("deploying", "running"):
1763 if not isinstance(spec, dict):
1764 error = "endpoint spec must be a mapping"
1765 elif "mooncake" in spec and "canary" in spec:
1766 error = "endpoint spec cannot combine 'mooncake' and 'canary' blocks"
1767 else:
1768 return await self._reconcile_running(name, ns, spec, endpoint)
1769 logger.error("Rejecting invalid endpoint %s: %s", name, error)
1770 self.store.update_region_status(
1771 name,
1772 self.region,
1773 "failed",
1774 error=error,
1775 **self._status_write_conditions(endpoint),
1776 )
1777 return {"action": "reject", "endpoint": name, "reason": "invalid_spec"}
1778 if desired_state == "stopped":
1779 return self._reconcile_stopped(
1780 name,
1781 ns,
1782 spec if isinstance(spec, dict) else None,
1783 endpoint,
1784 )
1785 return None
1787 async def _reconcile_running(
1788 self,
1789 name: str,
1790 namespace: str,
1791 spec: dict[str, Any],
1792 endpoint: dict[str, Any],
1793 ) -> dict[str, Any] | None:
1794 """Ensure the endpoint is running with the correct spec."""
1795 status_conditions = self._status_write_conditions(endpoint)
1796 # Specs carrying a ``mooncake`` block take the disaggregated path. The
1797 # branch returns ``None`` when no such block is present, so a plain
1798 # endpoint falls through to the single-Deployment path below unchanged.
1799 mooncake_action = await self._reconcile_mooncake(name, namespace, spec, endpoint)
1800 if mooncake_action is not None:
1801 return mooncake_action
1803 deployment = self._get_deployment(name, namespace)
1804 configured_replicas = spec.get("replicas", 1)
1805 raw_autoscaling = spec.get("autoscaling", {})
1806 autoscaling = raw_autoscaling if isinstance(raw_autoscaling, dict) else {}
1807 autoscaling_enabled = bool(autoscaling.get("enabled"))
1809 if deployment is None:
1810 # A missing Deployment is still an ownership handoff boundary. Do
1811 # not recreate it until every obsolete HPA/KEDA owner is absent.
1812 if autoscaling_enabled:
1813 ownership = self._reconcile_classic_autoscaler(
1814 name,
1815 namespace,
1816 spec,
1817 apply_desired=False,
1818 )
1819 else:
1820 ownership = self._delete_autoscalers(
1821 (name,),
1822 (name, f"keda-hpa-{name}"),
1823 namespace,
1824 )
1825 if not ownership.complete:
1826 ownership_status_kwargs: dict[str, Any] = {}
1827 if ownership.error_message:
1828 ownership_status_kwargs["error"] = ownership.error_message
1829 self.store.update_region_status(
1830 name,
1831 self.region,
1832 "updating",
1833 replicas_ready=0,
1834 replicas_desired=0,
1835 **status_conditions,
1836 **ownership_status_kwargs,
1837 )
1838 return {
1839 "action": "reconcile_autoscaler",
1840 "endpoint": name,
1841 "cleanup_complete": False,
1842 }
1844 logger.info("Creating endpoint %s in %s", name, self.region)
1845 self._create_deployment(name, namespace, spec)
1846 self._create_service(name, namespace, spec)
1847 if autoscaling_enabled:
1848 self._create_or_update_hpa(name, namespace, spec)
1849 self.store.update_region_status(
1850 name,
1851 self.region,
1852 "creating",
1853 replicas_desired=(
1854 int(autoscaling.get("min_replicas", 1))
1855 if autoscaling_enabled
1856 else configured_replicas
1857 ),
1858 **status_conditions,
1859 )
1860 return {"action": "create", "endpoint": name}
1862 # Deployment exists — ensure its Service exists. Public traffic follows
1863 # ``gco-system/gco-gateway``'s shared ``/inference`` HTTPRoute to
1864 # ``gco-system/inference-proxy``, which then reaches this endpoint's
1865 # ClusterIP Service.
1866 self._ensure_service(name, namespace, spec)
1868 # Once enabled, HPA/KEDA is the sole owner of Deployment
1869 # ``spec.replicas``; the static endpoint count must never fight it.
1870 # Controller creation/handoff runs after capturing live replica status
1871 # below so a pending ownership transition can report useful progress.
1873 observed_desired = getattr(deployment.spec, "replicas", None)
1874 live_desired_replicas = (
1875 int(observed_desired) if observed_desired is not None else configured_replicas
1876 )
1877 current_replicas = live_desired_replicas
1878 ready_replicas = int(getattr(deployment.status, "ready_replicas", 0) or 0)
1879 # During enabled -> disabled handoff the old owner still controls this
1880 # observed value. Report it until static reconciliation actually takes
1881 # ownership, rather than publishing an aspirational configured count.
1882 status_desired_replicas = live_desired_replicas
1883 readiness_floor = (
1884 int(autoscaling.get("min_replicas", 1)) if autoscaling_enabled else configured_replicas
1885 )
1887 self._check_health_watchdog(
1888 name,
1889 namespace,
1890 ready_replicas,
1891 status_desired_replicas,
1892 spec,
1893 endpoint,
1894 )
1896 autoscaler_cleanup: ResourceCleanupResult
1897 if autoscaling_enabled:
1898 # Reconcile on every existing-Deployment pass to repair partial
1899 # creation and configuration drift. Ownership handoffs remain
1900 # updating until the obsolete controller is actually absent.
1901 autoscaler_cleanup = self._reconcile_classic_autoscaler(name, namespace, spec)
1902 else:
1903 # Static ownership is explicit even when the autoscaling block was
1904 # removed entirely: stale HPA/KEDA objects must never regain count.
1905 autoscaler_cleanup = self._delete_autoscalers(
1906 (name,),
1907 (name, f"keda-hpa-{name}"),
1908 namespace,
1909 )
1911 if not autoscaler_cleanup.complete:
1912 status_kwargs: dict[str, Any] = {}
1913 if autoscaler_cleanup.error_message:
1914 status_kwargs["error"] = autoscaler_cleanup.error_message
1915 self.store.update_region_status(
1916 name,
1917 self.region,
1918 "updating",
1919 replicas_ready=ready_replicas,
1920 replicas_desired=status_desired_replicas,
1921 **status_conditions,
1922 **status_kwargs,
1923 )
1924 return {
1925 "action": "reconcile_autoscaler",
1926 "endpoint": name,
1927 "cleanup_complete": False,
1928 }
1930 if not autoscaling_enabled and current_replicas != configured_replicas:
1931 logger.info(
1932 "Scaling endpoint %s: %d → %d replicas",
1933 name,
1934 current_replicas,
1935 configured_replicas,
1936 )
1937 self._scale_deployment(name, namespace, configured_replicas)
1938 self.store.update_region_status(
1939 name,
1940 self.region,
1941 "updating",
1942 replicas_ready=ready_replicas,
1943 replicas_desired=configured_replicas,
1944 **status_conditions,
1945 )
1946 return {"action": "scale", "endpoint": name, "replicas": configured_replicas}
1948 # Check if image changed
1949 current_image = self._get_deployment_image(deployment)
1950 desired_image = self._resolve_image_for_region(spec) if spec.get("image") else ""
1951 if current_image and desired_image and current_image != desired_image:
1952 logger.info("Updating endpoint %s image: %s → %s", name, current_image, desired_image)
1953 self._update_deployment_image(name, namespace, desired_image)
1954 self.store.update_region_status(
1955 name,
1956 self.region,
1957 "updating",
1958 replicas_ready=ready_replicas,
1959 replicas_desired=status_desired_replicas,
1960 **status_conditions,
1961 )
1962 return {"action": "update_image", "endpoint": name, "image": desired_image}
1964 # Reconcile canary first and publish only observed readiness. The
1965 # authenticated proxy will not sample canary traffic until this exact
1966 # region reports the matching image fully Ready.
1967 canary = spec.get("canary")
1968 canary_status = None
1969 if isinstance(canary, dict):
1970 canary_status = self._reconcile_canary(name, namespace, spec, canary, endpoint)
1971 else:
1972 self._cleanup_canary(name, namespace)
1974 # For an autoscaled endpoint, readiness means the configured minimum
1975 # serving capacity is available. During ordinary scale-out the live HPA
1976 # target can temporarily exceed Ready pods without making a healthy
1977 # endpoint flap back to "creating". Keep reporting that live target so
1978 # status and watchdog diagnostics remain truthful.
1979 state = "running" if ready_replicas >= readiness_floor else "creating"
1980 self.store.update_region_status(
1981 name,
1982 self.region,
1983 state,
1984 replicas_ready=ready_replicas,
1985 replicas_desired=status_desired_replicas,
1986 extra={"canary": canary_status} if canary_status is not None else None,
1987 **status_conditions,
1988 )
1990 # Promote desired state only from live local readiness plus explicit
1991 # running observations for every *other* target region. The endpoint
1992 # object may contain a stale local region_status from before this pass.
1993 if state == "running" and endpoint.get("desired_state") == "deploying":
1994 stored_statuses = endpoint.get("region_status", {})
1995 target_regions = endpoint.get("target_regions", [])
1996 all_running = bool(target_regions)
1997 for target_region in target_regions:
1998 if target_region == self.region:
1999 continue
2000 target_status = (
2001 stored_statuses.get(target_region, {})
2002 if isinstance(stored_statuses, dict)
2003 else {}
2004 )
2005 if not isinstance(target_status, dict) or target_status.get("state") != "running":
2006 all_running = False
2007 break
2008 if all_running:
2009 lifecycle_id = endpoint.get("lifecycle_id")
2010 if isinstance(lifecycle_id, str) and lifecycle_id:
2011 self.store.update_desired_state(
2012 name,
2013 "running",
2014 expected_lifecycle_id=lifecycle_id,
2015 expected_desired_state="deploying",
2016 )
2018 return None
2020 # ------------------------------------------------------------------
2021 # Mooncake reconciliation branch
2022 # ------------------------------------------------------------------
2024 @staticmethod
2025 def _desired_roles(mode: str | None) -> list[str]:
2026 """Return the worker roles a mode materializes, in a stable order.
2028 Disaggregated and ``both`` modes split work across ``prefill`` then
2029 ``decode``; store mode runs a single ``kv_both`` instance under the
2030 ``single`` role. The order is fixed so role creation and status
2031 reporting are deterministic across passes.
2032 """
2033 roles = _WORKER_ROLES_BY_MODE.get(mode, set()) if isinstance(mode, str) else set()
2034 return [role for role in ("prefill", "decode", "single") if role in roles]
2036 @staticmethod
2037 def _needs_shared_master(mooncake: dict[str, Any]) -> bool:
2038 """Whether the endpoint depends on the shared per-region master.
2040 The store-bearing modes (``store`` and ``both``) always reach the
2041 master for KV metadata, and any endpoint transferring over RDMA reaches
2042 the master's built-in metadata server for the connector handshake. A
2043 disaggregated endpoint transferring over TCP needs no master.
2044 """
2045 mode = mooncake.get("mode")
2046 if mode in ("store", "both"):
2047 return True
2048 transfer = mooncake.get("transfer") or {}
2049 return bool(transfer.get("protocol", "rdma") == "rdma")
2051 def _ensure_mooncake_configmap(self, name: str, ns: str, cfg: dict[str, Any]) -> None:
2052 """Create or update the shared transport ConfigMap for an endpoint.
2054 The rendered transport settings (the dict produced by
2055 :func:`render_mooncake_config`) are written to a ConfigMap named
2056 ``{name}-mooncake`` under the ``mooncake.json`` key, which each role pod
2057 mounts at the configured path. Creation is idempotent: an existing
2058 ConfigMap is patched to the desired contents so a transport change on
2059 the spec propagates on the next pass.
2060 """
2061 cm_name = f"{name}-mooncake"
2062 config_map = client.V1ConfigMap(
2063 metadata=client.V1ObjectMeta(
2064 name=cm_name,
2065 namespace=ns,
2066 labels={"app": name, "project": "gco", "gco.io/type": "inference"},
2067 annotations=self._provenance_annotations(),
2068 ),
2069 data={"mooncake.json": json.dumps(cfg, sort_keys=True)},
2070 )
2071 self._assert_mutation_authority()
2072 try:
2073 self.core_v1.create_namespaced_config_map(
2074 ns, config_map, _request_timeout=self._k8s_timeout
2075 )
2076 self._confirm_created_resource(
2077 kind="configmap",
2078 resource_name=cm_name,
2079 read_resource=partial(
2080 self.core_v1.read_namespaced_config_map,
2081 cm_name,
2082 ns,
2083 _request_timeout=self._k8s_timeout,
2084 ),
2085 delete_resource=partial(
2086 self.core_v1.delete_namespaced_config_map,
2087 cm_name,
2088 ns,
2089 _request_timeout=self._k8s_timeout,
2090 ),
2091 )
2092 logger.info("Created mooncake config map %s/%s", ns, cm_name)
2093 except ApiException as error:
2094 if error.status != 409:
2095 raise
2096 existing = self.core_v1.read_namespaced_config_map(
2097 cm_name, ns, _request_timeout=self._k8s_timeout
2098 )
2099 existing = self._authorize_resource(
2100 existing,
2101 kind="configmap",
2102 resource_name=cm_name,
2103 patch_metadata=partial(
2104 self.core_v1.patch_namespaced_config_map,
2105 cm_name,
2106 ns,
2107 _request_timeout=self._k8s_timeout,
2108 ),
2109 read_resource=lambda: self.core_v1.read_namespaced_config_map(
2110 cm_name, ns, _request_timeout=self._k8s_timeout
2111 ),
2112 delete_resource=partial(
2113 self.core_v1.delete_namespaced_config_map,
2114 cm_name,
2115 ns,
2116 _request_timeout=self._k8s_timeout,
2117 ),
2118 )
2119 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
2120 config_map.metadata.resource_version = resource_version
2121 self._assert_mutation_authority()
2122 self.core_v1.patch_namespaced_config_map(
2123 cm_name, ns, config_map, _request_timeout=self._k8s_timeout
2124 )
2125 logger.info("Updated mooncake config map %s/%s", ns, cm_name)
2127 def _ensure_role_deployment(
2128 self, name: str, ns: str, spec: dict[str, Any], role: str
2129 ) -> tuple[int, int, bool]:
2130 """Ensure one role target has its static or restart-seed capacity.
2132 Existing autoscaled targets normally retain controller ownership, but a
2133 manually stopped zero target is seeded to the role minimum before its
2134 autoscaler is reapplied. Static roles always converge to topology.
2135 """
2136 mooncake = spec.get("mooncake") or {}
2137 deploy_name = name if role == "single" else f"{name}-{role}"
2138 desired = self._replica_count_for_role(mooncake, role)
2140 deployment = self._get_deployment(deploy_name, ns)
2141 if deployment is None:
2142 self._create_role_deployment(name, ns, spec, role)
2143 return 0, desired, False
2145 autoscaling = mooncake.get("autoscaling") or {}
2146 role_autoscaling = autoscaling.get(role)
2147 autoscaled = (
2148 bool(autoscaling.get("enabled"))
2149 and role in ("prefill", "decode")
2150 and isinstance(role_autoscaling, dict)
2151 )
2152 current = deployment.spec.replicas or 0
2153 restarted = autoscaled and current == 0
2154 if (not autoscaled and current != desired) or restarted:
2155 logger.info(
2156 "Scaling role deployment %s/%s: %d → %d",
2157 ns,
2158 deploy_name,
2159 current,
2160 desired,
2161 )
2162 self._scale_deployment(deploy_name, ns, desired)
2164 status = getattr(deployment, "status", None)
2165 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2166 return ready, desired, restarted
2168 def _report_role_status(
2169 self,
2170 name: str,
2171 ns: str,
2172 mooncake: dict[str, Any],
2173 region_services: dict[str, Any],
2174 endpoint: dict[str, Any] | None = None,
2175 ) -> str:
2176 """Write the role-keyed region status for a Mooncake endpoint.
2178 For split topologies the status carries a ``roles`` map of observed and
2179 desired replica counts per role; for store-bearing endpoints it carries
2180 a ``store`` sub-status with the master's readiness and address. The flat
2181 ``replicas_ready`` / ``replicas_desired`` fields are also populated with
2182 the totals so consumers that only read the flat shape still see motion.
2184 Returns:
2185 The reported endpoint state: ``"running"`` once every desired role
2186 replica (and, when applicable, the master) is Ready, otherwise
2187 ``"creating"``.
2188 """
2189 mode = mooncake.get("mode")
2190 roles = self._desired_roles(mode)
2191 extra: dict[str, Any] = {}
2192 total_ready = 0
2193 total_desired = 0
2194 all_ready = True
2196 if mode in ("disaggregated", "both"):
2197 roles_block: dict[str, Any] = {}
2198 for role in ("prefill", "decode"):
2199 if role not in roles:
2200 continue
2201 deploy_name = f"{name}-{role}"
2202 configured_floor = self._replica_count_for_role(mooncake, role)
2203 dep = self._get_deployment(deploy_name, ns)
2204 status = getattr(dep, "status", None) if dep else None
2205 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2206 deployment_spec = getattr(dep, "spec", None) if dep else None
2207 observed_desired = getattr(deployment_spec, "replicas", None)
2208 desired = (
2209 int(observed_desired) if observed_desired is not None else configured_floor
2210 )
2211 roles_block[role] = {"ready": ready, "desired": desired}
2212 total_ready += ready
2213 total_desired += desired
2214 if ready < configured_floor:
2215 all_ready = False
2216 extra["roles"] = roles_block
2217 else:
2218 # Store mode runs a single kv_both Deployment under the endpoint name.
2219 configured_floor = self._replica_count_for_role(mooncake, "single")
2220 dep = self._get_deployment(name, ns)
2221 status = getattr(dep, "status", None) if dep else None
2222 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2223 deployment_spec = getattr(dep, "spec", None) if dep else None
2224 observed_desired = getattr(deployment_spec, "replicas", None)
2225 desired = int(observed_desired) if observed_desired is not None else configured_floor
2226 total_ready += ready
2227 total_desired += desired
2228 if ready < configured_floor:
2229 all_ready = False
2231 store = mooncake.get("store") or {}
2232 if store.get("enabled"):
2233 master_ready = self._mooncake_master_ready_replicas(ns) >= 1
2234 extra["store"] = {
2235 "ready": master_ready,
2236 "master": region_services.get("master_server_address"),
2237 }
2238 if not master_ready:
2239 all_ready = False
2241 state = "running" if all_ready and total_desired > 0 else "creating"
2242 status_conditions = self._status_write_conditions(endpoint or {})
2243 self.store.update_region_status(
2244 name,
2245 self.region,
2246 state,
2247 replicas_ready=total_ready,
2248 replicas_desired=total_desired,
2249 extra=extra or None,
2250 **status_conditions,
2251 )
2252 return state
2254 async def _reconcile_mooncake(
2255 self,
2256 name: str,
2257 ns: str,
2258 spec: dict[str, Any],
2259 endpoint: dict[str, Any],
2260 ) -> dict[str, Any] | None:
2261 """Reconcile an endpoint whose spec carries a ``mooncake`` block.
2263 Returns ``None`` when the spec carries no ``mooncake`` block, signalling
2264 the caller to take the single-Deployment path: one Deployment at the
2265 configured replica count and one internal ClusterIP Service, with no role
2266 split, proxy, autoscaler, shared-master dependency, endpoint Ingress,
2267 Gateway, or HTTPRoute. The shared platform route remains unchanged.
2269 With a ``mooncake`` block present, the topology is materialized in
2270 dependency order, and the shared ConfigMap and master are laid down
2271 before any role pod, the roles before the front-end, and the front-end
2272 before status is written:
2274 1. Resolve the in-region addresses (master, metadata, optional cold
2275 tier). When the store is enabled but no own-region master is
2276 configured, nothing further is materialized; the existing
2277 configuration is left unchanged and the endpoint is reported as
2278 still coming up with the unresolved-master reason.
2279 2. Confirm every wired address stays inside the monitor's own region. A
2280 cross-region address fails the endpoint and materializes nothing,
2281 leaving any prior resources in place.
2282 3. Gate dependent pods on the shared per-region master, which also lays
2283 down the intra-namespace allow rules. While the master is not Ready,
2284 or if it could not be created, nothing further is materialized and
2285 the endpoint is reported as still coming up.
2286 4. Render and apply the shared transport ConfigMap.
2287 5. Materialize each role Deployment: prefill and decode for
2288 disaggregated and both modes, a single ``kv_both`` Deployment for
2289 store mode.
2290 6. Materialize each present role's autoscaler when autoscaling is on.
2291 7. Front disaggregated and both modes with the proxy and its internal
2292 ClusterIP Service; give store mode an internal ClusterIP Service.
2293 Public traffic remains on the shared ``gco-system/gco-gateway``
2294 HTTPRoute from ``/inference`` to ``gco-system/inference-proxy``.
2295 8. Write the role-keyed region status.
2297 Returns:
2298 An action record describing what the pass did, or ``None`` when the
2299 spec carries no ``mooncake`` block.
2300 """
2301 mooncake = spec.get("mooncake")
2302 if not mooncake:
2303 return None
2305 status_conditions = self._status_write_conditions(endpoint)
2306 mode = mooncake.get("mode")
2308 # Step 1: resolve in-region addresses. A store without an own-region
2309 # master is left untouched and reported as still coming up.
2310 services = self._resolve_region_services(name, mooncake)
2311 if services.render_skipped:
2312 self.store.update_region_status(
2313 name,
2314 self.region,
2315 "creating",
2316 error=services.error,
2317 **status_conditions,
2318 )
2319 return {
2320 "action": "reconcile_mooncake",
2321 "endpoint": name,
2322 "deferred": "store_master_unresolved",
2323 }
2325 region_services = services.region_services or {}
2327 # Step 2: keep the topology inside its own region.
2328 scope = self._resolve_regional_scope(name, ns, spec, region_services)
2329 if not scope.in_region:
2330 self.store.update_region_status(
2331 name,
2332 self.region,
2333 scope.state or "failed",
2334 error=scope.error,
2335 **status_conditions,
2336 )
2337 return {
2338 "action": "reconcile_mooncake",
2339 "endpoint": name,
2340 "failed": "cross_region_boundary",
2341 }
2343 # Step 3: gate dependent pods on the shared master (and its allow
2344 # rules). This is also where the master itself is created if absent.
2345 if self._needs_shared_master(mooncake):
2346 gate = self._gate_on_mooncake_master(name, ns, spec)
2347 if not gate.proceed:
2348 self.store.update_region_status(
2349 name,
2350 self.region,
2351 gate.state or "creating",
2352 error=gate.error,
2353 **status_conditions,
2354 )
2355 return {
2356 "action": "reconcile_mooncake",
2357 "endpoint": name,
2358 "deferred": "master_not_ready",
2359 }
2361 # Step 4: shared transport ConfigMap, applied once before role pods.
2362 cfg = render_mooncake_config(mooncake, region_services)
2363 self._ensure_mooncake_configmap(name, ns, cfg)
2365 # Steps 5-6: converge ownership before creating/scaling any role
2366 # Deployment. This is the same recreate/handoff barrier used by classic
2367 # endpoints, extended to prefill/decode and stale topology roles.
2368 desired_roles = self._desired_roles(mode)
2369 ownership_results = [
2370 self._delete_autoscalers(
2371 (name,),
2372 (name, f"keda-hpa-{name}"),
2373 ns,
2374 )
2375 ]
2376 for role in ("prefill", "decode"):
2377 role_name = f"{name}-{role}"
2378 if role in desired_roles:
2379 ownership_results.append(
2380 self._reconcile_role_autoscaler(
2381 name,
2382 ns,
2383 spec,
2384 role,
2385 apply_desired=False,
2386 )
2387 )
2388 else:
2389 ownership_results.append(
2390 self._delete_autoscalers(
2391 (role_name,),
2392 (role_name, f"keda-hpa-{role_name}"),
2393 ns,
2394 )
2395 )
2396 ownership = self._merge_cleanup_results(*ownership_results)
2397 if not ownership.complete:
2398 ready_total = 0
2399 desired_total = 0
2400 for role in desired_roles:
2401 deploy_name = name if role == "single" else f"{name}-{role}"
2402 deployment = self._get_deployment(deploy_name, ns)
2403 deployment_status = getattr(deployment, "status", None)
2404 deployment_spec = getattr(deployment, "spec", None)
2405 ready_total += int(getattr(deployment_status, "ready_replicas", 0) or 0)
2406 desired_total += int(getattr(deployment_spec, "replicas", 0) or 0)
2407 status_kwargs: dict[str, Any] = {}
2408 if ownership.error_message:
2409 status_kwargs["error"] = ownership.error_message
2410 self.store.update_region_status(
2411 name,
2412 self.region,
2413 "updating",
2414 replicas_ready=ready_total,
2415 replicas_desired=desired_total,
2416 **status_conditions,
2417 **status_kwargs,
2418 )
2419 return {
2420 "action": "reconcile_mooncake_autoscaler",
2421 "endpoint": name,
2422 "cleanup_complete": False,
2423 }
2425 role_ready_total = 0
2426 role_desired_total = 0
2427 for role in desired_roles:
2428 ready, desired, _restarted = self._ensure_role_deployment(name, ns, spec, role)
2429 role_ready_total += ready
2430 role_desired_total += desired
2432 owner_verifications: list[ResourceCleanupResult] = []
2433 for role in ("prefill", "decode"):
2434 role_config = self._role_autoscaling_config(spec, role)
2435 if role not in desired_roles or role_config is None:
2436 continue
2437 self._create_role_hpa(name, ns, spec, role)
2438 target_name = f"{name}-{role}"
2439 metrics_config = role_config.get("metrics", [{"type": "cpu", "target": 70}])
2440 hpa_name = (
2441 f"keda-hpa-{target_name}"
2442 if self._metrics_require_keda(metrics_config)
2443 else target_name
2444 )
2445 owner_verifications.append(self._verify_hpa_owner(hpa_name, ns, target_name))
2446 verified_owners = self._merge_cleanup_results(*owner_verifications)
2447 if not verified_owners.complete:
2448 verification_status_kwargs: dict[str, Any] = {}
2449 if verified_owners.error_message:
2450 verification_status_kwargs["error"] = verified_owners.error_message
2451 self.store.update_region_status(
2452 name,
2453 self.region,
2454 "updating",
2455 replicas_ready=role_ready_total,
2456 replicas_desired=role_desired_total,
2457 **status_conditions,
2458 **verification_status_kwargs,
2459 )
2460 return {
2461 "action": "reconcile_mooncake_autoscaler",
2462 "endpoint": name,
2463 "cleanup_complete": False,
2464 }
2466 # Step 7: front-end. Disaggregated and both run behind the proxy; store
2467 # exposes its single Deployment directly.
2468 if mode in ("disaggregated", "both"):
2469 # Per-role Services so the proxy can address prefill and decode by
2470 # stable in-cluster DNS. Routing through a Service means kube-proxy
2471 # load-balances across only the Ready pods of each role, which is
2472 # what gives the proxy ready-only decode routing for free.
2473 role_port = spec.get("port", 8000)
2474 for role in desired_roles:
2475 self._create_role_service(name, ns, role, role_port)
2476 try:
2477 self._create_pd_proxy(name, ns, spec, endpoint)
2478 except AdminApiKeySecretError as e:
2479 logger.error("Proxy for endpoint %s in %s not started: %s", name, ns, e)
2480 self.store.update_region_status(
2481 name,
2482 self.region,
2483 "failed",
2484 error=str(e),
2485 **status_conditions,
2486 )
2487 return {
2488 "action": "reconcile_mooncake",
2489 "endpoint": name,
2490 "failed": "admin_api_key",
2491 }
2492 else:
2493 self._create_service(name, ns, spec)
2495 # Step 8: role-keyed status.
2496 state = self._report_role_status(
2497 name,
2498 ns,
2499 mooncake,
2500 region_services,
2501 endpoint,
2502 )
2503 return {"action": "reconcile_mooncake", "endpoint": name, "state": state}
2505 def _reconcile_stopped(
2506 self,
2507 name: str,
2508 namespace: str,
2509 spec: dict[str, Any] | None = None,
2510 endpoint: dict[str, Any] | None = None,
2511 ) -> dict[str, Any] | None:
2512 """Remove every possible autoscaler owner before scaling all roles to zero."""
2513 mooncake_endpoint = isinstance(spec, dict) and isinstance(spec.get("mooncake"), dict)
2514 self._unready_since.pop(name, None)
2515 self._master_deferral_since.pop(name, None)
2516 status_conditions = self._status_write_conditions(endpoint or {})
2517 role_names = (name, f"{name}-prefill", f"{name}-decode")
2518 cleanup = self._delete_autoscalers(
2519 role_names,
2520 (
2521 name,
2522 f"{name}-prefill",
2523 f"{name}-decode",
2524 f"keda-hpa-{name}",
2525 f"keda-hpa-{name}-prefill",
2526 f"keda-hpa-{name}-decode",
2527 ),
2528 namespace,
2529 )
2530 deployment_names = (*role_names, f"{name}-proxy") if mooncake_endpoint else (name,)
2531 deployments = {
2532 deployment_name: self._get_deployment(deployment_name, namespace)
2533 for deployment_name in deployment_names
2534 }
2535 ready_replicas = sum(
2536 int(getattr(getattr(deployment, "status", None), "ready_replicas", 0) or 0)
2537 for deployment in deployments.values()
2538 if deployment is not None
2539 )
2540 desired_replicas = sum(
2541 int(getattr(getattr(deployment, "spec", None), "replicas", 0) or 0)
2542 for deployment in deployments.values()
2543 if deployment is not None
2544 )
2545 if not cleanup.complete:
2546 status_kwargs: dict[str, Any] = {}
2547 if cleanup.error_message:
2548 status_kwargs["error"] = cleanup.error_message
2549 self.store.update_region_status(
2550 name,
2551 self.region,
2552 "stopping",
2553 replicas_ready=ready_replicas,
2554 replicas_desired=desired_replicas,
2555 **status_conditions,
2556 **status_kwargs,
2557 )
2558 return {"action": "stop", "endpoint": name, "cleanup_complete": False}
2560 scaled = False
2561 for deployment_name, deployment in deployments.items():
2562 if deployment is None:
2563 continue
2564 current_replicas = int(getattr(getattr(deployment, "spec", None), "replicas", 0) or 0)
2565 if current_replicas <= 0:
2566 continue
2567 logger.info("Stopping endpoint role %s (scaling to 0)", deployment_name)
2568 self._scale_deployment(deployment_name, namespace, 0)
2569 scaled = True
2571 self.store.update_region_status(
2572 name,
2573 self.region,
2574 "stopped",
2575 replicas_ready=0,
2576 replicas_desired=0,
2577 **status_conditions,
2578 )
2579 if scaled:
2580 return {"action": "stop", "endpoint": name, "cleanup_complete": True}
2581 return None
2583 def _reconcile_deleted(
2584 self,
2585 endpoint: dict[str, Any],
2586 namespace: str,
2587 spec: dict[str, Any] | None = None,
2588 ) -> dict[str, Any]:
2589 """Converge all parents and generated children to stable absence."""
2590 name = endpoint["endpoint_name"]
2591 lifecycle_id = endpoint.get("lifecycle_id")
2592 if not isinstance(lifecycle_id, str) or not lifecycle_id:
2593 raise RuntimeError("Endpoint deletion requires an immutable lifecycle id")
2594 self._unready_since.pop(name, None)
2595 self._master_deferral_since.pop(name, None)
2596 logger.info("Reconciling deletion of endpoint %s from %s", name, self.region)
2597 cleanup = self._delete_resources(
2598 name,
2599 namespace,
2600 spec,
2601 expected_lifecycle_id=lifecycle_id,
2602 )
2603 state, written = self._record_cleanup_observation(endpoint, cleanup)
2604 return {
2605 "action": "delete",
2606 "endpoint": name,
2607 "cleanup_complete": state == "deleted" and written,
2608 }
2610 # ------------------------------------------------------------------
2611 # Kubernetes resource management
2612 # ------------------------------------------------------------------
2614 def _deployment_exists(self, name: str, namespace: str) -> bool:
2615 return self._get_deployment(name, namespace) is not None
2617 def _get_deployment(self, name: str, namespace: str) -> V1Deployment | None:
2618 try:
2619 deployment = self.apps_v1.read_namespaced_deployment(
2620 name, namespace, _request_timeout=self._k8s_timeout
2621 )
2622 except ApiException as error:
2623 if error.status == 404:
2624 return None
2625 raise
2626 return self._authorize_resource(
2627 deployment,
2628 kind="deployment",
2629 resource_name=name,
2630 patch_metadata=partial(
2631 self.apps_v1.patch_namespaced_deployment,
2632 name,
2633 namespace,
2634 _request_timeout=self._k8s_timeout,
2635 ),
2636 read_resource=lambda: self.apps_v1.read_namespaced_deployment(
2637 name, namespace, _request_timeout=self._k8s_timeout
2638 ),
2639 delete_resource=partial(
2640 self.apps_v1.delete_namespaced_deployment,
2641 name,
2642 namespace,
2643 _request_timeout=self._k8s_timeout,
2644 ),
2645 )
2647 def _get_deployment_image(self, deployment: V1Deployment) -> str | None:
2648 """Get the image of the first container in a deployment."""
2649 containers = deployment.spec.template.spec.containers
2650 if containers:
2651 image: str = containers[0].image
2652 return image
2653 return None
2655 def _resolve_image_for_region(self, spec: dict[str, Any]) -> str:
2656 """Pick the image URI this region should pull from.
2658 ``cli.inference.InferenceManager.deploy`` populates
2659 ``spec["region_image_uris"]`` with a per-region map when the
2660 primary image is an ECR URI, so each cluster can pull from its
2661 local replica instead of crossing the WAN. The map is omitted
2662 for non-ECR refs and for deploys with ``rewrite_image=False``,
2663 in which case we fall back to the flat ``spec["image"]`` URI.
2665 When the map is present but lacks an entry for ``self.region``
2666 (a target region was added after the spec was last written),
2667 the flat URI is also used so the deployment doesn't break — the
2668 next reconcile after a fresh deploy picks up the right URI.
2669 """
2670 region_map = spec.get("region_image_uris")
2671 if isinstance(region_map, dict):
2672 uri = region_map.get(self.region)
2673 if isinstance(uri, str) and uri:
2674 return uri
2675 return str(spec["image"])
2677 # ------------------------------------------------------------------
2678 # In-region service resolution (master address + cold-tier bucket)
2679 # ------------------------------------------------------------------
2681 def _resolve_region_services(
2682 self, name: str, mooncake: dict[str, Any]
2683 ) -> RegionServicesResolution:
2684 """Resolve the in-region addresses an endpoint's ``mooncake.json`` needs.
2686 Everything an endpoint wires to is resolved for the monitor's own
2687 region from regional configuration — never from values typed into the
2688 endpoint spec:
2690 - The shared master's RPC address comes from regional configuration. It
2691 is required whenever the store is enabled. When the store is enabled
2692 and no own-region master address is configured, rendering is skipped
2693 and the existing endpoint configuration is left unchanged; the result
2694 records the unresolved-master condition so the caller can report it.
2695 - The metadata server defaults to the master host on the metadata port
2696 unless regional configuration supplies one explicitly.
2697 - When the cold tier is opted in (``store.cold_tier_enabled`` is the
2698 boolean ``True``), the cold-tier object-store URI is resolved to the
2699 own-region general-purpose regional bucket from that region's
2700 ``/name`` discovery value. Any cold-tier bucket URI in the spec is
2701 ignored. Whether the endpoint writes to the cold tier is governed
2702 solely by the per-endpoint flag, independent of the always-on bucket.
2703 When the bucket cannot be resolved (the region's stack is not yet
2704 deployed), the cold tier is dropped and the condition is recorded,
2705 while the hot-path store stays configured.
2707 Args:
2708 name: The endpoint name, used to scope the cold-tier object key.
2709 mooncake: The ``spec["mooncake"]`` block.
2711 Returns:
2712 A :class:`RegionServicesResolution` carrying the resolved
2713 ``region_services`` dict and any skip/unresolved signals.
2714 """
2715 store = mooncake.get("store", {})
2716 store_enabled = bool(store.get("enabled"))
2718 master_address = os.environ.get(MOONCAKE_MASTER_ADDRESS_ENV, "").strip()
2720 # The store needs an own-region master. It is a fixed in-cluster Service
2721 # the monitor itself provisions per region (mooncake-master:50051), so
2722 # when no override is set in the environment, default to that Service
2723 # rather than deferring — the address is known by construction. An
2724 # operator may still override it via MOONCAKE_MASTER_ADDRESS.
2725 if store_enabled and not master_address:
2726 master_address = f"{MOONCAKE_MASTER_SERVICE}:{MOONCAKE_MASTER_RPC_PORT}"
2728 region_services: dict[str, Any] = {
2729 "metadata_server": self._metadata_server_url(master_address),
2730 }
2731 if store_enabled:
2732 region_services["master_server_address"] = master_address
2734 result = RegionServicesResolution(region_services=region_services)
2736 # Cold tier is opt-in per endpoint. The bucket is always resolved for
2737 # the monitor's own region; any URI supplied in the spec is ignored.
2738 if store_enabled and store.get("cold_tier_enabled") is True:
2739 bucket = self._resolve_regional_shared_bucket()
2740 if bucket:
2741 region_services["cold_tier_s3_uri"] = (
2742 f"s3://{bucket}/{MOONCAKE_COLD_TIER_KEY_PREFIX}/{name}/"
2743 )
2744 else:
2745 # The own-region general-purpose bucket is not resolvable yet.
2746 # Drop the cold tier but keep the hot-path store operating.
2747 result.cold_tier_unresolved = True
2748 result.error = (
2749 "general-purpose regional bucket for region "
2750 f"{self.region} could not be resolved; cold tier disabled, "
2751 "hot-path store still active"
2752 )
2754 return result
2756 def _metadata_server_url(self, master_address: str) -> str:
2757 """Return the metadata server URL, deriving it from the master host.
2759 Regional configuration may supply the metadata server URL directly.
2760 When it does not, the URL defaults to the master host on the metadata
2761 port; if no master host is known the conventional in-cluster service
2762 name is used.
2763 """
2764 configured = os.environ.get(MOONCAKE_METADATA_SERVER_ENV, "").strip()
2765 if configured:
2766 return configured
2767 host = master_address.rsplit(":", 1)[0] if master_address else MOONCAKE_MASTER_SERVICE
2768 return f"http://{host}:{MOONCAKE_METADATA_PORT}/metadata"
2770 def _resolve_regional_shared_bucket(self) -> str | None:
2771 """Resolve the own-region general-purpose bucket name, or ``None``.
2773 Reads the monitor's own region's ``/name`` discovery value for the
2774 always-on general-purpose regional bucket. Returns ``None`` when the
2775 value is absent (the region's stack is not yet deployed) or cannot be
2776 read, so the caller can drop the cold tier without disturbing the
2777 hot-path store.
2778 """
2779 from gco.services.aws_ssm import get_ssm_parameter_optional
2781 param_name = f"{_regional_shared_ssm_parameter_prefix()}/name"
2782 try:
2783 bucket = get_ssm_parameter_optional(param_name, region=self.region)
2784 return bucket if isinstance(bucket, str) and bucket else None
2785 except Exception as e: # noqa: BLE001 - any read failure means "unresolved"
2786 logger.warning(
2787 "Failed to resolve general-purpose regional bucket for %s: %s",
2788 self.region,
2789 e,
2790 )
2791 return None
2793 # ------------------------------------------------------------------
2794 # Regional scope boundary (intra-region RDMA enforcement)
2795 # ------------------------------------------------------------------
2797 def _region_of_address(self, address: str) -> str:
2798 """Return the address's explicit AWS region or safe local classification.
2800 AWS region tokens embedded in the host are authoritative. Bare Service
2801 names and Kubernetes ``.svc`` names are local by construction. Any
2802 other host without a region token is external and ambiguous, so it is
2803 classified as ``"unknown"`` and rejected by regional-scope checks.
2804 """
2805 candidate = (address or "").strip()
2806 if not candidate:
2807 return "unknown"
2809 try:
2810 parsed = urlsplit(candidate if "://" in candidate else f"//{candidate}")
2811 host = (parsed.hostname or "").rstrip(".").lower()
2812 except ValueError:
2813 return "unknown"
2814 if not host:
2815 return "unknown"
2817 match = _REGION_TOKEN_PATTERN.search(host)
2818 if match:
2819 return match.group(0)
2820 if "." not in host or host.endswith((".svc", ".svc.cluster.local")):
2821 return self.region
2822 return "unknown"
2824 def _resolve_regional_scope(
2825 self,
2826 name: str,
2827 ns: str,
2828 spec: dict[str, Any],
2829 region_services: dict[str, Any] | None,
2830 ) -> RegionalScopeResolution:
2831 """Confirm a disaggregated topology wires only to its own region.
2833 Gathers every address the own-region topology connects to — the
2834 ``MooncakeConnector`` peers (the sibling role Services for prefill and
2835 decode), the shared master's RPC address, and the metadata server — and
2836 confirms each resolves to the monitor's own region. Any explicitly
2837 supplied peer addresses in the spec are checked too, so a misconfigured
2838 endpoint that points a peer or master at another region is caught
2839 before any role pod is materialized.
2841 An endpoint that enumerates two or more target regions runs one
2842 independent topology per region: each region's monitor reconciles only
2843 its own topology (it reconciles only while ``self.region`` is one of the
2844 target regions), and this resolution confirms that topology's addresses
2845 never cross into another region.
2847 Args:
2848 name: The endpoint name; used to derive the in-cluster peer Service
2849 names for the disaggregated roles.
2850 ns: The namespace the topology materializes into.
2851 spec: The endpoint spec being reconciled.
2852 region_services: The resolved in-region addresses (from
2853 :meth:`_resolve_region_services`), or ``None`` when none were
2854 resolved. Supplies the master and metadata addresses to check.
2856 Returns:
2857 A :class:`RegionalScopeResolution`. When every resolved address is
2858 own-region, ``in_region`` is ``True`` and the caller may materialize
2859 the role Deployments. When any address resolves to another region,
2860 ``in_region`` is ``False``, ``state`` is ``"failed"``, and ``error``
2861 names the offending addresses; the caller then materializes no role
2862 Deployments and leaves any prior resources unchanged.
2863 """
2864 mooncake = spec.get("mooncake") or {}
2865 mode = mooncake.get("mode")
2866 roles = _WORKER_ROLES_BY_MODE.get(mode, set()) if isinstance(mode, str) else set()
2868 # MooncakeConnector peers are the sibling role Services within this
2869 # namespace; collect them in a stable order for deterministic reporting.
2870 addresses: list[str] = []
2871 if "prefill" in roles or "decode" in roles:
2872 for role in ("prefill", "decode"):
2873 addresses.append(f"{name}-{role}.{ns}.svc.cluster.local")
2875 # The shared master and metadata server the pods reach. These come from
2876 # the own-region resolution, but are checked here so a foreign address
2877 # supplied through regional configuration is still caught.
2878 if region_services:
2879 master = region_services.get("master_server_address")
2880 metadata = region_services.get("metadata_server")
2881 if master:
2882 addresses.append(str(master))
2883 if metadata:
2884 addresses.append(str(metadata))
2886 # Defensive: honor any explicit peer/master addresses authored on the
2887 # spec so a hand-edited endpoint cannot smuggle in an out-of-region peer.
2888 store = mooncake.get("store") or {}
2889 transfer = mooncake.get("transfer") or {}
2890 for candidate in (
2891 store.get("master_server_address"),
2892 store.get("metadata_server"),
2893 ):
2894 if candidate:
2895 addresses.append(str(candidate))
2896 explicit_peers = transfer.get("peer_addresses")
2897 if isinstance(explicit_peers, list):
2898 addresses.extend(str(peer) for peer in explicit_peers if peer)
2900 # De-duplicate while preserving first-seen order.
2901 seen: set[str] = set()
2902 ordered: list[str] = []
2903 for address in addresses:
2904 if address not in seen:
2905 seen.add(address)
2906 ordered.append(address)
2908 resolved_regions = [(address, self._region_of_address(address)) for address in ordered]
2909 out_of_region = [
2910 (address, region) for address, region in resolved_regions if region != self.region
2911 ]
2913 if out_of_region:
2914 detail = ", ".join(
2915 f"{address!r} resolves to region {region}" for address, region in out_of_region
2916 )
2917 logger.error(
2918 "Cross-region boundary violation for endpoint %s in %s: %s",
2919 name,
2920 self.region,
2921 detail,
2922 )
2923 return RegionalScopeResolution(
2924 in_region=False,
2925 peer_addresses=ordered,
2926 state="failed",
2927 error=(f"cross-region boundary violation: {detail}; expected region {self.region}"),
2928 )
2930 return RegionalScopeResolution(in_region=True, peer_addresses=ordered)
2932 def _ensure_mooncake_store(self, ns: str, spec: dict[str, Any]) -> None:
2933 """Maintain the single shared per-region Mooncake master, idempotently.
2935 The master is region-shared, not per-endpoint: every endpoint that
2936 needs the store reaches the same ``mooncake-master`` StatefulSet and the
2937 headless Service that fronts its RPC and metadata ports. This method
2938 uses create-if-absent semantics, so any number of calls within a region
2939 converge on exactly one StatefulSet with a single replica and one
2940 Service. An already-existing master is left untouched — a conflicting
2941 create is treated as success and never overwrites the running master.
2943 The StatefulSet runs the master daemon with its built-in HTTP metadata
2944 server, exposing RPC on :data:`MOONCAKE_MASTER_RPC_PORT` and the
2945 metadata endpoint on :data:`MOONCAKE_METADATA_PORT`. Both ports are
2946 published on the headless Service so in-namespace pods can resolve them.
2948 Args:
2949 ns: The namespace the master shares with the inference workloads.
2950 spec: The endpoint spec being reconciled. Its ``mooncake`` block may
2951 carry a master image override; otherwise the image is taken from
2952 the in-region deployment's environment.
2953 """
2954 mooncake = spec.get("mooncake", {}) or {}
2955 store = mooncake.get("store", {}) or {}
2956 image = store.get("master_image") or os.environ.get(MOONCAKE_MASTER_IMAGE_ENV, "").strip()
2958 labels = {"app": MOONCAKE_MASTER_SERVICE, "project": "gco"}
2960 # Headless Service exposing both the RPC and metadata ports. A None
2961 # cluster IP keeps it headless so the StatefulSet's stable network
2962 # identity resolves directly.
2963 service = client.V1Service(
2964 metadata=client.V1ObjectMeta(
2965 name=MOONCAKE_MASTER_SERVICE,
2966 namespace=ns,
2967 labels=labels,
2968 ),
2969 spec=client.V1ServiceSpec(
2970 cluster_ip="None",
2971 selector={"app": MOONCAKE_MASTER_SERVICE},
2972 ports=[
2973 client.V1ServicePort(
2974 name="rpc",
2975 port=MOONCAKE_MASTER_RPC_PORT,
2976 target_port="rpc",
2977 protocol="TCP",
2978 ),
2979 client.V1ServicePort(
2980 name="metadata",
2981 port=MOONCAKE_METADATA_PORT,
2982 target_port="metadata",
2983 protocol="TCP",
2984 ),
2985 ],
2986 ),
2987 )
2989 container = client.V1Container(
2990 name=MOONCAKE_MASTER_SERVICE,
2991 image=image,
2992 command=["mooncake_master"],
2993 args=[
2994 f"--port={MOONCAKE_MASTER_RPC_PORT}",
2995 "--enable_http_metadata_server=true",
2996 f"--http_metadata_server_port={MOONCAKE_METADATA_PORT}",
2997 ],
2998 ports=[
2999 client.V1ContainerPort(
3000 name="rpc",
3001 container_port=MOONCAKE_MASTER_RPC_PORT,
3002 protocol="TCP",
3003 ),
3004 client.V1ContainerPort(
3005 name="metadata",
3006 container_port=MOONCAKE_METADATA_PORT,
3007 protocol="TCP",
3008 ),
3009 ],
3010 security_context=client.V1SecurityContext(
3011 allow_privilege_escalation=False,
3012 # The upstream mooncake_master launcher chmods its bundled
3013 # binary on startup, which needs a writable root filesystem (a
3014 # read-only root raised OSError: Read-only file system). The pod
3015 # also runs as root so the chmod of the root-owned binary is
3016 # permitted. Privilege escalation stays disabled and all
3017 # capabilities are dropped, so this is constrained root.
3018 read_only_root_filesystem=False,
3019 capabilities=client.V1Capabilities(drop=["ALL"]),
3020 ),
3021 resources=client.V1ResourceRequirements(
3022 requests={"cpu": "250m", "memory": "512Mi"},
3023 limits={"cpu": "1", "memory": "2Gi"},
3024 ),
3025 startup_probe=client.V1Probe(
3026 tcp_socket=client.V1TCPSocketAction(port="rpc"),
3027 initial_delay_seconds=5,
3028 period_seconds=5,
3029 failure_threshold=30,
3030 ),
3031 liveness_probe=client.V1Probe(
3032 tcp_socket=client.V1TCPSocketAction(port="rpc"),
3033 initial_delay_seconds=15,
3034 period_seconds=30,
3035 ),
3036 readiness_probe=client.V1Probe(
3037 # The HTTP metadata server returns 400 for a bare GET /metadata
3038 # (it expects a ?key=), so an HTTP GET readiness probe never
3039 # passes. Confirm readiness by checking the metadata port is
3040 # accepting connections instead.
3041 tcp_socket=client.V1TCPSocketAction(port="metadata"),
3042 initial_delay_seconds=10,
3043 period_seconds=15,
3044 ),
3045 )
3047 stateful_set = client.V1StatefulSet(
3048 metadata=client.V1ObjectMeta(
3049 name=MOONCAKE_MASTER_SERVICE,
3050 namespace=ns,
3051 labels=labels,
3052 ),
3053 spec=client.V1StatefulSetSpec(
3054 service_name=MOONCAKE_MASTER_SERVICE,
3055 replicas=1,
3056 selector=client.V1LabelSelector(match_labels={"app": MOONCAKE_MASTER_SERVICE}),
3057 template=client.V1PodTemplateSpec(
3058 metadata=client.V1ObjectMeta(
3059 labels=labels,
3060 # Keep Karpenter from consolidating the node out from
3061 # under the master: it is a single-replica, stateful
3062 # control-plane daemon holding KV metadata for every
3063 # in-region endpoint, so an eviction drops that state and
3064 # disrupts inference. On lightly-loaded clusters
3065 # consolidation otherwise evicts it mid-image-pull before
3066 # it can even start.
3067 annotations={"karpenter.sh/do-not-disrupt": "true"},
3068 ),
3069 spec=client.V1PodSpec(
3070 service_account_name="gco-service-account",
3071 automount_service_account_token=False,
3072 # The upstream mooncake_master launcher chmods its
3073 # bundled binary (root-owned in the image) on startup, so
3074 # the master must run as root: a non-root uid cannot
3075 # chmod a root-owned file ("Operation not permitted").
3076 # gco-inference does not enforce restricted Pod Security
3077 # and the no-root rule applies only to user-submitted
3078 # jobs, so a root platform daemon is consistent here. The
3079 # container still drops all Linux capabilities and
3080 # disallows privilege escalation (see its securityContext).
3081 security_context=client.V1PodSecurityContext(
3082 run_as_user=0,
3083 run_as_group=0,
3084 ),
3085 containers=[container],
3086 restart_policy="Always",
3087 ),
3088 ),
3089 ),
3090 )
3092 # Create-if-absent: a 409 means the shared master already exists, which
3093 # is the steady state. Leave it untouched and treat it as success.
3094 self._assert_mutation_authority()
3095 try:
3096 self.core_v1.create_namespaced_service(ns, service, _request_timeout=self._k8s_timeout)
3097 logger.info("Created shared mooncake master service in %s", ns)
3098 except ApiException as e:
3099 if e.status == 409:
3100 logger.info("Shared mooncake master service already exists in %s", ns)
3101 else:
3102 raise
3104 self._assert_mutation_authority()
3105 try:
3106 self.apps_v1.create_namespaced_stateful_set(
3107 ns, stateful_set, _request_timeout=self._k8s_timeout
3108 )
3109 logger.info("Created shared mooncake master statefulset in %s", ns)
3110 except ApiException as e:
3111 if e.status == 409:
3112 logger.info("Shared mooncake master statefulset already exists in %s", ns)
3113 else:
3114 raise
3116 def _mooncake_master_ready_replicas(self, ns: str) -> int:
3117 """Return the shared master's Ready replica count, 0 when absent.
3119 Reads the ``mooncake-master`` StatefulSet status in ``ns``. A missing
3120 StatefulSet (404) reports zero Ready replicas rather than raising, so a
3121 caller gating on readiness simply keeps deferring until it appears.
3123 Args:
3124 ns: The namespace the shared master lives in.
3126 Returns:
3127 The number of Ready replicas the StatefulSet reports, or 0 when it
3128 does not yet exist or reports no Ready replicas.
3129 """
3130 try:
3131 status = self.apps_v1.read_namespaced_stateful_set_status(
3132 MOONCAKE_MASTER_SERVICE, ns, _request_timeout=self._k8s_timeout
3133 )
3134 except ApiException as e:
3135 if e.status == 404:
3136 return 0
3137 raise
3139 ready = getattr(getattr(status, "status", None), "ready_replicas", 0)
3140 return int(ready or 0)
3142 def _gate_on_mooncake_master(
3143 self, name: str, ns: str, spec: dict[str, Any]
3144 ) -> MasterReadinessGate:
3145 """Gate dependent role-pod creation on the shared master's readiness.
3147 Maintains the single shared master (create-if-absent) and then decides
3148 whether the endpoint's dependent role pods may be materialized:
3150 - If maintaining the master fails, no dependent pods are materialized,
3151 any existing master is left unmodified, and the endpoint stays in
3152 ``creating`` carrying a create-failure error.
3153 - While the master reports fewer than 1 Ready replica, creation is
3154 deferred and the endpoint stays in ``creating``. The first deferral
3155 starts a clock; once it exceeds the wait window the endpoint keeps
3156 deferring and stays in ``creating`` but also surfaces a not-ready
3157 error. The master is never deleted or modified on account of the
3158 timeout.
3159 - Once the master reports at least 1 Ready replica, the gate opens: the
3160 clock is cleared and the caller may create the role pods and advance
3161 out of ``creating``.
3163 Args:
3164 name: The endpoint name, used to track its first deferral.
3165 ns: The namespace the master shares with the workloads.
3166 spec: The endpoint spec being reconciled.
3168 Returns:
3169 A :class:`MasterReadinessGate` describing whether to proceed, the
3170 endpoint state to report, and any error to surface.
3171 """
3172 # Maintain the shared master first. A create failure must not produce
3173 # any dependent pods and must leave an existing master untouched.
3174 try:
3175 self._ensure_mooncake_store(ns, spec)
3176 except ApiException as e:
3177 logger.error(
3178 "Could not create shared mooncake master in %s for endpoint %s: %s",
3179 ns,
3180 name,
3181 e,
3182 )
3183 return MasterReadinessGate(
3184 proceed=False,
3185 state="creating",
3186 error="shared master could not be created",
3187 )
3189 # Apply the intra-namespace allow rules before any role pod is created.
3190 # A failure here must fail pod materialization while leaving the
3191 # default-deny posture intact; surface which rule could not be applied.
3192 try:
3193 self._ensure_intra_namespace_network_policies(ns, spec)
3194 except NetworkPolicyApplyError as e:
3195 logger.error(
3196 "Could not apply network policy %s in %s for endpoint %s: %s",
3197 e.rule,
3198 ns,
3199 name,
3200 e.reason,
3201 )
3202 return MasterReadinessGate(
3203 proceed=False,
3204 state="creating",
3205 error=f"network policy {e.rule} could not be applied",
3206 )
3208 ready_replicas = self._mooncake_master_ready_replicas(ns)
3209 if ready_replicas >= 1:
3210 # Master is Ready: open the gate and reset the deferral clock so a
3211 # later master restart restarts the wait window cleanly.
3212 if name in self._master_deferral_since:
3213 logger.info(
3214 "Shared master Ready in %s, resuming creation for endpoint %s",
3215 ns,
3216 name,
3217 )
3218 del self._master_deferral_since[name]
3219 return MasterReadinessGate(proceed=True, state=None, error=None)
3221 # Master not Ready: defer creation and report creating. Start the clock
3222 # on the first deferral.
3223 now = datetime.now(UTC)
3224 first_deferral = self._master_deferral_since.setdefault(name, now)
3225 deferred_for = (now - first_deferral).total_seconds()
3227 if deferred_for >= MOONCAKE_MASTER_READY_TIMEOUT_SECONDS:
3228 logger.error(
3229 "Shared master not Ready in %s after %.0fs, still deferring endpoint %s",
3230 ns,
3231 deferred_for,
3232 name,
3233 )
3234 return MasterReadinessGate(
3235 proceed=False,
3236 state="creating",
3237 error="shared master did not become Ready",
3238 )
3240 logger.info(
3241 "Deferring creation for endpoint %s in %s until shared master is Ready",
3242 name,
3243 ns,
3244 )
3245 return MasterReadinessGate(proceed=False, state="creating", error=None)
3247 def _ensure_intra_namespace_network_policies(self, ns: str, spec: dict[str, Any]) -> None:
3248 """Apply the intra-namespace allow rules disaggregated inference needs.
3250 Alongside the default-deny posture in ``gco-inference`` (defined in
3251 ``03-network-policies.yaml`` and never touched here), this maintains
3252 four widening allow rules with create-if-absent semantics:
3254 - ``allow-inference-internal`` — managed inference pods exchange TCP
3255 traffic and may reach the shared master's two fixed ports. This
3256 permits proxy-to-role serving and bootstrap traffic while excluding
3257 unselected sources such as the ALB.
3258 - ``allow-pod-to-master`` — inference pods reach the shared master RPC
3259 port (:data:`MOONCAKE_MASTER_RPC_PORT`).
3260 - ``allow-pod-to-metadata`` — inference pods reach the shared metadata
3261 server (:data:`MOONCAKE_METADATA_PORT`).
3262 - ``allow-rdma-bootstrap`` — inference pods reach each other on the
3263 contiguous KV-transfer bootstrap port window starting at the spec's
3264 ``mooncake.transfer.bootstrap_base_port`` (default
3265 :data:`MOONCAKE_BOOTSTRAP_BASE_PORT`).
3267 Each rule is created independently; an already-present rule (409) is the
3268 steady state and counts as success. No deny rule is ever read, modified,
3269 or deleted, so the default-deny policy is preserved regardless of
3270 outcome.
3272 Args:
3273 ns: The inference namespace the rules apply to.
3274 spec: The endpoint spec being reconciled; its
3275 ``mooncake.transfer.bootstrap_base_port`` sizes the bootstrap
3276 port window.
3278 Raises:
3279 NetworkPolicyApplyError: If any single rule cannot be created. The
3280 error names the failing rule; rules created before the failure
3281 remain in place and the default-deny policy is untouched.
3282 """
3283 transfer = (spec.get("mooncake", {}) or {}).get("transfer", {}) or {}
3284 base_port = transfer.get("bootstrap_base_port", MOONCAKE_BOOTSTRAP_BASE_PORT)
3285 try:
3286 base_port = int(base_port)
3287 except TypeError, ValueError:
3288 base_port = MOONCAKE_BOOTSTRAP_BASE_PORT
3289 end_port = min(base_port + MOONCAKE_BOOTSTRAP_PORT_SPAN, MAX_BOOTSTRAP_PORT)
3291 labels = {"project": "gco"}
3292 master_selector = client.V1LabelSelector(match_labels={"app": MOONCAKE_MASTER_SERVICE})
3293 inference_selector = client.V1LabelSelector(match_labels=INFERENCE_POD_SELECTOR)
3294 inference_peer = [client.V1NetworkPolicyPeer(pod_selector=inference_selector)]
3295 master_peer = [client.V1NetworkPolicyPeer(pod_selector=master_selector)]
3296 all_tcp = [client.V1NetworkPolicyPort(protocol="TCP")]
3298 policies = [
3299 (
3300 NETWORK_POLICY_INFERENCE_INTERNAL,
3301 client.V1NetworkPolicy(
3302 metadata=client.V1ObjectMeta(
3303 name=NETWORK_POLICY_INFERENCE_INTERNAL, namespace=ns, labels=labels
3304 ),
3305 spec=client.V1NetworkPolicySpec(
3306 pod_selector=inference_selector,
3307 policy_types=["Ingress", "Egress"],
3308 ingress=[
3309 client.V1NetworkPolicyIngressRule(
3310 _from=inference_peer,
3311 ports=all_tcp,
3312 )
3313 ],
3314 egress=[
3315 client.V1NetworkPolicyEgressRule(
3316 to=inference_peer,
3317 ports=all_tcp,
3318 ),
3319 client.V1NetworkPolicyEgressRule(
3320 to=master_peer,
3321 ports=[
3322 client.V1NetworkPolicyPort(
3323 protocol="TCP", port=MOONCAKE_MASTER_RPC_PORT
3324 ),
3325 client.V1NetworkPolicyPort(
3326 protocol="TCP", port=MOONCAKE_METADATA_PORT
3327 ),
3328 ],
3329 ),
3330 ],
3331 ),
3332 ),
3333 ),
3334 (
3335 NETWORK_POLICY_POD_TO_MASTER,
3336 client.V1NetworkPolicy(
3337 metadata=client.V1ObjectMeta(
3338 name=NETWORK_POLICY_POD_TO_MASTER, namespace=ns, labels=labels
3339 ),
3340 spec=client.V1NetworkPolicySpec(
3341 pod_selector=master_selector,
3342 policy_types=["Ingress"],
3343 ingress=[
3344 client.V1NetworkPolicyIngressRule(
3345 _from=inference_peer,
3346 ports=[
3347 client.V1NetworkPolicyPort(
3348 protocol="TCP", port=MOONCAKE_MASTER_RPC_PORT
3349 )
3350 ],
3351 )
3352 ],
3353 ),
3354 ),
3355 ),
3356 (
3357 NETWORK_POLICY_POD_TO_METADATA,
3358 client.V1NetworkPolicy(
3359 metadata=client.V1ObjectMeta(
3360 name=NETWORK_POLICY_POD_TO_METADATA, namespace=ns, labels=labels
3361 ),
3362 spec=client.V1NetworkPolicySpec(
3363 pod_selector=master_selector,
3364 policy_types=["Ingress"],
3365 ingress=[
3366 client.V1NetworkPolicyIngressRule(
3367 _from=inference_peer,
3368 ports=[
3369 client.V1NetworkPolicyPort(
3370 protocol="TCP", port=MOONCAKE_METADATA_PORT
3371 )
3372 ],
3373 )
3374 ],
3375 ),
3376 ),
3377 ),
3378 (
3379 NETWORK_POLICY_RDMA_BOOTSTRAP,
3380 client.V1NetworkPolicy(
3381 metadata=client.V1ObjectMeta(
3382 name=NETWORK_POLICY_RDMA_BOOTSTRAP, namespace=ns, labels=labels
3383 ),
3384 spec=client.V1NetworkPolicySpec(
3385 pod_selector=inference_selector,
3386 policy_types=["Ingress"],
3387 ingress=[
3388 client.V1NetworkPolicyIngressRule(
3389 _from=inference_peer,
3390 ports=[
3391 client.V1NetworkPolicyPort(
3392 protocol="TCP",
3393 port=base_port,
3394 end_port=end_port,
3395 )
3396 ],
3397 )
3398 ],
3399 ),
3400 ),
3401 ),
3402 ]
3404 for rule_name, policy in policies:
3405 self._assert_mutation_authority()
3406 try:
3407 self.networking_v1.create_namespaced_network_policy(
3408 ns, policy, _request_timeout=self._k8s_timeout
3409 )
3410 logger.info("Applied network policy %s in %s", rule_name, ns)
3411 except ApiException as e:
3412 if e.status == 409:
3413 # Already present — the steady state. Leave it untouched.
3414 logger.info("Network policy %s already present in %s", rule_name, ns)
3415 continue
3416 logger.error(
3417 "Could not apply network policy %s in %s: %s",
3418 rule_name,
3419 ns,
3420 e,
3421 )
3422 raise NetworkPolicyApplyError(rule_name, e.reason or str(e)) from e
3424 def _create_deployment(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
3425 """Create a Kubernetes Deployment for an inference endpoint."""
3426 replicas = spec.get("replicas", 1)
3427 deployment = self._build_inference_deployment_object(
3428 name=name,
3429 deploy_name=name,
3430 app_label=name,
3431 namespace=namespace,
3432 spec=spec,
3433 replicas=replicas,
3434 )
3435 self._assert_mutation_authority()
3436 self.apps_v1.create_namespaced_deployment(
3437 namespace, deployment, _request_timeout=self._k8s_timeout
3438 )
3439 self._confirm_created_resource(
3440 kind="deployment",
3441 resource_name=name,
3442 read_resource=partial(
3443 self.apps_v1.read_namespaced_deployment,
3444 name,
3445 namespace,
3446 _request_timeout=self._k8s_timeout,
3447 ),
3448 delete_resource=partial(
3449 self.apps_v1.delete_namespaced_deployment,
3450 name,
3451 namespace,
3452 _request_timeout=self._k8s_timeout,
3453 ),
3454 )
3455 logger.info("Created deployment %s/%s", namespace, name)
3457 def _build_inference_deployment_object(
3458 self,
3459 name: str,
3460 deploy_name: str,
3461 app_label: str,
3462 namespace: str,
3463 spec: dict[str, Any],
3464 replicas: int,
3465 extra_args: list[str] | None = None,
3466 extra_labels: dict[str, str] | None = None,
3467 ) -> client.V1Deployment:
3468 """Build the ``V1Deployment`` object for an inference workload.
3470 Shared by the single-Deployment path and the role-split prefill/decode/
3471 store paths. ``name`` is the endpoint name (used for the in-cluster
3472 serving prefix and model cache directory), ``deploy_name`` is the
3473 Kubernetes object name, and ``app_label`` is the selector label that
3474 Services and autoscalers target. ``extra_args`` are appended to the
3475 container args (for example the rendered ``--kv-transfer-config``), and
3476 ``extra_labels`` are merged into both the Deployment and pod-template
3477 labels so role pods carry a stable role marker.
3478 """
3479 image = self._resolve_image_for_region(spec)
3480 port = spec.get("port", 8000)
3481 gpu_count = spec.get("gpu_count", 1)
3482 health_path = spec.get("health_check_path", "/health")
3483 env_vars = spec.get("env", {})
3484 # Stable block hashing across data-parallel ranks: identical prompts
3485 # must hash identically so shared prefix-cache hits are not lost
3486 # between pods. The disaggregated serving image (upstream vLLM) no
3487 # longer bakes this in, so default it here; an explicit spec env wins.
3488 env_vars = {"PYTHONHASHSEED": "0", **env_vars}
3489 resources = spec.get("resources", {})
3490 model_path = spec.get("model_path")
3491 command = spec.get("command")
3492 args = spec.get("args")
3494 # Build container
3495 container_env = [client.V1EnvVar(name=k, value=str(v)) for k, v in env_vars.items()]
3497 # Runtime behavior is persisted explicitly by callers that need a
3498 # strict adapter. Legacy endpoints without the field retain vLLM image
3499 # detection, but TGI is never given vLLM's unsupported --root-path:
3500 # the authenticated platform proxy strips /inference/{name} before
3501 # forwarding /health, /generate, or /info to the model Service.
3502 serving_prefix = f"/inference/{name}"
3503 runtime_framework = spec.get("framework")
3504 if runtime_framework not in ("vllm", "tgi"):
3505 image_lower = image.lower()
3506 if "vllm" in image_lower:
3507 runtime_framework = "vllm"
3508 elif "text-generation-inference" in image_lower or "/tgi" in image_lower:
3509 runtime_framework = "tgi"
3510 else:
3511 runtime_framework = None
3512 if not command and runtime_framework == "vllm":
3513 if args:
3514 if "--root-path" not in args:
3515 args = list(args) + ["--root-path", serving_prefix]
3516 else:
3517 args = ["--root-path", serving_prefix]
3519 # Append caller-supplied arguments (for example the rendered
3520 # --kv-transfer-config) after any root-path injection so they survive
3521 # alongside user --extra-args.
3522 if extra_args:
3523 args = (list(args) if args else []) + list(extra_args)
3525 resource_reqs = client.V1ResourceRequirements(
3526 requests=resources.get("requests", {"cpu": "1", "memory": "4Gi"}),
3527 limits=resources.get("limits", {"cpu": "4", "memory": "16Gi"}),
3528 )
3529 # Add accelerator resources (GPU or Neuron)
3530 accelerator = spec.get("accelerator", "nvidia")
3531 if gpu_count > 0:
3532 if accelerator == "neuron":
3533 # AWS Trainium/Inferentia — request Neuron devices
3534 if resource_reqs.limits is None:
3535 resource_reqs.limits = {}
3536 resource_reqs.limits["aws.amazon.com/neuron"] = str(gpu_count)
3537 if resource_reqs.requests is None:
3538 resource_reqs.requests = {}
3539 resource_reqs.requests["aws.amazon.com/neuron"] = str(gpu_count)
3540 else:
3541 # NVIDIA GPU (default)
3542 if resource_reqs.limits is None:
3543 resource_reqs.limits = {}
3544 resource_reqs.limits["nvidia.com/gpu"] = str(gpu_count)
3545 if resource_reqs.requests is None:
3546 resource_reqs.requests = {}
3547 resource_reqs.requests["nvidia.com/gpu"] = str(gpu_count)
3549 volume_mounts = []
3550 volumes = []
3551 init_containers = []
3552 model_source = spec.get("model_source")
3554 if model_path or model_source:
3555 volume_mounts.append(
3556 client.V1VolumeMount(
3557 name="model-storage",
3558 mount_path="/models",
3559 )
3560 )
3561 volumes.append(
3562 client.V1Volume(
3563 name="model-storage",
3564 persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
3565 claim_name="efs-claim",
3566 ),
3567 )
3568 )
3570 # Mooncake role pods read the shared transport config (metadata-server
3571 # address, protocol, device) from the per-endpoint ``{name}-mooncake``
3572 # ConfigMap mounted read-only at MOONCAKE_CONFIG_MOUNT_DIR, pointed at by
3573 # MOONCAKE_CONFIG_PATH, and learn the KV-transfer bootstrap base port via
3574 # VLLM_MOONCAKE_BOOTSTRAP_PORT. Plain (non-mooncake) endpoints are
3575 # untouched: no volume, mount, or env is added.
3576 mooncake_block = spec.get("mooncake")
3577 if mooncake_block:
3578 volume_mounts.append(
3579 client.V1VolumeMount(
3580 name="mooncake-config",
3581 mount_path=MOONCAKE_CONFIG_MOUNT_DIR,
3582 read_only=True,
3583 )
3584 )
3585 volumes.append(
3586 client.V1Volume(
3587 name="mooncake-config",
3588 config_map=client.V1ConfigMapVolumeSource(name=f"{name}-mooncake"),
3589 )
3590 )
3591 transfer_block = mooncake_block.get("transfer") or {}
3592 base_port = transfer_block.get("bootstrap_base_port", MOONCAKE_BOOTSTRAP_BASE_PORT)
3593 try:
3594 base_port = int(base_port)
3595 except TypeError, ValueError:
3596 base_port = MOONCAKE_BOOTSTRAP_BASE_PORT
3597 container_env.append(
3598 client.V1EnvVar(name=MOONCAKE_CONFIG_PATH_ENV, value=MOONCAKE_CONFIG_FILE_PATH)
3599 )
3600 container_env.append(
3601 client.V1EnvVar(name=VLLM_MOONCAKE_BOOTSTRAP_PORT_ENV, value=str(base_port))
3602 )
3604 # Sync directly with literal argv so a model URI can never become
3605 # shell syntax. ``aws s3 sync`` avoids retransferring unchanged
3606 # objects on reruns while also repairing partial downloads.
3607 if model_source and model_source.startswith("s3://"):
3608 model_dest = f"/models/{name}"
3609 init_containers.append(
3610 client.V1Container(
3611 name="model-sync",
3612 image=AWS_CLI_IMAGE,
3613 command=["aws"],
3614 args=["s3", "sync", model_source, model_dest, "--quiet"],
3615 volume_mounts=[
3616 client.V1VolumeMount(
3617 name="model-storage",
3618 mount_path="/models",
3619 )
3620 ],
3621 resources=client.V1ResourceRequirements(
3622 requests={"cpu": "1", "memory": "2Gi"},
3623 limits={"cpu": "4", "memory": "8Gi"},
3624 ),
3625 )
3626 )
3628 # Probe path depends on whether the server handles the prefix
3629 uses_root_path = args is not None and "--root-path" in args
3630 probe_health = f"{serving_prefix}{health_path}" if uses_root_path else health_path
3632 container = client.V1Container(
3633 name="inference",
3634 image=image,
3635 ports=[client.V1ContainerPort(container_port=port)],
3636 env=container_env if container_env else None,
3637 resources=resource_reqs,
3638 volume_mounts=volume_mounts if volume_mounts else None,
3639 command=command,
3640 args=args,
3641 startup_probe=(
3642 client.V1Probe(
3643 http_get=client.V1HTTPGetAction(path=health_path, port=port),
3644 period_seconds=15,
3645 failure_threshold=80,
3646 )
3647 if runtime_framework == "tgi"
3648 else None
3649 ),
3650 liveness_probe=client.V1Probe(
3651 http_get=client.V1HTTPGetAction(path=probe_health, port=port),
3652 initial_delay_seconds=120,
3653 period_seconds=15,
3654 failure_threshold=5,
3655 ),
3656 readiness_probe=client.V1Probe(
3657 http_get=client.V1HTTPGetAction(path=probe_health, port=port),
3658 initial_delay_seconds=30,
3659 period_seconds=10,
3660 ),
3661 )
3663 # Build tolerations based on accelerator type
3664 if accelerator == "neuron":
3665 tolerations = [
3666 client.V1Toleration(
3667 key="aws.amazon.com/neuron",
3668 operator="Equal",
3669 value="true",
3670 effect="NoSchedule",
3671 )
3672 ]
3673 else:
3674 tolerations = [
3675 client.V1Toleration(
3676 key="nvidia.com/gpu",
3677 operator="Equal",
3678 value="true",
3679 effect="NoSchedule",
3680 )
3681 ]
3683 # Node selector based on accelerator type
3684 node_selector = spec.get("node_selector", {})
3685 if gpu_count > 0 and not node_selector:
3686 if accelerator == "neuron":
3687 node_selector = {"accelerator": "neuron"}
3688 else:
3689 node_selector = {"eks.amazonaws.com/instance-gpu-manufacturer": "nvidia"}
3691 # Apply capacity type preference (spot/on-demand)
3692 capacity_type = spec.get("capacity_type")
3693 if capacity_type in ("spot", "on-demand"):
3694 node_selector["karpenter.sh/capacity-type"] = capacity_type
3696 labels = {
3697 "app": app_label,
3698 "project": "gco",
3699 "gco.io/type": "inference",
3700 }
3701 if extra_labels:
3702 labels.update(extra_labels)
3704 deployment = client.V1Deployment(
3705 metadata=client.V1ObjectMeta(
3706 name=deploy_name,
3707 namespace=namespace,
3708 labels=dict(labels),
3709 annotations=self._provenance_annotations(),
3710 ),
3711 spec=client.V1DeploymentSpec(
3712 replicas=replicas,
3713 selector=client.V1LabelSelector(
3714 match_labels={"app": app_label},
3715 ),
3716 template=client.V1PodTemplateSpec(
3717 metadata=client.V1ObjectMeta(
3718 labels=dict(labels),
3719 annotations=self._provenance_annotations(),
3720 ),
3721 spec=client.V1PodSpec(
3722 service_account_name="gco-service-account",
3723 automount_service_account_token=False,
3724 containers=[container],
3725 init_containers=init_containers if init_containers else None,
3726 tolerations=tolerations,
3727 node_selector=node_selector if node_selector else None,
3728 volumes=volumes if volumes else None,
3729 ),
3730 ),
3731 ),
3732 )
3734 return deployment
3736 def _replica_count_for_role(self, mooncake: dict[str, Any], role: str) -> int:
3737 """Resolve the materialized replica count for a role.
3739 When per-role autoscaling is enabled and supplies a ``min_replicas``
3740 for the role, the role Deployment is materialized at that lower bound so
3741 the autoscaler owns the count from there. Otherwise the count comes from
3742 the topology: ``topology.prefill`` for prefill and ``topology.decode``
3743 for decode. The single store instance is always one replica.
3744 """
3745 autoscaling = mooncake.get("autoscaling") or {}
3746 if autoscaling.get("enabled") and role in ("prefill", "decode"):
3747 role_cfg = autoscaling.get(role) or {}
3748 min_replicas = role_cfg.get("min_replicas")
3749 if isinstance(min_replicas, int) and not isinstance(min_replicas, bool):
3750 return min_replicas
3752 topology = mooncake.get("topology") or {}
3753 if role == "prefill":
3754 return int(topology.get("prefill", 1))
3755 if role == "decode":
3756 return int(topology.get("decode", 1))
3757 # Single store instance: kv_both runs as one replica.
3758 return 1
3760 def _create_role_deployment(self, name: str, ns: str, spec: dict[str, Any], role: str) -> None:
3761 """Materialize one role Deployment for a Mooncake endpoint.
3763 Disaggregated and ``both`` modes split work across ``{name}-prefill``
3764 and ``{name}-decode``; store mode runs a single ``{name}`` instance with
3765 the ``kv_both`` role. The role's ``--kv-transfer-config`` is attached to
3766 the vLLM container, EFA scheduling is applied when transfer runs over
3767 RDMA, and the replica count is taken from the topology (or the
3768 autoscaling lower bound when that is enabled).
3770 Args:
3771 name: The endpoint name.
3772 ns: The namespace to materialize into.
3773 spec: The endpoint spec; ``spec["mooncake"]`` selects the mode and
3774 topology.
3775 role: One of ``"prefill"``, ``"decode"``, or ``"single"``.
3776 """
3777 mooncake = spec.get("mooncake") or {}
3779 # The store's single instance keeps the endpoint name; prefill and decode
3780 # are suffixed so Services and autoscalers can target each role.
3781 deploy_name = name if role == "single" else f"{name}-{role}"
3783 kv_transfer_config = build_kv_transfer_config(mooncake, role)
3784 replicas = self._replica_count_for_role(mooncake, role)
3786 deployment = self._build_inference_deployment_object(
3787 name=name,
3788 deploy_name=deploy_name,
3789 app_label=deploy_name,
3790 namespace=ns,
3791 spec=spec,
3792 replicas=replicas,
3793 extra_args=["--kv-transfer-config", kv_transfer_config],
3794 extra_labels={"gco.io/role": role},
3795 )
3797 # Land role pods on the EFA fabric when KV transfer runs over RDMA,
3798 # preserving the GPU asks already built into the pod.
3799 apply_efa_scheduling(mooncake, deployment.spec.template.spec)
3801 self._assert_mutation_authority()
3802 self.apps_v1.create_namespaced_deployment(
3803 ns, deployment, _request_timeout=self._k8s_timeout
3804 )
3805 self._confirm_created_resource(
3806 kind="deployment",
3807 resource_name=deploy_name,
3808 read_resource=partial(
3809 self.apps_v1.read_namespaced_deployment,
3810 deploy_name,
3811 ns,
3812 _request_timeout=self._k8s_timeout,
3813 ),
3814 delete_resource=partial(
3815 self.apps_v1.delete_namespaced_deployment,
3816 deploy_name,
3817 ns,
3818 _request_timeout=self._k8s_timeout,
3819 ),
3820 )
3821 logger.info("Created role deployment %s/%s (role=%s)", ns, deploy_name, role)
3823 def _verify_admin_api_key_secret(self, proxy: dict[str, Any], ns: str) -> str:
3824 """Confirm the proxy admin key Secret exists and carries a key value.
3826 The proxy guards a privileged admin path and must never run without a
3827 usable ``ADMIN_API_KEY``. This reads the Secret named by
3828 ``proxy.admin_api_key_secret`` and confirms it holds a non-empty
3829 ``ADMIN_API_KEY`` value, so the value itself never has to be carried on
3830 the spec or a command argument.
3832 Args:
3833 proxy: The ``spec["mooncake"]["proxy"]`` block; ``admin_api_key_secret``
3834 names the backing Secret.
3835 ns: The namespace the Secret lives in.
3837 Returns:
3838 The verified Secret name, suitable for a Secret reference.
3840 Raises:
3841 AdminApiKeySecretError: If the spec names no Secret, the Secret is
3842 absent, or its ``ADMIN_API_KEY`` value is empty or missing.
3843 """
3844 secret_name = proxy.get("admin_api_key_secret")
3845 if not isinstance(secret_name, str) or not secret_name:
3846 raise AdminApiKeySecretError(None, "no admin API key Secret was named")
3848 try:
3849 secret = self.core_v1.read_namespaced_secret(
3850 secret_name, ns, _request_timeout=self._k8s_timeout
3851 )
3852 except ApiException as e:
3853 if e.status == 404:
3854 raise AdminApiKeySecretError(secret_name, "Secret not found") from e
3855 raise
3857 if not self._secret_has_admin_api_key(secret):
3858 raise AdminApiKeySecretError(
3859 secret_name,
3860 f"{ADMIN_API_KEY_SECRET_DATA_KEY} value is empty or missing",
3861 )
3863 return secret_name
3865 @staticmethod
3866 def _secret_has_admin_api_key(secret: client.V1Secret) -> bool:
3867 """Return whether ``secret`` carries a non-empty ``ADMIN_API_KEY``.
3869 Both the base64 ``data`` and the plaintext ``string_data`` views are
3870 considered, and a value is treated as present only when it decodes to a
3871 non-empty string.
3872 """
3873 string_data = secret.string_data or {}
3874 plain = string_data.get(ADMIN_API_KEY_SECRET_DATA_KEY)
3875 if plain:
3876 return True
3878 data = secret.data or {}
3879 encoded = data.get(ADMIN_API_KEY_SECRET_DATA_KEY)
3880 if not encoded:
3881 return False
3882 try:
3883 return bool(base64.b64decode(encoded))
3884 except ValueError, TypeError:
3885 # A value that cannot be decoded is unusable as an admin key.
3886 return False
3888 def _ensure_admin_api_key_secret(
3889 self,
3890 name: str,
3891 proxy: dict[str, Any],
3892 ns: str,
3893 lifecycle_id: str,
3894 ) -> str:
3895 """Return the proxy admin-key Secret name, provisioning one if needed.
3897 The prefill-decode proxy guards a privileged admin path and must never
3898 run without a usable ``ADMIN_API_KEY``. Two paths satisfy that:
3900 - **Bring-your-own**: when the proxy block names a Secret, that Secret
3901 must already exist and carry a non-empty ``ADMIN_API_KEY``; otherwise
3902 the deployment is rejected, so a typo or a missing pre-created Secret
3903 fails fast. The named Secret is only read, never created or mutated.
3904 - **Auto-managed**: when the proxy names no Secret, a per-endpoint
3905 ``{name}-admin`` Secret is provisioned create-if-absent with a
3906 generated key, so a split deploy needs no manual Secret. The generated
3907 key only ever lives in the cluster — it is never written to the
3908 endpoint spec, a command argument, or a log line.
3910 Args:
3911 name: The endpoint name, used to derive the auto-managed Secret name.
3912 proxy: The ``spec["mooncake"]["proxy"]`` block.
3913 ns: The namespace the Secret lives in.
3915 Returns:
3916 The Secret name to reference from the proxy container.
3918 Raises:
3919 AdminApiKeySecretError: Only on the bring-your-own path, when the
3920 named Secret is absent or its ``ADMIN_API_KEY`` is empty. The
3921 auto-managed path never raises this.
3922 """
3923 named = proxy.get("admin_api_key_secret")
3924 if isinstance(named, str) and named:
3925 return self._verify_admin_api_key_secret(proxy, ns)
3926 return self._provision_admin_api_key_secret(f"{name}-admin", ns, lifecycle_id)
3928 def _provision_admin_api_key_secret(
3929 self,
3930 secret_name: str,
3931 ns: str,
3932 lifecycle_id: str,
3933 ) -> str:
3934 """Create the auto-managed proxy admin-key Secret if absent.
3936 Uses create-if-absent semantics so the key stays stable across reconcile
3937 passes: an existing Secret (the steady state, or one a prior pass
3938 created) is left untouched, and a concurrent create (409) is treated as
3939 success. A freshly created Secret carries a cryptographically strong
3940 64-character hex ``ADMIN_API_KEY`` from :func:`secrets.token_hex`, which
3941 reaches the proxy only through a Secret reference — the value is never
3942 logged or written to the spec.
3944 Args:
3945 secret_name: The Secret to ensure exists (``{endpoint}-admin``).
3946 ns: The namespace to create it in.
3948 Returns:
3949 The Secret name, ready for a Secret reference.
3950 """
3951 try:
3952 existing = self.core_v1.read_namespaced_secret(
3953 secret_name,
3954 ns,
3955 _request_timeout=self._k8s_timeout,
3956 )
3957 except ApiException as e:
3958 if e.status != 404:
3959 raise
3960 else:
3961 existing = self._authorize_resource(
3962 existing,
3963 kind="secret",
3964 resource_name=secret_name,
3965 patch_metadata=partial(
3966 self.core_v1.patch_namespaced_secret,
3967 secret_name,
3968 ns,
3969 _request_timeout=self._k8s_timeout,
3970 ),
3971 read_resource=lambda: self.core_v1.read_namespaced_secret(
3972 secret_name, ns, _request_timeout=self._k8s_timeout
3973 ),
3974 delete_resource=partial(
3975 self.core_v1.delete_namespaced_secret,
3976 secret_name,
3977 ns,
3978 _request_timeout=self._k8s_timeout,
3979 ),
3980 )
3981 self._require_owned_admin_secret(
3982 existing,
3983 secret_name,
3984 ns,
3985 lifecycle_id,
3986 )
3987 return secret_name
3989 ownership = self._generated_admin_secret_labels(secret_name)
3990 annotations = self._provenance_annotations() or {_LIFECYCLE_ANNOTATION: lifecycle_id}
3991 secret = client.V1Secret(
3992 metadata=client.V1ObjectMeta(
3993 name=secret_name,
3994 namespace=ns,
3995 labels=ownership,
3996 annotations=annotations,
3997 ),
3998 string_data={ADMIN_API_KEY_SECRET_DATA_KEY: secrets.token_hex(32)},
3999 type="Opaque",
4000 )
4001 # The two logger.info calls below carry a bare `# nosemgrep`: the
4002 # logger-credential-disclosure rule matches the literal word "Secret" in
4003 # the message, but only the Secret's name and namespace (%s/%s) are
4004 # logged here — never the generated key value set above in string_data.
4005 self._assert_mutation_authority()
4006 try:
4007 self.core_v1.create_namespaced_secret(ns, secret, _request_timeout=self._k8s_timeout)
4008 self._confirm_created_resource(
4009 kind="secret",
4010 resource_name=secret_name,
4011 read_resource=partial(
4012 self.core_v1.read_namespaced_secret,
4013 secret_name,
4014 ns,
4015 _request_timeout=self._k8s_timeout,
4016 ),
4017 delete_resource=partial(
4018 self.core_v1.delete_namespaced_secret,
4019 secret_name,
4020 ns,
4021 _request_timeout=self._k8s_timeout,
4022 ),
4023 )
4024 logger.info("Provisioned proxy admin-key Secret %s/%s", ns, secret_name) # nosemgrep
4025 except ApiException as e:
4026 if e.status != 409:
4027 raise
4028 existing = self.core_v1.read_namespaced_secret(
4029 secret_name,
4030 ns,
4031 _request_timeout=self._k8s_timeout,
4032 )
4033 existing = self._authorize_resource(
4034 existing,
4035 kind="secret",
4036 resource_name=secret_name,
4037 patch_metadata=partial(
4038 self.core_v1.patch_namespaced_secret,
4039 secret_name,
4040 ns,
4041 _request_timeout=self._k8s_timeout,
4042 ),
4043 read_resource=lambda: self.core_v1.read_namespaced_secret(
4044 secret_name, ns, _request_timeout=self._k8s_timeout
4045 ),
4046 delete_resource=partial(
4047 self.core_v1.delete_namespaced_secret,
4048 secret_name,
4049 ns,
4050 _request_timeout=self._k8s_timeout,
4051 ),
4052 )
4053 try:
4054 self._require_owned_admin_secret(
4055 existing,
4056 secret_name,
4057 ns,
4058 lifecycle_id,
4059 )
4060 except AdminApiKeySecretError as error:
4061 raise AdminApiKeySecretError(
4062 secret_name,
4063 "a concurrent conventional Secret lacks matching lifecycle provenance",
4064 ) from error
4065 logger.info("Proxy admin-key Secret %s/%s exists", ns, secret_name) # nosemgrep
4066 return secret_name
4068 def _create_pd_proxy(
4069 self, name: str, ns: str, spec: dict[str, Any], endpoint: dict[str, Any]
4070 ) -> None:
4071 """Materialize the prefill-decode proxy front for a disaggregated endpoint.
4073 Disaggregated and ``both`` modes are fronted by a lightweight proxy that
4074 runs the residency check and dispatches each request to the prefill and
4075 decode pods. It materializes a ConfigMap, a proxy Deployment with at
4076 least one replica, and a Service whose selector matches only the proxy
4077 pods. The shared HTTPRoute attached to ``gco-system/gco-gateway`` sends
4078 ``/inference`` to ``gco-system/inference-proxy``; that authenticated
4079 platform proxy then reaches this internal ClusterIP Service.
4080 Endpoint-specific Ingresses are removed as an unsafe legacy path.
4082 Before those resources are created, a user-named
4083 ``mooncake.proxy.admin_api_key_secret`` is verified to contain a usable
4084 ``ADMIN_API_KEY``. When no Secret is named, the monitor auto-provisions
4085 a generated ``{name}-admin`` Secret instead. A missing or empty named
4086 Secret rejects the proxy; the key itself reaches the container only as
4087 a Secret reference at pod start and is never written to the spec or a
4088 command argument.
4090 Creation is idempotent at the API boundary: an already-present Deployment
4091 or ClusterIP Service is left in place, and historical direct Ingresses are
4092 deleted if present. No endpoint Gateway or HTTPRoute is created.
4094 Args:
4095 name: The endpoint name.
4096 ns: The namespace to materialize into.
4097 spec: The endpoint spec; ``spec["mooncake"]`` supplies the proxy
4098 image and behavior.
4099 endpoint: The endpoint record. Legacy per-endpoint routing metadata
4100 is ignored because the shared platform HTTPRoute owns the prefix.
4102 Raises:
4103 AdminApiKeySecretError: If the admin key Secret is missing, names no
4104 Secret, or holds an empty ``ADMIN_API_KEY``. No proxy resource
4105 is created in that case.
4106 """
4107 mooncake = spec.get("mooncake") or {}
4108 proxy = mooncake.get("proxy") or {}
4109 proxy_name = f"{name}-proxy"
4110 lifecycle_id = endpoint.get("lifecycle_id")
4111 if not isinstance(lifecycle_id, str) or not lifecycle_id:
4112 raise AdminApiKeySecretError(
4113 None,
4114 "endpoint record has no immutable lifecycle identity",
4115 )
4117 # The proxy fronts a privileged admin path, so it never starts without a
4118 # usable admin key. When the spec names a Secret it must already exist
4119 # and be non-empty (the deployment is rejected otherwise); when it names
4120 # none, a per-endpoint admin-key Secret is auto-provisioned with a
4121 # generated key. Either way the key reaches the container only by Secret
4122 # reference.
4123 admin_secret_name = self._ensure_admin_api_key_secret(
4124 name,
4125 proxy,
4126 ns,
4127 lifecycle_id,
4128 )
4130 proxy_env = build_pd_proxy_config(mooncake)
4131 container_env = [client.V1EnvVar(name=k, value=v) for k, v in proxy_env.items()]
4132 # Deliver the admin key by Secret reference only — its value is never
4133 # placed on the spec or a command argument.
4134 container_env.append(
4135 client.V1EnvVar(
4136 name=PD_PROXY_ADMIN_API_KEY_ENV,
4137 value_from=client.V1EnvVarSource(
4138 secret_key_ref=client.V1SecretKeySelector(
4139 name=admin_secret_name,
4140 key=ADMIN_API_KEY_SECRET_DATA_KEY,
4141 )
4142 ),
4143 )
4144 )
4146 # The proxy fronts at least one replica; a spec may ask for more.
4147 replicas = proxy.get("replicas", 1)
4148 if not isinstance(replicas, int) or isinstance(replicas, bool) or replicas < 1:
4149 replicas = 1
4151 labels = {
4152 "app": proxy_name,
4153 "project": "gco",
4154 "gco.io/type": "inference",
4155 "gco.io/role": PD_PROXY_ROLE_LABEL,
4156 }
4158 # The proxy reaches prefill and decode through their per-role Services
4159 # and listens on PD_PROXY_PORT for requests from the authenticated API
4160 # proxy. Routing via Services means only Ready role pods receive traffic.
4161 port = spec.get("port", 8000)
4162 container_env.extend(
4163 [
4164 client.V1EnvVar(name=PD_PROXY_PORT_ENV, value=str(PD_PROXY_PORT)),
4165 client.V1EnvVar(
4166 name=PD_PROXY_PREFILL_URL_ENV, value=f"http://{name}-prefill:{port}"
4167 ),
4168 client.V1EnvVar(name=PD_PROXY_DECODE_URL_ENV, value=f"http://{name}-decode:{port}"),
4169 ]
4170 )
4172 # Ship the proxy program to the pod as a ConfigMap and run it from there.
4173 self._ensure_pd_proxy_configmap(name, ns)
4174 proxy_volume_name = "pd-proxy-script"
4176 container = client.V1Container(
4177 name="proxy",
4178 image=proxy.get("image"),
4179 command=["python3", PD_PROXY_SCRIPT_PATH],
4180 ports=[client.V1ContainerPort(container_port=PD_PROXY_PORT)],
4181 env=container_env if container_env else None,
4182 resources=client.V1ResourceRequirements(
4183 requests={"cpu": "250m", "memory": "256Mi"},
4184 limits={"cpu": "1", "memory": "1Gi"},
4185 ),
4186 volume_mounts=[
4187 client.V1VolumeMount(
4188 name=proxy_volume_name,
4189 mount_path=PD_PROXY_CONFIG_MOUNT_DIR,
4190 read_only=True,
4191 )
4192 ],
4193 readiness_probe=client.V1Probe(
4194 tcp_socket=client.V1TCPSocketAction(port=PD_PROXY_PORT),
4195 initial_delay_seconds=10,
4196 period_seconds=10,
4197 ),
4198 liveness_probe=client.V1Probe(
4199 tcp_socket=client.V1TCPSocketAction(port=PD_PROXY_PORT),
4200 initial_delay_seconds=30,
4201 period_seconds=15,
4202 failure_threshold=5,
4203 ),
4204 )
4206 deployment = client.V1Deployment(
4207 metadata=client.V1ObjectMeta(
4208 name=proxy_name,
4209 namespace=ns,
4210 labels=dict(labels),
4211 annotations=self._provenance_annotations(),
4212 ),
4213 spec=client.V1DeploymentSpec(
4214 replicas=replicas,
4215 selector=client.V1LabelSelector(match_labels={"app": proxy_name}),
4216 template=client.V1PodTemplateSpec(
4217 metadata=client.V1ObjectMeta(
4218 labels=dict(labels),
4219 annotations=self._provenance_annotations(),
4220 ),
4221 spec=client.V1PodSpec(
4222 service_account_name="gco-service-account",
4223 automount_service_account_token=False,
4224 containers=[container],
4225 volumes=[
4226 client.V1Volume(
4227 name=proxy_volume_name,
4228 config_map=client.V1ConfigMapVolumeSource(
4229 name=f"{name}-pd-proxy",
4230 default_mode=0o555,
4231 ),
4232 )
4233 ],
4234 ),
4235 ),
4236 ),
4237 )
4239 self._assert_mutation_authority()
4240 try:
4241 self.apps_v1.create_namespaced_deployment(
4242 ns, deployment, _request_timeout=self._k8s_timeout
4243 )
4244 self._confirm_created_resource(
4245 kind="deployment",
4246 resource_name=proxy_name,
4247 read_resource=partial(
4248 self.apps_v1.read_namespaced_deployment,
4249 proxy_name,
4250 ns,
4251 _request_timeout=self._k8s_timeout,
4252 ),
4253 delete_resource=partial(
4254 self.apps_v1.delete_namespaced_deployment,
4255 proxy_name,
4256 ns,
4257 _request_timeout=self._k8s_timeout,
4258 ),
4259 )
4260 logger.info("Created proxy deployment %s/%s", ns, proxy_name)
4261 except ApiException as error:
4262 if error.status != 409:
4263 raise
4264 existing = self._get_deployment(proxy_name, ns)
4265 if existing is None:
4266 raise ReconcileFencedError(
4267 "proxy deployment disappeared during reconciliation"
4268 ) from error
4269 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
4270 deployment.metadata.resource_version = resource_version
4271 self._assert_mutation_authority()
4272 self.apps_v1.patch_namespaced_deployment(
4273 proxy_name,
4274 ns,
4275 body=deployment,
4276 _request_timeout=self._k8s_timeout,
4277 )
4278 logger.info("Reconciled proxy deployment %s/%s", ns, proxy_name)
4280 self._create_proxy_service(proxy_name, ns)
4282 def _authorize_existing_service(self, name: str, namespace: str) -> Any:
4283 service = self.core_v1.read_namespaced_service(
4284 name, namespace, _request_timeout=self._k8s_timeout
4285 )
4286 return self._authorize_resource(
4287 service,
4288 kind="service",
4289 resource_name=name,
4290 patch_metadata=partial(
4291 self.core_v1.patch_namespaced_service,
4292 name,
4293 namespace,
4294 _request_timeout=self._k8s_timeout,
4295 ),
4296 read_resource=lambda: self.core_v1.read_namespaced_service(
4297 name, namespace, _request_timeout=self._k8s_timeout
4298 ),
4299 delete_resource=partial(
4300 self.core_v1.delete_namespaced_service,
4301 name,
4302 namespace,
4303 _request_timeout=self._k8s_timeout,
4304 ),
4305 )
4307 def _create_role_service(self, name: str, ns: str, role: str, port: int = 8000) -> None:
4308 """Create the ClusterIP Service that fronts one role's pods.
4310 Named ``{name}-{role}`` and selecting that role Deployment's app label,
4311 so the PD proxy can address prefill or decode by stable in-cluster DNS.
4312 Routing through a Service means kube-proxy load-balances across only the
4313 role's Ready pods, which is what gives the proxy ready-only decode
4314 routing without watching the Kubernetes API. Idempotent at the API
4315 boundary: an already-present Service is left in place.
4316 """
4317 deploy_name = f"{name}-{role}"
4318 service = client.V1Service(
4319 metadata=client.V1ObjectMeta(
4320 name=deploy_name,
4321 namespace=ns,
4322 labels={
4323 "app": deploy_name,
4324 "project": "gco",
4325 "gco.io/type": "inference",
4326 "gco.io/role": role,
4327 },
4328 annotations=self._provenance_annotations(),
4329 ),
4330 spec=client.V1ServiceSpec(
4331 selector=_inference_service_selector(deploy_name),
4332 ports=[client.V1ServicePort(port=port, target_port=port, protocol="TCP")],
4333 type="ClusterIP",
4334 ),
4335 )
4336 self._assert_mutation_authority()
4337 try:
4338 self.core_v1.create_namespaced_service(ns, service, _request_timeout=self._k8s_timeout)
4339 self._confirm_created_resource(
4340 kind="service",
4341 resource_name=deploy_name,
4342 read_resource=partial(
4343 self.core_v1.read_namespaced_service,
4344 deploy_name,
4345 ns,
4346 _request_timeout=self._k8s_timeout,
4347 ),
4348 delete_resource=partial(
4349 self.core_v1.delete_namespaced_service,
4350 deploy_name,
4351 ns,
4352 _request_timeout=self._k8s_timeout,
4353 ),
4354 )
4355 logger.info("Created role service %s/%s", ns, deploy_name)
4356 except ApiException as error:
4357 if error.status != 409:
4358 raise
4359 self._authorize_existing_service(deploy_name, ns)
4360 logger.info("Role service %s/%s already exists", ns, deploy_name)
4362 def _ensure_pd_proxy_configmap(self, name: str, ns: str) -> None:
4363 """Publish the PD proxy program to the pod as a ConfigMap.
4365 The proxy program (``mooncake_pd_proxy.py``) ships in this image
4366 alongside the monitor; its source is read here and mounted into the
4367 ``{name}-proxy`` pod, which runs it with ``python3`` from
4368 ``PD_PROXY_SCRIPT_PATH``. The ConfigMap is patched on conflict so the
4369 program tracks the running monitor build.
4370 """
4371 script = (Path(__file__).resolve().parent / PD_PROXY_SCRIPT_FILENAME).read_text(
4372 encoding="utf-8"
4373 )
4374 cm_name = f"{name}-pd-proxy"
4375 body = client.V1ConfigMap(
4376 metadata=client.V1ObjectMeta(
4377 name=cm_name,
4378 namespace=ns,
4379 labels={
4380 "app": f"{name}-proxy",
4381 "project": "gco",
4382 "gco.io/type": "inference",
4383 "gco.io/role": PD_PROXY_ROLE_LABEL,
4384 },
4385 annotations=self._provenance_annotations(),
4386 ),
4387 data={PD_PROXY_SCRIPT_FILENAME: script},
4388 )
4389 self._assert_mutation_authority()
4390 try:
4391 self.core_v1.create_namespaced_config_map(ns, body, _request_timeout=self._k8s_timeout)
4392 self._confirm_created_resource(
4393 kind="configmap",
4394 resource_name=cm_name,
4395 read_resource=partial(
4396 self.core_v1.read_namespaced_config_map,
4397 cm_name,
4398 ns,
4399 _request_timeout=self._k8s_timeout,
4400 ),
4401 delete_resource=partial(
4402 self.core_v1.delete_namespaced_config_map,
4403 cm_name,
4404 ns,
4405 _request_timeout=self._k8s_timeout,
4406 ),
4407 )
4408 logger.info("Created PD proxy ConfigMap %s/%s", ns, cm_name)
4409 except ApiException as error:
4410 if error.status != 409:
4411 raise
4412 existing = self.core_v1.read_namespaced_config_map(
4413 cm_name, ns, _request_timeout=self._k8s_timeout
4414 )
4415 existing = self._authorize_resource(
4416 existing,
4417 kind="configmap",
4418 resource_name=cm_name,
4419 patch_metadata=partial(
4420 self.core_v1.patch_namespaced_config_map,
4421 cm_name,
4422 ns,
4423 _request_timeout=self._k8s_timeout,
4424 ),
4425 read_resource=lambda: self.core_v1.read_namespaced_config_map(
4426 cm_name, ns, _request_timeout=self._k8s_timeout
4427 ),
4428 delete_resource=partial(
4429 self.core_v1.delete_namespaced_config_map,
4430 cm_name,
4431 ns,
4432 _request_timeout=self._k8s_timeout,
4433 ),
4434 )
4435 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
4436 body.metadata.resource_version = resource_version
4437 self._assert_mutation_authority()
4438 self.core_v1.patch_namespaced_config_map(
4439 cm_name, ns, body, _request_timeout=self._k8s_timeout
4440 )
4441 logger.info("Updated PD proxy ConfigMap %s/%s", ns, cm_name)
4443 def _create_proxy_service(self, proxy_name: str, namespace: str) -> None:
4444 """Create the Service that fronts only the proxy pods.
4446 The selector is the ``{name}-proxy`` app label together with the proxy
4447 role marker, so the Service resolves exclusively to proxy pods and never
4448 to the prefill or decode role pods that share the namespace.
4449 """
4450 service = client.V1Service(
4451 metadata=client.V1ObjectMeta(
4452 name=proxy_name,
4453 namespace=namespace,
4454 labels={
4455 "app": proxy_name,
4456 "project": "gco",
4457 "gco.io/type": "inference",
4458 "gco.io/role": PD_PROXY_ROLE_LABEL,
4459 },
4460 annotations=self._provenance_annotations(),
4461 ),
4462 spec=client.V1ServiceSpec(
4463 selector=_inference_service_selector(
4464 proxy_name, **{"gco.io/role": PD_PROXY_ROLE_LABEL}
4465 ),
4466 ports=[
4467 client.V1ServicePort(
4468 port=80,
4469 target_port=PD_PROXY_PORT,
4470 protocol="TCP",
4471 )
4472 ],
4473 type="ClusterIP",
4474 ),
4475 )
4477 self._assert_mutation_authority()
4478 try:
4479 self.core_v1.create_namespaced_service(
4480 namespace, service, _request_timeout=self._k8s_timeout
4481 )
4482 self._confirm_created_resource(
4483 kind="service",
4484 resource_name=proxy_name,
4485 read_resource=partial(
4486 self.core_v1.read_namespaced_service,
4487 proxy_name,
4488 namespace,
4489 _request_timeout=self._k8s_timeout,
4490 ),
4491 delete_resource=partial(
4492 self.core_v1.delete_namespaced_service,
4493 proxy_name,
4494 namespace,
4495 _request_timeout=self._k8s_timeout,
4496 ),
4497 )
4498 logger.info("Created proxy service %s/%s", namespace, proxy_name)
4499 except ApiException as error:
4500 if error.status != 409:
4501 raise
4502 self._authorize_existing_service(proxy_name, namespace)
4503 logger.info("Proxy service %s/%s already exists", namespace, proxy_name)
4505 def _create_service(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
4506 """Create the internal ClusterIP Service for an inference endpoint."""
4507 port = spec.get("port", 8000)
4509 service = client.V1Service(
4510 metadata=client.V1ObjectMeta(
4511 name=name,
4512 namespace=namespace,
4513 labels={
4514 "app": name,
4515 "project": "gco",
4516 "gco.io/type": "inference",
4517 },
4518 annotations=self._provenance_annotations(),
4519 ),
4520 spec=client.V1ServiceSpec(
4521 selector=_inference_service_selector(name),
4522 ports=[
4523 client.V1ServicePort(
4524 port=80,
4525 target_port=port,
4526 protocol="TCP",
4527 )
4528 ],
4529 type="ClusterIP",
4530 ),
4531 )
4533 self._assert_mutation_authority()
4534 try:
4535 self.core_v1.create_namespaced_service(
4536 namespace, service, _request_timeout=self._k8s_timeout
4537 )
4538 self._confirm_created_resource(
4539 kind="service",
4540 resource_name=name,
4541 read_resource=partial(
4542 self.core_v1.read_namespaced_service,
4543 name,
4544 namespace,
4545 _request_timeout=self._k8s_timeout,
4546 ),
4547 delete_resource=partial(
4548 self.core_v1.delete_namespaced_service,
4549 name,
4550 namespace,
4551 _request_timeout=self._k8s_timeout,
4552 ),
4553 )
4554 logger.info("Created service %s/%s", namespace, name)
4555 except ApiException as error:
4556 if error.status != 409:
4557 raise
4558 self._authorize_existing_service(name, namespace)
4559 logger.info("Service %s/%s already exists", namespace, name)
4561 def _ensure_service(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
4562 """Ensure an owned endpoint Service exists, recreating it if absent."""
4563 try:
4564 self._authorize_existing_service(name, namespace)
4565 except ApiException as error:
4566 if error.status == 404:
4567 logger.warning("Service %s/%s missing, recreating", namespace, name)
4568 self._create_service(name, namespace, spec)
4569 else:
4570 raise
4572 def _check_health_watchdog(
4573 self,
4574 name: str,
4575 namespace: str,
4576 ready_replicas: int,
4577 desired_replicas: int,
4578 spec: dict[str, Any],
4579 endpoint: dict[str, Any],
4580 ) -> bool:
4581 """Track prolonged unavailability without changing shared routing.
4583 ``gco-system/gco-gateway`` owns the shared ``/inference`` HTTPRoute to
4584 ``gco-system/inference-proxy``. Individual models are reached through
4585 internal ClusterIP Services, so their readiness never changes the shared
4586 Gateway or HTTPRoute. The threshold still drives degraded-state logging.
4587 """
4588 del namespace, spec, endpoint
4589 if ready_replicas > 0:
4590 if name in self._unready_since:
4591 logger.info("Endpoint %s recovered", name)
4592 del self._unready_since[name]
4593 return False
4595 now = datetime.now(UTC)
4596 if name not in self._unready_since:
4597 self._unready_since[name] = now
4598 logger.warning(
4599 "Endpoint %s has 0/%d ready replicas, starting health watchdog timer",
4600 name,
4601 desired_replicas,
4602 )
4603 return False
4605 unready_duration = (now - self._unready_since[name]).total_seconds()
4606 threshold_exceeded = unready_duration >= self._unhealthy_threshold_seconds
4607 if threshold_exceeded:
4608 logger.warning(
4609 "WATCHDOG: Endpoint %s has been unavailable for %ds (threshold %ds); "
4610 "the authenticated proxy will return 503 until it recovers",
4611 name,
4612 int(unready_duration),
4613 self._unhealthy_threshold_seconds,
4614 )
4615 return threshold_exceeded
4617 def _scale_deployment(self, name: str, namespace: str, replicas: int) -> None:
4618 """Scale only the exact authorized Deployment resourceVersion."""
4619 deployment = self._get_deployment(name, namespace)
4620 if deployment is None:
4621 raise ReconcileFencedError(f"deployment {name} disappeared before scaling")
4622 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
4623 self._assert_mutation_authority()
4624 body: dict[str, Any] = {"spec": {"replicas": replicas}}
4625 if resource_version:
4626 body["metadata"] = {"resourceVersion": resource_version}
4627 self.apps_v1.patch_namespaced_deployment(
4628 name,
4629 namespace,
4630 body=body,
4631 _request_timeout=self._k8s_timeout,
4632 )
4634 def _update_deployment_image(self, name: str, namespace: str, image: str) -> None:
4635 """Update only the exact authorized Deployment resourceVersion."""
4636 deployment = self._get_deployment(name, namespace)
4637 if deployment is None:
4638 raise ReconcileFencedError(f"deployment {name} disappeared before image update")
4639 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
4640 self._assert_mutation_authority()
4641 body: dict[str, Any] = {
4642 "spec": {"template": {"spec": {"containers": [{"name": "inference", "image": image}]}}}
4643 }
4644 if resource_version:
4645 body["metadata"] = {"resourceVersion": resource_version}
4646 self.apps_v1.patch_namespaced_deployment(
4647 name,
4648 namespace,
4649 body=body,
4650 _request_timeout=self._k8s_timeout,
4651 )
4653 def _reconcile_canary(
4654 self,
4655 name: str,
4656 namespace: str,
4657 spec: dict[str, Any],
4658 canary: dict[str, Any],
4659 endpoint: dict[str, Any],
4660 ) -> dict[str, Any]:
4661 """Reconcile a classic canary and return observed readiness for routing."""
4662 canary_image_value = canary.get("image")
4663 if not isinstance(canary_image_value, str) or not canary_image_value.strip():
4664 raise ValueError("canary.image must be a non-empty string")
4665 canary_image = canary_image_value.strip()
4667 canary_replicas = canary.get("replicas", 1)
4668 if (
4669 not isinstance(canary_replicas, int)
4670 or isinstance(canary_replicas, bool)
4671 or canary_replicas < 1
4672 ):
4673 raise ValueError("canary.replicas must be a positive integer")
4675 canary_weight = canary.get("weight", 10)
4676 if (
4677 not isinstance(canary_weight, int)
4678 or isinstance(canary_weight, bool)
4679 or not 1 <= canary_weight <= 99
4680 ):
4681 raise ValueError("canary.weight must be an integer between 1 and 99")
4683 canary_name = f"{name}-canary"
4684 del endpoint # Per-endpoint routing metadata is legacy and intentionally ignored.
4686 canary_spec = dict(spec)
4687 canary_spec["image"] = canary_image
4688 canary_spec["replicas"] = canary_replicas
4689 canary_spec.pop("canary", None)
4690 # A canary image is explicit and global. Retaining the primary's
4691 # region_image_uris would silently deploy the old regional image.
4692 canary_spec.pop("region_image_uris", None)
4694 deployment = self._get_deployment(canary_name, namespace)
4695 state = "creating"
4696 ready_replicas = 0
4697 if deployment is None:
4698 logger.info("Creating canary deployment %s with image %s", canary_name, canary_image)
4699 self._create_deployment(canary_name, namespace, canary_spec)
4700 self._create_service(canary_name, namespace, canary_spec)
4701 else:
4702 self._ensure_service(canary_name, namespace, canary_spec)
4703 current_image = self._get_deployment_image(deployment)
4704 current_replicas = deployment.spec.replicas or 1
4705 ready_replicas = deployment.status.ready_replicas or 0
4706 if current_image != canary_image:
4707 self._update_deployment_image(canary_name, namespace, canary_image)
4708 ready_replicas = 0
4709 state = "updating"
4710 elif current_replicas != canary_replicas:
4711 self._scale_deployment(canary_name, namespace, canary_replicas)
4712 ready_replicas = min(ready_replicas, canary_replicas)
4713 state = "updating"
4714 elif ready_replicas >= canary_replicas:
4715 state = "running"
4717 # Canary selection happens behind ``gco-system/inference-proxy``; the
4718 # shared ``gco-system/gco-gateway`` HTTPRoute is never changed per
4719 # endpoint. The shared inference proxy consumes the observed canary
4720 # status returned here.
4721 return {
4722 "state": state,
4723 "image": canary_image,
4724 "weight": canary_weight,
4725 "replicas_ready": ready_replicas,
4726 "replicas_desired": canary_replicas,
4727 }
4729 def _cleanup_canary(self, name: str, namespace: str) -> None:
4730 """Remove only canary objects carrying this reconciliation authority."""
4731 canary_name = f"{name}-canary"
4732 deployment = self._get_deployment(canary_name, namespace)
4733 if deployment is not None:
4734 self._assert_mutation_authority()
4735 try:
4736 self.apps_v1.delete_namespaced_deployment(
4737 canary_name,
4738 namespace,
4739 body=self._delete_options_for(
4740 deployment, kind="deployment", resource_name=canary_name
4741 ),
4742 _request_timeout=self._k8s_timeout,
4743 )
4744 logger.info("Deleted canary deployment %s", canary_name)
4745 except ApiException as error:
4746 if error.status != 404:
4747 logger.error("Failed to delete canary deployment %s: %s", canary_name, error)
4748 try:
4749 service = self._authorize_existing_service(canary_name, namespace)
4750 except ApiException as error:
4751 if error.status != 404:
4752 logger.error("Failed to read canary service %s: %s", canary_name, error)
4753 else:
4754 self._assert_mutation_authority()
4755 try:
4756 self.core_v1.delete_namespaced_service(
4757 canary_name,
4758 namespace,
4759 body=self._delete_options_for(
4760 service, kind="service", resource_name=canary_name
4761 ),
4762 _request_timeout=self._k8s_timeout,
4763 )
4764 logger.info("Deleted canary service %s", canary_name)
4765 except ApiException as error:
4766 if error.status != 404:
4767 logger.error("Failed to delete canary service %s: %s", canary_name, error)
4769 @staticmethod
4770 def _endpoint_resource_inventory(name: str) -> EndpointResourceInventory:
4771 """Return every deterministic endpoint-owned resource name.
4773 The shared ``mooncake-master`` Service/StatefulSet and any Secret named
4774 by the user are deliberately excluded.
4775 """
4776 return EndpointResourceInventory(
4777 deployments=(
4778 name,
4779 f"{name}-canary",
4780 f"{name}-prefill",
4781 f"{name}-decode",
4782 f"{name}-proxy",
4783 ),
4784 services=(
4785 name,
4786 f"{name}-canary",
4787 f"{name}-prefill",
4788 f"{name}-decode",
4789 f"{name}-proxy",
4790 ),
4791 horizontal_pod_autoscalers=(
4792 name,
4793 f"{name}-prefill",
4794 f"{name}-decode",
4795 f"keda-hpa-{name}",
4796 f"keda-hpa-{name}-prefill",
4797 f"keda-hpa-{name}-decode",
4798 ),
4799 scaled_objects=(
4800 name,
4801 f"{name}-prefill",
4802 f"{name}-decode",
4803 ),
4804 config_maps=(f"{name}-mooncake", f"{name}-pd-proxy"),
4805 legacy_ingresses=(name, f"{name}-canary", f"{name}-proxy"),
4806 legacy_http_routes=(name, f"{name}-canary", f"{name}-proxy"),
4807 generated_admin_secret=f"{name}-admin",
4808 )
4810 @staticmethod
4811 def _cleanup_error(
4812 operation: str,
4813 kind: str,
4814 resource_name: str,
4815 error: Exception,
4816 ) -> str:
4817 """Build a bounded error containing only an endpoint-owned name."""
4818 status_value = getattr(error, "status", None)
4819 status = f" (status {status_value})" if status_value is not None else ""
4820 return f"{operation} {kind} {resource_name} failed{status}"
4822 def _delete_and_confirm(
4823 self,
4824 *,
4825 kind: str,
4826 resource_name: str,
4827 delete_call: Callable[..., Any],
4828 read_call: Callable[[], Any],
4829 patch_metadata: Callable[..., Any] | None,
4830 pending: list[str],
4831 errors: list[str],
4832 observed_resource: Any | None = None,
4833 ) -> bool:
4834 """Read-authorize-delete one exact UID/resourceVersion, then re-observe."""
4835 resource_id = f"{kind}/{resource_name}"
4836 if observed_resource is not None:
4837 observed = observed_resource
4838 else:
4839 try:
4840 observed = read_call()
4841 except ApiException as error:
4842 if error.status == 404:
4843 return False
4844 errors.append(self._cleanup_error("read", kind, resource_name, error))
4845 return False
4846 except Exception as error:
4847 errors.append(self._cleanup_error("read", kind, resource_name, error))
4848 return False
4850 observed = self._authorize_resource(
4851 observed,
4852 kind=kind,
4853 resource_name=resource_name,
4854 patch_metadata=patch_metadata,
4855 read_resource=read_call,
4856 delete_resource=delete_call,
4857 allow_region_mismatch=bool(
4858 self._active_authority and self._active_authority.region_removed
4859 ),
4860 )
4861 _metadata, _annotations, observed_uid, _resource_version = self._object_metadata(observed)
4862 self._assert_mutation_authority()
4863 try:
4864 delete_call(
4865 body=self._delete_options_for(observed, kind=kind, resource_name=resource_name)
4866 )
4867 except ApiException as error:
4868 if error.status != 404:
4869 errors.append(self._cleanup_error("delete", kind, resource_name, error))
4870 except Exception as error:
4871 errors.append(self._cleanup_error("delete", kind, resource_name, error))
4873 try:
4874 remaining = read_call()
4875 except ApiException as error:
4876 if error.status == 404:
4877 return True
4878 errors.append(self._cleanup_error("read", kind, resource_name, error))
4879 return True
4880 except Exception as error:
4881 errors.append(self._cleanup_error("read", kind, resource_name, error))
4882 return True
4884 _metadata, _annotations, remaining_uid, _remaining_version = self._object_metadata(
4885 remaining
4886 )
4887 # A replacement with a different UID is never deleted by this stale
4888 # observation. Its continued presence intentionally blocks cleanup.
4889 if observed_uid and remaining_uid and remaining_uid != observed_uid:
4890 pending.append(f"{resource_id}:replacement")
4891 else:
4892 pending.append(resource_id)
4893 return True
4895 @staticmethod
4896 def _merge_cleanup_results(*results: ResourceCleanupResult) -> ResourceCleanupResult:
4897 """Combine independently attempted cleanup groups without losing failures."""
4898 return ResourceCleanupResult(
4899 pending=tuple(item for result in results for item in result.pending),
4900 errors=tuple(item for result in results for item in result.errors),
4901 resources_found=any(result.resources_found for result in results),
4902 )
4904 def _delete_scaled_objects(
4905 self,
4906 names: tuple[str, ...],
4907 namespace: str,
4908 ) -> ResourceCleanupResult:
4909 """Delete and verify the requested KEDA ScaledObjects."""
4910 pending: list[str] = []
4911 errors: list[str] = []
4912 resources_found = False
4913 custom_objects = client.CustomObjectsApi()
4914 for autoscaler_name in names:
4915 resources_found |= self._delete_and_confirm(
4916 kind="scaledobject",
4917 resource_name=autoscaler_name,
4918 delete_call=partial(
4919 custom_objects.delete_namespaced_custom_object,
4920 group=KEDA_API_GROUP,
4921 version=KEDA_API_VERSION,
4922 namespace=namespace,
4923 plural=KEDA_SCALEDOBJECT_PLURAL,
4924 name=autoscaler_name,
4925 _request_timeout=self._k8s_timeout,
4926 ),
4927 read_call=partial(
4928 custom_objects.get_namespaced_custom_object,
4929 group=KEDA_API_GROUP,
4930 version=KEDA_API_VERSION,
4931 namespace=namespace,
4932 plural=KEDA_SCALEDOBJECT_PLURAL,
4933 name=autoscaler_name,
4934 _request_timeout=self._k8s_timeout,
4935 ),
4936 patch_metadata=partial(
4937 custom_objects.patch_namespaced_custom_object,
4938 group=KEDA_API_GROUP,
4939 version=KEDA_API_VERSION,
4940 namespace=namespace,
4941 plural=KEDA_SCALEDOBJECT_PLURAL,
4942 name=autoscaler_name,
4943 _request_timeout=self._k8s_timeout,
4944 ),
4945 pending=pending,
4946 errors=errors,
4947 )
4948 return ResourceCleanupResult(
4949 pending=tuple(pending),
4950 errors=tuple(errors),
4951 resources_found=resources_found,
4952 )
4954 def _delete_hpas(
4955 self,
4956 names: tuple[str, ...],
4957 namespace: str,
4958 ) -> ResourceCleanupResult:
4959 """Delete and verify the requested native or KEDA-generated HPAs."""
4960 pending: list[str] = []
4961 errors: list[str] = []
4962 resources_found = False
4963 autoscaling_v2 = client.AutoscalingV2Api()
4964 for autoscaler_name in names:
4965 resources_found |= self._delete_and_confirm(
4966 kind="hpa",
4967 resource_name=autoscaler_name,
4968 delete_call=partial(
4969 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
4970 autoscaler_name,
4971 namespace,
4972 _request_timeout=self._k8s_timeout,
4973 ),
4974 read_call=partial(
4975 autoscaling_v2.read_namespaced_horizontal_pod_autoscaler,
4976 autoscaler_name,
4977 namespace,
4978 _request_timeout=self._k8s_timeout,
4979 ),
4980 patch_metadata=partial(
4981 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
4982 autoscaler_name,
4983 namespace,
4984 _request_timeout=self._k8s_timeout,
4985 ),
4986 pending=pending,
4987 errors=errors,
4988 )
4989 return ResourceCleanupResult(
4990 pending=tuple(pending),
4991 errors=tuple(errors),
4992 resources_found=resources_found,
4993 )
4995 def _delete_autoscalers(
4996 self,
4997 scaled_object_names: tuple[str, ...],
4998 hpa_names: tuple[str, ...],
4999 namespace: str,
5000 ) -> ResourceCleanupResult:
5001 """Delete every autoscaler owner and wait for all of them to disappear."""
5002 # KEDA must stop first; only then is it safe to remove its generated HPA
5003 # alongside any native HPA that may remain from a prior configuration.
5004 scaled_objects = self._delete_scaled_objects(scaled_object_names, namespace)
5005 hpas = self._delete_hpas(hpa_names, namespace)
5006 return self._merge_cleanup_results(scaled_objects, hpas)
5008 @staticmethod
5009 def _generated_admin_secret_labels(expected_name: str) -> dict[str, str]:
5010 """Return the schema-valid static provenance used by generated Secrets."""
5011 return {
5012 "app": expected_name,
5013 "project": "gco",
5014 "gco.io/type": "inference",
5015 }
5017 @classmethod
5018 def _is_monitor_owned_admin_secret(
5019 cls,
5020 secret: Any,
5021 expected_name: str,
5022 expected_lifecycle_id: str,
5023 ) -> bool:
5024 """Return whether a generated Secret is bound to this endpoint lifecycle."""
5025 metadata = getattr(secret, "metadata", None)
5026 labels = getattr(metadata, "labels", None)
5027 annotations = getattr(metadata, "annotations", None)
5028 return (
5029 labels == cls._generated_admin_secret_labels(expected_name)
5030 and isinstance(annotations, dict)
5031 and annotations.get("gco.io/lifecycle-id") == expected_lifecycle_id
5032 )
5034 @classmethod
5035 def _is_legacy_monitor_admin_secret(cls, secret: Any, expected_name: str) -> bool:
5036 """Recognize only the exact pre-v7 monitor-generated Secret shape."""
5037 metadata = getattr(secret, "metadata", None)
5038 labels = getattr(metadata, "labels", None)
5039 annotations = getattr(metadata, "annotations", None)
5040 lifecycle_annotation = (
5041 annotations.get("gco.io/lifecycle-id") if isinstance(annotations, dict) else None
5042 )
5043 secret_type = getattr(secret, "type", None)
5044 return (
5045 labels == cls._generated_admin_secret_labels(expected_name)
5046 and lifecycle_annotation is None
5047 and secret_type == "Opaque"
5048 and cls._secret_has_admin_api_key(secret)
5049 )
5051 def _adopt_legacy_admin_secret(
5052 self,
5053 secret: Any,
5054 secret_name: str,
5055 namespace: str,
5056 lifecycle_id: str,
5057 ) -> Any:
5058 """Patch exact legacy provenance and verify the persisted lifecycle annotation."""
5059 if not self._is_legacy_monitor_admin_secret(secret, secret_name):
5060 raise AdminApiKeySecretError(
5061 secret_name,
5062 "the conventional generated Secret has ambiguous ownership",
5063 )
5064 metadata = getattr(secret, "metadata", None)
5065 resource_version = getattr(metadata, "resource_version", None)
5066 if not isinstance(resource_version, str) or not resource_version:
5067 raise AdminApiKeySecretError(
5068 secret_name,
5069 "legacy generated Secret has no resource version for safe migration",
5070 )
5071 annotations = self._provenance_annotations() or {_LIFECYCLE_ANNOTATION: lifecycle_id}
5072 self._assert_mutation_authority()
5073 try:
5074 self.core_v1.patch_namespaced_secret(
5075 secret_name,
5076 namespace,
5077 body={
5078 "metadata": {
5079 "resourceVersion": resource_version,
5080 "annotations": annotations,
5081 }
5082 },
5083 _request_timeout=self._k8s_timeout,
5084 )
5085 except Exception as error:
5086 raise AdminApiKeySecretError(
5087 secret_name,
5088 "legacy generated Secret changed during lifecycle migration",
5089 ) from error
5090 adopted = self.core_v1.read_namespaced_secret(
5091 secret_name,
5092 namespace,
5093 _request_timeout=self._k8s_timeout,
5094 )
5095 if not self._is_monitor_owned_admin_secret(
5096 adopted,
5097 secret_name,
5098 lifecycle_id,
5099 ) or not self._secret_has_admin_api_key(adopted):
5100 raise AdminApiKeySecretError(
5101 secret_name,
5102 "legacy generated Secret migration could not be verified",
5103 )
5104 return adopted
5106 def _require_owned_admin_secret(
5107 self,
5108 secret: Any,
5109 secret_name: str,
5110 namespace: str,
5111 lifecycle_id: str,
5112 ) -> Any:
5113 """Accept current provenance or safely migrate one exact legacy shape."""
5114 if self._is_monitor_owned_admin_secret(
5115 secret,
5116 secret_name,
5117 lifecycle_id,
5118 ) and self._secret_has_admin_api_key(secret):
5119 return secret
5120 if self._is_legacy_monitor_admin_secret(secret, secret_name):
5121 return self._adopt_legacy_admin_secret(
5122 secret,
5123 secret_name,
5124 namespace,
5125 lifecycle_id,
5126 )
5127 raise AdminApiKeySecretError(
5128 secret_name,
5129 "the conventional generated Secret exists without matching lifecycle provenance",
5130 )
5132 @staticmethod
5133 def _generated_child_matches(
5134 item: Any,
5135 kind: str,
5136 deployment_names: tuple[str, ...],
5137 service_names: tuple[str, ...],
5138 replica_set_names: tuple[str, ...] = (),
5139 ) -> bool:
5140 """Match generated children only through exact parent identity."""
5141 metadata = getattr(item, "metadata", None)
5142 child_name = getattr(metadata, "name", None)
5143 labels = getattr(metadata, "labels", None)
5144 labels = labels if isinstance(labels, dict) else {}
5145 owner_references = getattr(metadata, "owner_references", None)
5146 owners = owner_references if isinstance(owner_references, (list, tuple)) else ()
5148 if kind in {"replicaset", "pod"}:
5149 app_name = labels.get("app")
5150 if app_name not in deployment_names:
5151 return False
5152 if labels.get("project") != "gco" or labels.get("gco.io/type") != "inference":
5153 return False
5154 if not owners:
5155 return True
5156 for owner in owners:
5157 owner_kind = getattr(owner, "kind", None)
5158 owner_name = getattr(owner, "name", None)
5159 if owner_kind == "Deployment" and owner_name in deployment_names:
5160 return True
5161 if kind == "pod" and owner_kind == "ReplicaSet" and owner_name in replica_set_names:
5162 return True
5163 return False
5165 if kind == "endpoints":
5166 return isinstance(child_name, str) and child_name in service_names
5167 if kind == "endpointslice":
5168 service_name = labels.get("kubernetes.io/service-name")
5169 return isinstance(service_name, str) and service_name in service_names
5170 return False
5172 def _observe_generated_children(
5173 self,
5174 name: str,
5175 namespace: str,
5176 inventory: EndpointResourceInventory,
5177 pending: list[str],
5178 errors: list[str],
5179 ) -> bool:
5180 """Inventory exact endpoint children once per kind and dependency order."""
5181 resources_found = False
5182 matched_replica_sets: list[str] = []
5183 list_calls: tuple[tuple[str, Callable[[], Any]], ...] = (
5184 (
5185 "replicaset",
5186 partial(
5187 self.apps_v1.list_namespaced_replica_set,
5188 namespace,
5189 _request_timeout=self._k8s_timeout,
5190 ),
5191 ),
5192 (
5193 "pod",
5194 partial(
5195 self.core_v1.list_namespaced_pod,
5196 namespace,
5197 _request_timeout=self._k8s_timeout,
5198 ),
5199 ),
5200 (
5201 "endpoints",
5202 partial(
5203 self.core_v1.list_namespaced_endpoints,
5204 namespace,
5205 _request_timeout=self._k8s_timeout,
5206 ),
5207 ),
5208 (
5209 "endpointslice",
5210 partial(
5211 self.discovery_v1.list_namespaced_endpoint_slice,
5212 namespace,
5213 _request_timeout=self._k8s_timeout,
5214 ),
5215 ),
5216 )
5217 for kind, list_call in list_calls:
5218 try:
5219 response = list_call()
5220 except ApiException as error:
5221 errors.append(self._cleanup_error("list", kind, name, error))
5222 continue
5223 except Exception as error:
5224 errors.append(self._cleanup_error("list", kind, name, error))
5225 continue
5226 items = getattr(response, "items", None)
5227 if not isinstance(items, (list, tuple)):
5228 items = ()
5229 for item in items:
5230 if not self._generated_child_matches(
5231 item,
5232 kind,
5233 inventory.deployments,
5234 inventory.services,
5235 tuple(matched_replica_sets),
5236 ):
5237 continue
5238 metadata = getattr(item, "metadata", None)
5239 child_name = getattr(metadata, "name", "unknown")
5240 if kind == "replicaset" and isinstance(child_name, str):
5241 matched_replica_sets.append(child_name)
5242 pending.append(f"{kind}/{child_name}")
5243 resources_found = True
5244 return resources_found
5246 def _delete_resources(
5247 self,
5248 name: str,
5249 namespace: str,
5250 spec: dict[str, Any] | None = None,
5251 *,
5252 expected_lifecycle_id: str | None = None,
5253 ) -> ResourceCleanupResult:
5254 """Delete parents and prove all top-level/generated children absent."""
5255 inventory = self._endpoint_resource_inventory(name)
5256 autoscaler_cleanup = self._delete_autoscalers(
5257 inventory.scaled_objects,
5258 inventory.horizontal_pod_autoscalers,
5259 namespace,
5260 )
5261 pending = list(autoscaler_cleanup.pending)
5262 errors = list(autoscaler_cleanup.errors)
5263 resources_found = autoscaler_cleanup.resources_found
5265 for deployment_name in inventory.deployments:
5266 resources_found |= self._delete_and_confirm(
5267 kind="deployment",
5268 resource_name=deployment_name,
5269 delete_call=partial(
5270 self.apps_v1.delete_namespaced_deployment,
5271 deployment_name,
5272 namespace,
5273 _request_timeout=self._k8s_timeout,
5274 ),
5275 read_call=partial(
5276 self.apps_v1.read_namespaced_deployment,
5277 deployment_name,
5278 namespace,
5279 _request_timeout=self._k8s_timeout,
5280 ),
5281 patch_metadata=partial(
5282 self.apps_v1.patch_namespaced_deployment,
5283 deployment_name,
5284 namespace,
5285 _request_timeout=self._k8s_timeout,
5286 ),
5287 pending=pending,
5288 errors=errors,
5289 )
5291 for service_name in inventory.services:
5292 resources_found |= self._delete_and_confirm(
5293 kind="service",
5294 resource_name=service_name,
5295 delete_call=partial(
5296 self.core_v1.delete_namespaced_service,
5297 service_name,
5298 namespace,
5299 _request_timeout=self._k8s_timeout,
5300 ),
5301 read_call=partial(
5302 self.core_v1.read_namespaced_service,
5303 service_name,
5304 namespace,
5305 _request_timeout=self._k8s_timeout,
5306 ),
5307 patch_metadata=partial(
5308 self.core_v1.patch_namespaced_service,
5309 service_name,
5310 namespace,
5311 _request_timeout=self._k8s_timeout,
5312 ),
5313 pending=pending,
5314 errors=errors,
5315 )
5317 for config_map_name in inventory.config_maps:
5318 resources_found |= self._delete_and_confirm(
5319 kind="configmap",
5320 resource_name=config_map_name,
5321 delete_call=partial(
5322 self.core_v1.delete_namespaced_config_map,
5323 config_map_name,
5324 namespace,
5325 _request_timeout=self._k8s_timeout,
5326 ),
5327 read_call=partial(
5328 self.core_v1.read_namespaced_config_map,
5329 config_map_name,
5330 namespace,
5331 _request_timeout=self._k8s_timeout,
5332 ),
5333 patch_metadata=partial(
5334 self.core_v1.patch_namespaced_config_map,
5335 config_map_name,
5336 namespace,
5337 _request_timeout=self._k8s_timeout,
5338 ),
5339 pending=pending,
5340 errors=errors,
5341 )
5343 for ingress_name in inventory.legacy_ingresses:
5344 resources_found |= self._delete_and_confirm(
5345 kind="ingress",
5346 resource_name=ingress_name,
5347 delete_call=partial(
5348 self.networking_v1.delete_namespaced_ingress,
5349 ingress_name,
5350 namespace,
5351 _request_timeout=self._k8s_timeout,
5352 ),
5353 read_call=partial(
5354 self.networking_v1.read_namespaced_ingress,
5355 ingress_name,
5356 namespace,
5357 _request_timeout=self._k8s_timeout,
5358 ),
5359 patch_metadata=partial(
5360 self.networking_v1.patch_namespaced_ingress,
5361 ingress_name,
5362 namespace,
5363 _request_timeout=self._k8s_timeout,
5364 ),
5365 pending=pending,
5366 errors=errors,
5367 )
5369 custom_objects = client.CustomObjectsApi()
5370 for route_name in inventory.legacy_http_routes:
5371 resources_found |= self._delete_and_confirm(
5372 kind="httproute",
5373 resource_name=route_name,
5374 delete_call=partial(
5375 custom_objects.delete_namespaced_custom_object,
5376 group="gateway.networking.k8s.io",
5377 version="v1",
5378 namespace=namespace,
5379 plural="httproutes",
5380 name=route_name,
5381 _request_timeout=self._k8s_timeout,
5382 ),
5383 read_call=partial(
5384 custom_objects.get_namespaced_custom_object,
5385 group="gateway.networking.k8s.io",
5386 version="v1",
5387 namespace=namespace,
5388 plural="httproutes",
5389 name=route_name,
5390 _request_timeout=self._k8s_timeout,
5391 ),
5392 patch_metadata=partial(
5393 custom_objects.patch_namespaced_custom_object,
5394 group="gateway.networking.k8s.io",
5395 version="v1",
5396 namespace=namespace,
5397 plural="httproutes",
5398 name=route_name,
5399 _request_timeout=self._k8s_timeout,
5400 ),
5401 pending=pending,
5402 errors=errors,
5403 )
5405 resources_found |= self._observe_generated_children(
5406 name,
5407 namespace,
5408 inventory,
5409 pending,
5410 errors,
5411 )
5413 # A user-named Secret is external and survives. The conventional name
5414 # is auto-managed only when both labels and immutable provenance match;
5415 # an ambiguous same-name Secret blocks terminal cleanup.
5416 mooncake = spec.get("mooncake") if isinstance(spec, dict) else None
5417 proxy = mooncake.get("proxy") if isinstance(mooncake, dict) else None
5418 named_secret = proxy.get("admin_api_key_secret") if isinstance(proxy, dict) else None
5419 generated_secret = inventory.generated_admin_secret
5420 if named_secret != generated_secret:
5421 try:
5422 secret = self.core_v1.read_namespaced_secret(
5423 generated_secret,
5424 namespace,
5425 _request_timeout=self._k8s_timeout,
5426 )
5427 except ApiException as error:
5428 if error.status != 404:
5429 errors.append(self._cleanup_error("read", "secret", generated_secret, error))
5430 except Exception as error:
5431 errors.append(self._cleanup_error("read", "secret", generated_secret, error))
5432 else:
5433 resources_found = True
5434 if isinstance(expected_lifecycle_id, str) and expected_lifecycle_id:
5435 try:
5436 owned_secret = self._require_owned_admin_secret(
5437 secret,
5438 generated_secret,
5439 namespace,
5440 expected_lifecycle_id,
5441 )
5442 except AdminApiKeySecretError:
5443 errors.append(
5444 f"ambiguous secret {generated_secret} is not owned by lifecycle "
5445 f"{expected_lifecycle_id}"
5446 )
5447 else:
5448 resources_found |= self._delete_and_confirm(
5449 kind="secret",
5450 resource_name=generated_secret,
5451 delete_call=partial(
5452 self.core_v1.delete_namespaced_secret,
5453 generated_secret,
5454 namespace,
5455 _request_timeout=self._k8s_timeout,
5456 ),
5457 read_call=partial(
5458 self.core_v1.read_namespaced_secret,
5459 generated_secret,
5460 namespace,
5461 _request_timeout=self._k8s_timeout,
5462 ),
5463 patch_metadata=partial(
5464 self.core_v1.patch_namespaced_secret,
5465 generated_secret,
5466 namespace,
5467 _request_timeout=self._k8s_timeout,
5468 ),
5469 pending=pending,
5470 errors=errors,
5471 observed_resource=owned_secret,
5472 )
5473 else:
5474 errors.append(
5475 f"ambiguous secret {generated_secret} is not owned by lifecycle unknown"
5476 )
5478 return ResourceCleanupResult(
5479 pending=tuple(sorted(set(pending))),
5480 errors=tuple(errors),
5481 resources_found=resources_found,
5482 )
5484 def _build_hpa_metrics(self, metrics_config: list[dict[str, Any]]) -> list[Any]:
5485 """Translate a metrics config list into autoscaler metric specs.
5487 Each entry names a resource (``cpu`` or ``memory``) and a target
5488 average utilization. Unrecognized entries are skipped, and when nothing
5489 recognizable remains the autoscaler falls back to scaling on CPU at 70%
5490 so a Deployment is never left without a scaling signal.
5491 """
5492 hpa_metrics = []
5493 for m in metrics_config:
5494 metric_type = m.get("type", "cpu")
5495 target_value = m.get("target", 70)
5497 if metric_type == "cpu":
5498 hpa_metrics.append(
5499 client.V2MetricSpec(
5500 type="Resource",
5501 resource=client.V2ResourceMetricSource(
5502 name="cpu",
5503 target=client.V2MetricTarget(
5504 type="Utilization",
5505 average_utilization=target_value,
5506 ),
5507 ),
5508 )
5509 )
5510 elif metric_type == "memory":
5511 hpa_metrics.append(
5512 client.V2MetricSpec(
5513 type="Resource",
5514 resource=client.V2ResourceMetricSource(
5515 name="memory",
5516 target=client.V2MetricTarget(
5517 type="Utilization",
5518 average_utilization=target_value,
5519 ),
5520 ),
5521 )
5522 )
5524 if not hpa_metrics:
5525 # Default to CPU if no recognized metrics
5526 hpa_metrics.append(
5527 client.V2MetricSpec(
5528 type="Resource",
5529 resource=client.V2ResourceMetricSource(
5530 name="cpu",
5531 target=client.V2MetricTarget(
5532 type="Utilization",
5533 average_utilization=70,
5534 ),
5535 ),
5536 )
5537 )
5539 return hpa_metrics
5541 @staticmethod
5542 def _metrics_require_keda(metrics_config: list[dict[str, Any]]) -> bool:
5543 """Return True when any metric can only be scaled via KEDA/CloudWatch.
5545 GPU metrics are not Kubernetes Resource metrics, so a native HPA cannot
5546 consume them. Their presence forces the whole autoscaler onto the KEDA
5547 ScaledObject path, where cpu/memory targets become native KEDA triggers
5548 alongside the aws-cloudwatch GPU trigger.
5549 """
5550 return any(m.get("type") in _CLOUDWATCH_METRIC_BY_TYPE for m in metrics_config)
5552 def _build_keda_triggers(
5553 self,
5554 metrics_config: list[dict[str, Any]],
5555 target_name: str,
5556 namespace: str,
5557 ) -> list[dict[str, Any]]:
5558 """Translate a metrics config list into KEDA ScaledObject triggers.
5560 ``cpu`` and ``memory`` map to KEDA's native resource triggers (the same
5561 utilization signal a plain HPA would use). ``gpu``/``gpu_memory`` map to
5562 an ``aws-cloudwatch`` trigger reading the matching ContainerInsights
5563 metric for this Deployment, identified by the
5564 ClusterName/Namespace/PodName dimension triple. Unrecognized entries are
5565 skipped; when nothing recognizable remains the autoscaler falls back to
5566 CPU at 70% so a Deployment is never left without a scaling signal.
5567 """
5568 triggers: list[dict[str, Any]] = []
5569 for m in metrics_config:
5570 metric_type = m.get("type", "cpu")
5571 target_value = m.get("target", 70)
5573 if metric_type in ("cpu", "memory"):
5574 triggers.append(
5575 {
5576 "type": metric_type,
5577 "metricType": "Utilization",
5578 "metadata": {"value": str(target_value)},
5579 }
5580 )
5581 elif metric_type in _CLOUDWATCH_METRIC_BY_TYPE:
5582 triggers.append(
5583 {
5584 "type": "aws-cloudwatch",
5585 "metadata": {
5586 "namespace": GPU_METRIC_NAMESPACE,
5587 "metricName": _CLOUDWATCH_METRIC_BY_TYPE[metric_type],
5588 "dimensionName": "ClusterName;Namespace;PodName",
5589 "dimensionValue": f"{self.cluster_id};{namespace};{target_name}",
5590 "targetMetricValue": str(target_value),
5591 "minMetricValue": "0",
5592 "metricStat": "Average",
5593 "awsRegion": self.region,
5594 "identityOwner": "operator",
5595 },
5596 }
5597 )
5599 if not triggers:
5600 triggers.append(
5601 {
5602 "type": "cpu",
5603 "metricType": "Utilization",
5604 "metadata": {"value": "70"},
5605 }
5606 )
5608 return triggers
5610 def _apply_scaled_object(
5611 self,
5612 name: str,
5613 namespace: str,
5614 target_name: str,
5615 min_replicas: int,
5616 max_replicas: int,
5617 metrics_config: list[dict[str, Any]],
5618 ) -> None:
5619 """Create or patch a KEDA ScaledObject targeting one Deployment.
5621 Used whenever the metric set includes a GPU signal (see
5622 :meth:`_metrics_require_keda`). KEDA owns the backing HPA and reads GPU
5623 utilization from CloudWatch via the keda-operator's IRSA role, scaling
5624 ``target_name`` between ``min_replicas`` and ``max_replicas``. An
5625 already-present ScaledObject of the same name is merge-patched rather
5626 than duplicated.
5627 """
5628 body: dict[str, Any] = {
5629 "apiVersion": f"{KEDA_API_GROUP}/{KEDA_API_VERSION}",
5630 "kind": "ScaledObject",
5631 "metadata": {
5632 "name": name,
5633 "namespace": namespace,
5634 "labels": {
5635 "app": name,
5636 "project": "gco",
5637 "gco.io/type": "inference",
5638 },
5639 "annotations": self._provenance_annotations() or {},
5640 },
5641 "spec": {
5642 "scaleTargetRef": {"name": target_name},
5643 "minReplicaCount": min_replicas,
5644 "maxReplicaCount": max_replicas,
5645 "triggers": self._build_keda_triggers(metrics_config, target_name, namespace),
5646 },
5647 }
5649 custom = client.CustomObjectsApi()
5650 self._assert_mutation_authority()
5651 try:
5652 custom.create_namespaced_custom_object(
5653 group=KEDA_API_GROUP,
5654 version=KEDA_API_VERSION,
5655 namespace=namespace,
5656 plural=KEDA_SCALEDOBJECT_PLURAL,
5657 body=body,
5658 _request_timeout=self._k8s_timeout,
5659 )
5660 self._confirm_created_resource(
5661 kind="scaledobject",
5662 resource_name=name,
5663 read_resource=partial(
5664 custom.get_namespaced_custom_object,
5665 group=KEDA_API_GROUP,
5666 version=KEDA_API_VERSION,
5667 namespace=namespace,
5668 plural=KEDA_SCALEDOBJECT_PLURAL,
5669 name=name,
5670 _request_timeout=self._k8s_timeout,
5671 ),
5672 delete_resource=partial(
5673 custom.delete_namespaced_custom_object,
5674 group=KEDA_API_GROUP,
5675 version=KEDA_API_VERSION,
5676 namespace=namespace,
5677 plural=KEDA_SCALEDOBJECT_PLURAL,
5678 name=name,
5679 _request_timeout=self._k8s_timeout,
5680 ),
5681 )
5682 logger.info(
5683 "Created KEDA ScaledObject %s targeting %s (min=%d, max=%d)",
5684 name,
5685 target_name,
5686 min_replicas,
5687 max_replicas,
5688 )
5689 except ApiException as error:
5690 if error.status != 409:
5691 raise
5692 existing = custom.get_namespaced_custom_object(
5693 group=KEDA_API_GROUP,
5694 version=KEDA_API_VERSION,
5695 namespace=namespace,
5696 plural=KEDA_SCALEDOBJECT_PLURAL,
5697 name=name,
5698 _request_timeout=self._k8s_timeout,
5699 )
5700 existing = self._authorize_resource(
5701 existing,
5702 kind="scaledobject",
5703 resource_name=name,
5704 patch_metadata=partial(
5705 custom.patch_namespaced_custom_object,
5706 group=KEDA_API_GROUP,
5707 version=KEDA_API_VERSION,
5708 namespace=namespace,
5709 plural=KEDA_SCALEDOBJECT_PLURAL,
5710 name=name,
5711 _request_timeout=self._k8s_timeout,
5712 ),
5713 read_resource=lambda: custom.get_namespaced_custom_object(
5714 group=KEDA_API_GROUP,
5715 version=KEDA_API_VERSION,
5716 namespace=namespace,
5717 plural=KEDA_SCALEDOBJECT_PLURAL,
5718 name=name,
5719 _request_timeout=self._k8s_timeout,
5720 ),
5721 delete_resource=partial(
5722 custom.delete_namespaced_custom_object,
5723 group=KEDA_API_GROUP,
5724 version=KEDA_API_VERSION,
5725 namespace=namespace,
5726 plural=KEDA_SCALEDOBJECT_PLURAL,
5727 name=name,
5728 _request_timeout=self._k8s_timeout,
5729 ),
5730 )
5731 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
5732 body["metadata"]["resourceVersion"] = resource_version
5733 self._assert_mutation_authority()
5734 custom.patch_namespaced_custom_object(
5735 group=KEDA_API_GROUP,
5736 version=KEDA_API_VERSION,
5737 namespace=namespace,
5738 plural=KEDA_SCALEDOBJECT_PLURAL,
5739 name=name,
5740 body=body,
5741 _request_timeout=self._k8s_timeout,
5742 )
5743 logger.info("Updated KEDA ScaledObject %s", name)
5745 def _apply_hpa(
5746 self,
5747 hpa_name: str,
5748 namespace: str,
5749 target_name: str,
5750 min_replicas: int,
5751 max_replicas: int,
5752 metrics_config: list[dict[str, Any]],
5753 ) -> None:
5754 """Create or patch a single autoscaler targeting one Deployment.
5756 Builds a V2 autoscaler that scales ``target_name`` between
5757 ``min_replicas`` and ``max_replicas`` on the given metrics, then creates
5758 it. An already-present autoscaler of the same name is patched in place
5759 rather than duplicated. When the metric set includes a GPU signal the
5760 autoscaler is materialized as a KEDA ScaledObject instead (native HPA
5761 Resource metrics cannot read GPU utilization).
5762 """
5763 if self._metrics_require_keda(metrics_config):
5764 self._apply_scaled_object(
5765 name=hpa_name,
5766 namespace=namespace,
5767 target_name=target_name,
5768 min_replicas=min_replicas,
5769 max_replicas=max_replicas,
5770 metrics_config=metrics_config,
5771 )
5772 return
5774 hpa = client.V2HorizontalPodAutoscaler(
5775 metadata=client.V1ObjectMeta(
5776 name=hpa_name,
5777 namespace=namespace,
5778 labels={
5779 "app": hpa_name,
5780 "project": "gco",
5781 "gco.io/type": "inference",
5782 },
5783 annotations=self._provenance_annotations(),
5784 ),
5785 spec=client.V2HorizontalPodAutoscalerSpec(
5786 scale_target_ref=client.V2CrossVersionObjectReference(
5787 api_version="apps/v1",
5788 kind="Deployment",
5789 name=target_name,
5790 ),
5791 min_replicas=min_replicas,
5792 max_replicas=max_replicas,
5793 metrics=self._build_hpa_metrics(metrics_config),
5794 ),
5795 )
5797 autoscaling_v2 = client.AutoscalingV2Api()
5798 self._assert_mutation_authority()
5799 try:
5800 autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(namespace, hpa)
5801 self._confirm_created_resource(
5802 kind="hpa",
5803 resource_name=hpa_name,
5804 read_resource=partial(
5805 autoscaling_v2.read_namespaced_horizontal_pod_autoscaler,
5806 hpa_name,
5807 namespace,
5808 _request_timeout=self._k8s_timeout,
5809 ),
5810 delete_resource=partial(
5811 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
5812 hpa_name,
5813 namespace,
5814 _request_timeout=self._k8s_timeout,
5815 ),
5816 )
5817 logger.info(
5818 "Created HPA %s targeting %s (min=%d, max=%d)",
5819 hpa_name,
5820 target_name,
5821 min_replicas,
5822 max_replicas,
5823 )
5824 except ApiException as error:
5825 if error.status != 409:
5826 raise
5827 existing = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
5828 hpa_name, namespace, _request_timeout=self._k8s_timeout
5829 )
5830 existing = self._authorize_resource(
5831 existing,
5832 kind="hpa",
5833 resource_name=hpa_name,
5834 patch_metadata=partial(
5835 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
5836 hpa_name,
5837 namespace,
5838 _request_timeout=self._k8s_timeout,
5839 ),
5840 read_resource=lambda: autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
5841 hpa_name, namespace, _request_timeout=self._k8s_timeout
5842 ),
5843 delete_resource=partial(
5844 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
5845 hpa_name,
5846 namespace,
5847 _request_timeout=self._k8s_timeout,
5848 ),
5849 )
5850 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
5851 hpa.metadata.resource_version = resource_version
5852 self._assert_mutation_authority()
5853 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler(
5854 hpa_name,
5855 namespace,
5856 hpa,
5857 _request_timeout=self._k8s_timeout,
5858 )
5859 logger.info("Updated HPA %s", hpa_name)
5861 def _verify_hpa_owner(
5862 self,
5863 hpa_name: str,
5864 namespace: str,
5865 target_name: str,
5866 ) -> ResourceCleanupResult:
5867 """Verify that the exact native/KEDA HPA owns the expected Deployment."""
5868 autoscaling_v2 = client.AutoscalingV2Api()
5869 try:
5870 hpa = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
5871 hpa_name,
5872 namespace,
5873 )
5874 except ApiException as error:
5875 if error.status == 404:
5876 return ResourceCleanupResult(pending=(f"hpa/{hpa_name}",))
5877 return ResourceCleanupResult(
5878 errors=(self._cleanup_error("read", "hpa", hpa_name, error),)
5879 )
5880 except Exception as error:
5881 return ResourceCleanupResult(
5882 errors=(self._cleanup_error("read", "hpa", hpa_name, error),)
5883 )
5884 try:
5885 hpa = self._authorize_resource(
5886 hpa,
5887 kind="hpa",
5888 resource_name=hpa_name,
5889 patch_metadata=partial(
5890 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
5891 hpa_name,
5892 namespace,
5893 _request_timeout=self._k8s_timeout,
5894 ),
5895 read_resource=lambda: autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
5896 hpa_name, namespace, _request_timeout=self._k8s_timeout
5897 ),
5898 delete_resource=partial(
5899 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
5900 hpa_name,
5901 namespace,
5902 _request_timeout=self._k8s_timeout,
5903 ),
5904 )
5905 except ReconcileFencedError:
5906 raise
5907 hpa_spec = getattr(hpa, "spec", None)
5908 target_ref = getattr(hpa_spec, "scale_target_ref", None)
5909 observed_api_version = getattr(target_ref, "api_version", None)
5910 observed_kind = getattr(target_ref, "kind", None)
5911 observed_target = getattr(target_ref, "name", None)
5912 if (
5913 observed_api_version != "apps/v1"
5914 or observed_kind != "Deployment"
5915 or observed_target != target_name
5916 ):
5917 return ResourceCleanupResult(
5918 pending=(f"hpa/{hpa_name}",),
5919 resources_found=True,
5920 )
5921 return ResourceCleanupResult(resources_found=True)
5923 def _reconcile_classic_autoscaler(
5924 self,
5925 name: str,
5926 namespace: str,
5927 spec: dict[str, Any],
5928 *,
5929 apply_desired: bool = True,
5930 ) -> ResourceCleanupResult:
5931 """Converge a classic endpoint to exactly one autoscaler owner.
5933 A stopped target is first restored to a positive minimum, then the
5934 desired native/KEDA owner is applied and read back before reconciliation
5935 may proceed. Obsolete owners are always confirmed absent first.
5936 """
5937 autoscaling = spec.get("autoscaling", {})
5938 metrics_config = autoscaling.get("metrics", [{"type": "cpu", "target": 70}])
5939 keda = self._metrics_require_keda(metrics_config)
5940 if keda:
5941 cleanup = self._delete_hpas((name,), namespace)
5942 else:
5943 cleanup = self._merge_cleanup_results(
5944 self._delete_scaled_objects((name,), namespace),
5945 self._delete_hpas((f"keda-hpa-{name}",), namespace),
5946 )
5948 if cleanup.complete and apply_desired:
5949 target = self._get_deployment(name, namespace)
5950 current = getattr(getattr(target, "spec", None), "replicas", None)
5951 if target is not None and current == 0:
5952 minimum = max(1, int(autoscaling.get("min_replicas", 1)))
5953 self._scale_deployment(name, namespace, minimum)
5954 self._create_or_update_hpa(name, namespace, spec)
5955 desired_hpa = f"keda-hpa-{name}" if keda else name
5956 return self._merge_cleanup_results(
5957 cleanup,
5958 self._verify_hpa_owner(desired_hpa, namespace, name),
5959 )
5960 return cleanup
5962 def _create_or_update_hpa(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
5963 """Create or update a Horizontal Pod Autoscaler for an inference endpoint."""
5964 autoscaling_config = spec.get("autoscaling", {})
5965 if not autoscaling_config.get("enabled"):
5966 return
5968 min_replicas = autoscaling_config.get("min_replicas", 1)
5969 max_replicas = autoscaling_config.get("max_replicas", 10)
5970 metrics_config = autoscaling_config.get("metrics", [{"type": "cpu", "target": 70}])
5972 self._apply_hpa(
5973 hpa_name=name,
5974 namespace=namespace,
5975 target_name=name,
5976 min_replicas=min_replicas,
5977 max_replicas=max_replicas,
5978 metrics_config=metrics_config,
5979 )
5981 @staticmethod
5982 def _role_autoscaling_config(
5983 spec: dict[str, Any],
5984 role: str,
5985 ) -> dict[str, Any] | None:
5986 """Return an enabled role config, or ``None`` for static ownership."""
5987 mooncake = spec.get("mooncake")
5988 if not isinstance(mooncake, dict):
5989 return None
5990 autoscaling = mooncake.get("autoscaling")
5991 if not isinstance(autoscaling, dict) or autoscaling.get("enabled") is not True:
5992 return None
5993 role_config = autoscaling.get(role)
5994 return role_config if isinstance(role_config, dict) else None
5996 def _reconcile_role_autoscaler(
5997 self,
5998 name: str,
5999 namespace: str,
6000 spec: dict[str, Any],
6001 role: str,
6002 *,
6003 apply_desired: bool = True,
6004 ) -> ResourceCleanupResult:
6005 """Converge one Mooncake role to exactly one or zero replica owners."""
6006 target_name = f"{name}-{role}"
6007 role_config = self._role_autoscaling_config(spec, role)
6008 if role_config is None:
6009 return self._delete_autoscalers(
6010 (target_name,),
6011 (target_name, f"keda-hpa-{target_name}"),
6012 namespace,
6013 )
6015 metrics_config = role_config.get("metrics", [{"type": "cpu", "target": 70}])
6016 if self._metrics_require_keda(metrics_config):
6017 cleanup = self._delete_hpas((target_name,), namespace)
6018 else:
6019 cleanup = self._merge_cleanup_results(
6020 self._delete_scaled_objects((target_name,), namespace),
6021 self._delete_hpas((f"keda-hpa-{target_name}",), namespace),
6022 )
6023 if cleanup.complete and apply_desired:
6024 self._create_role_hpa(name, namespace, spec, role)
6025 return cleanup
6027 def _create_role_hpa(self, name: str, ns: str, spec: dict[str, Any], role: str) -> None:
6028 """Create or update one autoscaler for a single Mooncake role.
6030 When the endpoint's ``mooncake.autoscaling`` block is enabled and
6031 carries a config for this role, this materializes exactly one autoscaler
6032 named ``{name}-{role}`` that scales the matching ``{name}-{role}``
6033 Deployment between the role's ``min_replicas`` and ``max_replicas``. The
6034 role Deployment itself is already materialized at ``min_replicas`` by
6035 :meth:`_replica_count_for_role`, so the autoscaler owns the count from
6036 that lower bound. When autoscaling is absent or disabled, or when the
6037 role carries no config, no autoscaler is created and the role's replicas
6038 stay at their topology value.
6040 Args:
6041 name: The endpoint name.
6042 ns: The namespace the role Deployment lives in.
6043 spec: The endpoint spec; ``spec["mooncake"]["autoscaling"]`` drives
6044 the bounds and metrics.
6045 role: One of ``"prefill"`` or ``"decode"``.
6046 """
6047 role_cfg = self._role_autoscaling_config(spec, role)
6048 if role_cfg is None:
6049 return
6051 min_replicas = role_cfg.get("min_replicas", 1)
6052 max_replicas = role_cfg.get("max_replicas", 10)
6053 metrics_config = role_cfg.get("metrics", [{"type": "cpu", "target": 70}])
6055 target_name = f"{name}-{role}"
6056 self._apply_hpa(
6057 hpa_name=target_name,
6058 namespace=ns,
6059 target_name=target_name,
6060 min_replicas=min_replicas,
6061 max_replicas=max_replicas,
6062 metrics_config=metrics_config,
6063 )
6065 # ------------------------------------------------------------------
6066 # Metrics
6067 # ------------------------------------------------------------------
6069 def get_metrics(self) -> dict[str, Any]:
6070 return {
6071 "cluster_id": self.cluster_id,
6072 "region": self.region,
6073 "running": self._running,
6074 "reconcile_count": self._reconcile_count,
6075 "errors_count": self._errors_count,
6076 "seconds_since_last_pass": time.monotonic() - self._last_loop_completed_at,
6077 }
6080def create_inference_monitor_from_env() -> InferenceMonitor:
6081 """Create an InferenceMonitor from environment variables."""
6082 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
6083 region = os.getenv("REGION", "unknown-region")
6084 namespace = os.getenv("INFERENCE_NAMESPACE", "gco-inference")
6085 interval = int(os.getenv("RECONCILE_INTERVAL_SECONDS", "15"))
6087 # Enable structured JSON logging for CloudWatch Insights
6088 configure_structured_logging(
6089 service_name="inference-monitor",
6090 cluster_id=cluster_id,
6091 region=region,
6092 )
6094 store = InferenceEndpointStore() # Uses DYNAMODB_REGION env var, falls back to REGION
6096 return InferenceMonitor(
6097 cluster_id=cluster_id,
6098 region=region,
6099 store=store,
6100 namespace=namespace,
6101 reconcile_interval=interval,
6102 )
6105async def main() -> None:
6106 """Entry point for the inference monitor."""
6107 monitor = create_inference_monitor_from_env()
6108 logger.info("Inference monitor initialized: %s", monitor.get_metrics())
6110 # Expose Prometheus metrics on a dedicated port for the in-cluster
6111 # observability scrape. A scrape-time collector reflects the monitor's live
6112 # counters (reconcile_count, errors_count, running), so no push from the
6113 # reconcile loop is needed.
6114 from gco.services.service_metrics import start_metrics_server
6116 metrics_port = int(os.getenv("METRICS_PORT", "9090"))
6117 start_metrics_server(metrics_port, "inference-monitor", monitor.get_metrics)
6119 # Kubernetes stops pods with SIGTERM. This process is PID 1 in its
6120 # container, and PID 1 receives no kernel-default signal handling — so
6121 # without an explicit handler SIGTERM was silently ignored, every pod
6122 # rotation burned the full terminationGracePeriodSeconds, and the kubelet
6123 # SIGKILLed the monitor mid-reconcile (exit 137). The handler flips the
6124 # same stop flag the reconcile loop already honors, so shutdown waits for
6125 # the in-flight cycle and exits 0 within one reconcile interval.
6126 shutdown_requested = asyncio.Event()
6127 loop = asyncio.get_running_loop()
6129 def _handle_sigterm() -> None:
6130 logger.info("SIGTERM received; stopping inference monitor after current cycle")
6131 shutdown_requested.set()
6132 monitor.stop()
6134 try:
6135 loop.add_signal_handler(signal.SIGTERM, _handle_sigterm)
6136 except NotImplementedError:
6137 # Non-POSIX event loops (Windows dev environments running the unit
6138 # suite) don't support loop signal handlers; in the Linux container
6139 # this always succeeds.
6140 logger.debug("Event loop does not support signal handlers; skipping SIGTERM hook")
6142 try:
6143 while not shutdown_requested.is_set():
6144 try:
6145 await monitor.start()
6146 except KeyboardInterrupt:
6147 logger.info("Shutting down inference monitor")
6148 monitor.stop()
6149 break
6150 except Exception as e:
6151 logger.error("Monitor crashed, restarting in 10s: %s", e, exc_info=True)
6152 monitor.stop()
6153 monitor._running = False
6154 await asyncio.sleep(10)
6155 logger.info("Inference monitor exited cleanly")
6156 finally:
6157 with contextlib.suppress(NotImplementedError, ValueError):
6158 loop.remove_signal_handler(signal.SIGTERM)
6161if __name__ == "__main__":
6162 asyncio.run(main())