Coverage for gco / services / inference_monitor.py: 100.00%
2225 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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- Terminates in-cluster TLS in an ``endpoint-tls-proxy`` sidecar in every model
13 pod; model Services publish only that sidecar's port 8443
14- Leaves public routing on the shared ``gco-system/gco-gateway`` HTTPRoute:
15 ``/inference`` -> ``gco-system/inference-proxy``
16- Removes legacy endpoint-specific Ingresses so upgrades cannot retain a bypass
17- Updates existing deployments when spec changes
18- Scales deployments up/down
19- Tears down resources when endpoints are deleted
20- Reports per-region status back to DynamoDB
22Environment Variables:
23 CLUSTER_NAME: Name of the EKS cluster
24 REGION: AWS region this monitor runs in
25 INFERENCE_ENDPOINTS_TABLE_NAME: DynamoDB table name
26 RECONCILE_INTERVAL_SECONDS: Seconds between reconciliation loops (default: 15)
27 INFERENCE_NAMESPACE: Namespace for inference workloads (default: gco-inference)
28 ENDPOINT_TLS_PROXY_IMAGE: Image for the model pods' TLS sidecar (default: the
29 pinned multi-arch ``ENDPOINT_TLS_PROXY_IMAGE`` constant)
30 METRICS_HOST / METRICS_PORT: Prometheus listener (default: 0.0.0.0:9090)
31"""
33import asyncio
34import base64
35import contextlib
36import hashlib
37import json
38import logging
39import os
40import re
41import secrets
42import signal
43import threading
44import time
45from collections.abc import Callable, Iterator, Mapping
46from dataclasses import dataclass, field
47from datetime import UTC, datetime
48from functools import partial
49from pathlib import Path
50from typing import Any, NoReturn, TypedDict
51from urllib.parse import urlsplit
53from kubernetes import client, config
54from kubernetes.client.models import V1Deployment
55from kubernetes.client.rest import ApiException
57from gco.models.inference_models import (
58 GPU_NAME_NODE_LABEL,
59 INFERENCE_PROBE_TIMEOUT_SECONDS,
60 INSTANCE_TYPE_NODE_LABEL,
61 SGLANG_UNSUPPORTED_GPU_NAMES,
62)
63from gco.services.inference_store import InferenceEndpointStore
64from gco.services.structured_logging import configure_structured_logging
66# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
67# Generated at (UTC): 2026-10-20T06:07:53Z
68# Generated from Git commit: a79cb382e149c05b653fcf1d50a9793f0c861209
69# Flowchart(s) generated from this file:
70# * ``InferenceMonitor._reconcile_endpoint_authorized`` -> ``diagrams/code_diagrams/gco/services/inference_monitor.InferenceMonitor__reconcile_endpoint_authorized.html``
71# (PNG: ``diagrams/code_diagrams/gco/services/inference_monitor.InferenceMonitor__reconcile_endpoint_authorized.png``)
72# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
73# <pyflowchart-code-diagram> END
76logging.basicConfig(
77 level=logging.INFO,
78 format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
79)
80logger = logging.getLogger(__name__)
83class NetworkPolicyApplyError(Exception):
84 """An intra-namespace allow rule could not be applied.
86 Raised when materialization-time enforcement fails to create or verify one
87 of the allow rules that disaggregated inference depends on. The default-deny
88 posture is left intact — only the widening allow rule failed — and ``rule``
89 names the offending NetworkPolicy so callers can surface exactly which rule
90 could not be applied.
91 """
93 def __init__(self, rule: str, reason: str):
94 self.rule = rule
95 self.reason = reason
96 super().__init__(f"Network policy {rule!r} could not be applied: {reason}")
99class AdminApiKeySecretError(Exception):
100 """A user-named proxy admin API key Secret is missing or empty.
102 Raised before the prefill-decode proxy is materialized when a Secret named
103 by ``proxy.admin_api_key_secret`` is absent or carries no usable
104 ``ADMIN_API_KEY`` value. An endpoint that names no Secret takes the separate
105 auto-managed path and receives a generated ``{name}-admin`` Secret. The
106 proxy never starts without a usable key, so no proxy Deployment or Service
107 is created on this error. ``secret`` records the rejected Secret name.
108 """
110 def __init__(self, secret: str | None, reason: str):
111 self.secret = secret
112 self.reason = reason
113 named = repr(secret) if secret else "<unnamed>"
114 super().__init__(f"Admin API key Secret {named} is unusable: {reason}")
117# Official AWS CLI v2 multi-architecture image. Keep the readable release tag
118# and immutable manifest-list digest together: both amd64 and arm64 inference
119# nodes resolve through this single verified index.
120AWS_CLI_IMAGE = (
121 "public.ecr.aws/aws-cli/aws-cli:2.37.10@"
122 "sha256:3dacc5db57c923c4223e949795f538ecf1f2212b2b7d5a028b47b97f91564c0d"
123)
125# Valid TCP port boundaries for KV-transfer bootstrap ports.
126MIN_BOOTSTRAP_PORT = 1024
127MAX_BOOTSTRAP_PORT = 65535
129# vLLM kv_role for each worker role: prefill produces KV, decode consumes it,
130# and a single-instance store node both produces and consumes.
131_KV_ROLE_BY_WORKER_ROLE = {
132 "prefill": "kv_producer",
133 "decode": "kv_consumer",
134 "single": "kv_both",
135}
137# The worker roles each mooncake mode supports. Disaggregated and both split
138# work across prefill/decode; store runs a single kv_both instance.
139_WORKER_ROLES_BY_MODE = {
140 "disaggregated": {"prefill", "decode"},
141 "store": {"single"},
142 "both": {"prefill", "decode"},
143}
145# The EFA RDMA fabric is advertised as a Kubernetes extended resource, gated by
146# a node taint, and selected through a node label. KV cache transfer over
147# RoCE only runs on nodes that carry all three.
148EFA_RESOURCE_NAME = "vpc.amazonaws.com/efa"
149EFA_NODE_SELECTOR_KEY = "efa"
150EFA_NODE_SELECTOR_VALUE = "true"
152# Mooncake KV-transfer role pods are pinned to a dedicated EFA NodePool
153# (mooncake-efa-pool, manifest 46-nodepool-mooncake-efa.yaml) that only offers
154# instance families with >=80GB of GPU memory and FP8-capable Hopper/Blackwell
155# GPUs. The shared training EFA pool (43-nodepool-efa.yaml) also offers p4d
156# (A100 40GB, Ampere, no FP8), which is too small for many disaggregated/store
157# models and can be selected by Karpenter whenever a pod asks only for efa=true.
158# Selecting this extra label keeps role pods off p4d without disturbing the
159# training pool. The value must match the label on the dedicated NodePool.
160MOONCAKE_EFA_NODE_SELECTOR_KEY = "mooncake-efa"
161MOONCAKE_EFA_NODE_SELECTOR_VALUE = "true"
163# The shared per-region Mooncake master exposes its RPC service and the
164# built-in HTTP metadata server on these fixed ports.
165MOONCAKE_MASTER_RPC_PORT = 50051
166MOONCAKE_METADATA_PORT = 8080
167MOONCAKE_MASTER_SERVICE = "mooncake-master"
169# GPU utilization is not a Kubernetes Resource metric, so a native
170# HorizontalPodAutoscaler cannot scale on it (Resource metrics are limited to
171# cpu and memory). The cluster's amazon-cloudwatch-observability agent publishes
172# per-pod GPU utilization to CloudWatch ContainerInsights, so any autoscaler
173# that requests a GPU metric is materialized as a KEDA ScaledObject with an
174# aws-cloudwatch trigger instead. KEDA generates the backing HPA under the hood,
175# and cpu/memory targets ride along as native cpu/memory triggers on the same
176# ScaledObject. KEDA is a mandatory cluster component, so this path is always
177# available.
178KEDA_API_GROUP = "keda.sh"
179KEDA_API_VERSION = "v1alpha1"
180KEDA_SCALEDOBJECT_PLURAL = "scaledobjects"
182# Metric types that can only be served via CloudWatch (KEDA), keyed to the
183# ContainerInsights metric the aws-cloudwatch trigger reads. PodName in the
184# ContainerInsights dimension set is the workload (Deployment) name, so the
185# dimension triple ClusterName/Namespace/PodName yields the average across a
186# Deployment's pods — exactly the signal autoscaling needs.
187GPU_METRIC_NAMESPACE = "ContainerInsights"
188_CLOUDWATCH_METRIC_BY_TYPE = {
189 "gpu": "pod_gpu_utilization",
190 "gpu_memory": "pod_gpu_memory_utilization",
191}
193# Default base port for the KV-transfer bootstrap handshake (VLLM_MOONCAKE_
194# BOOTSTRAP_PORT) and the span of per-worker ports derived from it. vLLM assigns
195# each worker base_port + dp_rank * tp_size + tp_rank, so the intra-namespace
196# allow rule opens a contiguous window starting at the base port. A spec may
197# override the base via mooncake.transfer.bootstrap_base_port.
198MOONCAKE_BOOTSTRAP_BASE_PORT = 8998
199MOONCAKE_BOOTSTRAP_PORT_SPAN = 100
201# Environment variable through which each role pod receives the KV-transfer
202# bootstrap base port. vLLM derives per-worker ports (base + dp_rank * tp_size +
203# tp_rank) from it, so prefill and decode agree on the handshake ports.
204VLLM_MOONCAKE_BOOTSTRAP_PORT_ENV = "VLLM_MOONCAKE_BOOTSTRAP_PORT"
206# Each role pod reads the shared transport settings (metadata-server address,
207# protocol, device) from the rendered mooncake.json. The per-endpoint
208# ``{name}-mooncake`` ConfigMap is mounted read-only at the directory below, and
209# the connector is pointed at the file through MOONCAKE_CONFIG_PATH.
210MOONCAKE_CONFIG_PATH_ENV = "MOONCAKE_CONFIG_PATH"
211MOONCAKE_CONFIG_MOUNT_DIR = "/etc/mooncake"
212MOONCAKE_CONFIG_FILE_PATH = f"{MOONCAKE_CONFIG_MOUNT_DIR}/mooncake.json"
214# Label selector identifying inference workload pods (prefill/decode/proxy and
215# legacy single-Deployment endpoints). Used as both the target and the peer of
216# the intra-namespace allow rules.
217INFERENCE_POD_SELECTOR = {"gco.io/type": "inference"}
220def _inference_service_selector(app: str, **extra: str) -> dict[str, str]:
221 """The ``spec.selector`` of every ClusterIP Service fronting inference pods.
223 The selector carries ``gco.io/type: inference`` next to the per-endpoint
224 ``app`` label, and that is load-bearing under enforced NetworkPolicies on
225 the VPC CNI: its egress probe sees ClusterIP traffic before kube-proxy's
226 DNAT, so the network policy controller admits a Service's ClusterIP only
227 when the Service's own selector matches the policy peer's ``podSelector``
228 (``gco.io/type: inference`` in ``allow-inference-proxy-to-inference`` and
229 ``allow-inference-internal``). With ``app`` alone the pods were admitted
230 and the Service in front of them was not, and the inference proxy's
231 requests timed out against a healthy endpoint.
232 """
233 return {"app": app, **INFERENCE_POD_SELECTOR, **extra}
236# Names of the intra-namespace allow rules the monitor maintains alongside the
237# default-deny posture in gco-inference. These mirror the manifest names in
238# 03-network-policies.yaml so a failure can point at the same object an operator
239# would inspect with kubectl.
240NETWORK_POLICY_INFERENCE_INTERNAL = "allow-inference-internal"
241NETWORK_POLICY_POD_TO_MASTER = "allow-pod-to-master"
242NETWORK_POLICY_POD_TO_METADATA = "allow-pod-to-metadata"
243NETWORK_POLICY_RDMA_BOOTSTRAP = "allow-rdma-bootstrap"
245# Regional configuration keys through which the in-region deployment supplies
246# the shared master's address. The store cannot be wired without an own-region
247# master address, so a store-bearing endpoint defers configuration when the
248# master address is absent or blank. The metadata server defaults to the master
249# host on the metadata port when not supplied explicitly.
250MOONCAKE_MASTER_ADDRESS_ENV = "MOONCAKE_MASTER_ADDRESS"
251MOONCAKE_METADATA_SERVER_ENV = "MOONCAKE_METADATA_SERVER"
253# Container image for the shared per-region master. Supplied by the in-region
254# deployment so the master tracks the same pinned build the manifests use.
255MOONCAKE_MASTER_IMAGE_ENV = "MOONCAKE_MASTER_IMAGE"
257# Maximum time a store-bearing endpoint keeps deferring role-pod creation while
258# the shared master has not reported a Ready replica. Past this window the
259# monitor keeps deferring and stays in the ``creating`` state, but also surfaces
260# an error so operators can see the master never came up. The master itself is
261# never deleted or modified on account of this timeout.
262MOONCAKE_MASTER_READY_TIMEOUT_SECONDS = 600
264# Object-key prefix under which cold-tier KV objects are written in the
265# general-purpose regional bucket. Mirrors the value the regional stack and the
266# `gco inference populate-kv` upload surface use; kept local so the monitor
267# needs no infrastructure (CDK) imports at runtime.
268MOONCAKE_COLD_TIER_KEY_PREFIX = "mooncake-kv"
270# SSM namespace suffix publishing the always-on general-purpose regional
271# bucket's discovery values for a region. The full namespace is
272# ``/<project_name>/regional-shared-bucket`` (see
273# ``constants.regional_shared_ssm_parameter_prefix``); the monitor builds it at
274# runtime from the injected ``PROJECT_NAME`` env var rather than importing the
275# CDK constant, so it needs no infrastructure imports at runtime. Kept as a
276# suffix constant so the ``/name``, ``/arn``, ``/region`` contract stays in one
277# place. See ``_regional_shared_ssm_parameter_prefix``.
278REGIONAL_SHARED_SSM_PARAMETER_SUFFIX = "regional-shared-bucket"
281def _regional_shared_ssm_parameter_prefix() -> str:
282 """Return this deployment's regional-shared-bucket SSM namespace.
284 Built from the ``PROJECT_NAME`` environment variable (default ``"gco"``)
285 so the monitor reads the same project-scoped path the regional stack
286 writes (``/<project_name>/regional-shared-bucket``). Mirrors
287 ``constants.regional_shared_ssm_parameter_prefix`` without importing CDK.
288 """
289 project_name = os.environ.get("PROJECT_NAME", "gco")
290 return f"/{project_name}/{REGIONAL_SHARED_SSM_PARAMETER_SUFFIX}"
293# Matches an AWS region identifier embedded in an address (host or URI), e.g.
294# ``us-east-1``, ``eu-west-2``, ``ap-southeast-1``, ``us-gov-west-1``. KV
295# transfer over RoCE is intra-region, so any address a topology wires to must
296# resolve to the monitor's own region; an embedded token naming a different
297# region marks the address as out-of-region. An address that carries no token
298# (a bare in-cluster Service name) is region-local by construction.
299_REGION_TOKEN_PATTERN = re.compile(r"\b[a-z]{2}-(?:gov-)?[a-z]+-\d+\b")
301# --- PD proxy behavior -------------------------------------------------------
302#
303# The prefill-decode proxy that fronts a disaggregated endpoint checks whether a
304# prompt's KV blocks already live in the shared store before it sends the prompt
305# to a prefill pod. That check is bounded: it is given this many seconds, and a
306# miss or a check that does not finish in time is treated as "not resident" so
307# the prompt goes to prefill without the request waiting any longer. Holding the
308# bound here keeps the proxy responsive even when the store is slow or
309# unreachable.
310PD_PROXY_RESIDENCY_TIMEOUT_SECONDS = 2
312# Default strategy for spreading requests across the backends of a single role.
313PD_PROXY_DEFAULT_SCHEDULING = "round_robin"
315# Where a residency miss or timed-out lookup is sent. The prompt always goes to
316# a prefill pod in that case; the proxy never stalls the request on the store.
317PD_PROXY_RESIDENCY_MISS_TARGET = "prefill"
319# The residency lookup never blocks the request: a slow or failed store check
320# falls through to prefill rather than holding the client.
321PD_PROXY_RESIDENCY_BLOCKING = "false"
323# Decode-phase requests only ever reach decode pods that report Ready; pods that
324# are still starting are skipped.
325PD_PROXY_DECODE_ROUTING_READY_ONLY = "ready_only"
327# When no decode pod reports Ready, the proxy refuses the request outright
328# instead of streaming a partial generation. The refusal carries a stable
329# status and message so clients can distinguish "no backend yet" from a model
330# error.
331PD_PROXY_NO_DECODE_BACKEND_ACTION_REJECT = "reject"
332PD_PROXY_NO_DECODE_BACKEND_STATUS = "503"
333PD_PROXY_NO_DECODE_BACKEND_MESSAGE = "no available decode backend"
335# Environment variable names the proxy container reads to pick up the behavior
336# above. Surfacing them here keeps the proxy's runtime contract in one place;
337# the reconcile path attaches the values produced by ``build_pd_proxy_config``.
338PD_PROXY_RESIDENCY_TIMEOUT_ENV = "PD_PROXY_RESIDENCY_TIMEOUT_SECONDS"
339PD_PROXY_RESIDENCY_BLOCKING_ENV = "PD_PROXY_RESIDENCY_CHECK_BLOCKING"
340PD_PROXY_RESIDENCY_MISS_TARGET_ENV = "PD_PROXY_RESIDENCY_MISS_TARGET"
341PD_PROXY_DECODE_ROUTING_ENV = "PD_PROXY_DECODE_ROUTING"
342PD_PROXY_NO_DECODE_BACKEND_ACTION_ENV = "PD_PROXY_NO_DECODE_BACKEND_ACTION"
343PD_PROXY_NO_DECODE_BACKEND_STATUS_ENV = "PD_PROXY_NO_DECODE_BACKEND_STATUS"
344PD_PROXY_NO_DECODE_BACKEND_MESSAGE_ENV = "PD_PROXY_NO_DECODE_BACKEND_MESSAGE"
345PD_PROXY_SCHEDULING_ENV = "PD_PROXY_SCHEDULING"
346PD_PROXY_STORE_ADDRESS_ENV = "PD_PROXY_STORE_ADDRESS"
348# Marker label carried by proxy pods so a Service can select the proxy alone,
349# distinct from the prefill/decode role pods (which carry their own role marker).
350PD_PROXY_ROLE_LABEL = "proxy"
352# TCP port the proxy container listens on for authenticated serving requests.
353PD_PROXY_PORT = 8000
355# Environment variable the proxy reads its admin key from, and the data key the
356# backing Kubernetes Secret stores it under. The key value is delivered to the
357# container through a Secret reference at pod start — it is never written to the
358# endpoint spec or passed as a command-line argument.
359PD_PROXY_ADMIN_API_KEY_ENV = "ADMIN_API_KEY"
360ADMIN_API_KEY_SECRET_DATA_KEY = "ADMIN_API_KEY" # nosec B105 # Secret data key name, not the key value
362# The proxy program (gco/services/mooncake_pd_proxy.py) is shipped to the proxy
363# pod as a ConfigMap and run from this mount path. The prefill/decode backend
364# URLs and the listen port are passed to it through these env vars; it routes to
365# the role pods through their in-cluster Services so kube-proxy load-balances
366# across only the Ready endpoints of each role.
367PD_PROXY_SCRIPT_FILENAME = "mooncake_pd_proxy.py"
368PD_PROXY_CONFIG_MOUNT_DIR = "/etc/pd-proxy"
369PD_PROXY_SCRIPT_PATH = f"{PD_PROXY_CONFIG_MOUNT_DIR}/{PD_PROXY_SCRIPT_FILENAME}"
370PD_PROXY_PORT_ENV = "PD_PROXY_PORT"
371PD_PROXY_PREFILL_URL_ENV = "PD_PROXY_PREFILL_URL"
372PD_PROXY_DECODE_URL_ENV = "PD_PROXY_DECODE_URL"
374# The router binds pod loopback, so the only way into it from the pod network
375# is the pod's own TLS sidecar: nothing else in the namespace can reach it in
376# plaintext, and it needs no NetworkPolicy of its own to stay that way.
377PD_PROXY_HOST_ENV = "PD_PROXY_HOST"
378PD_PROXY_BIND_HOST = "127.0.0.1"
380# The router verifies the prefill and decode Services against the GCO internal
381# CA alone. cert-manager writes the issuing CA into every leaf Secret it signs,
382# so the CA certificate is projected, on its own, from the endpoint wildcard
383# Secret into a separate read-only volume: the router container never mounts a
384# private key. The path matches ``internal_tls.DEFAULT_INTERNAL_CA_FILE``, the
385# convention every GCO in-cluster client follows.
386PD_PROXY_CA_FILE_ENV = "PD_PROXY_CA_FILE"
387INTERNAL_CA_VOLUME = "internal-ca"
388INTERNAL_CA_KEY = "ca.crt"
389INTERNAL_CA_MOUNT_DIR = "/var/run/gco/ca"
390PD_PROXY_CA_FILE = f"{INTERNAL_CA_MOUNT_DIR}/{INTERNAL_CA_KEY}"
392# --- Endpoint TLS sidecar ----------------------------------------------------
393#
394# Every managed model pod (classic, canary, prefill, decode, store, and the PD
395# proxy) carries an ``endpoint-tls-proxy`` sidecar that serves the
396# ``gco-inference`` wildcard certificate on ENDPOINT_TLS_PORT and forwards the
397# decrypted bytes to the pod's own server over loopback. Model Services publish
398# only that port, so the inference proxy and the PD proxy reach a model solely
399# over verified HTTPS, while the model server keeps its bind address, port, and
400# probes: KV-transfer side channels and the kubelet still depend on them.
401ENDPOINT_TLS_PROXY_CONTAINER = "endpoint-tls-proxy"
402ENDPOINT_TLS_PORT = 8443
403ENDPOINT_TLS_PORT_NAME = "https"
404ENDPOINT_TLS_SECRET = "gco-inference-tls" # nosec B105 # Kubernetes Secret name, not a credential
405ENDPOINT_TLS_VOLUME = "endpoint-tls"
406ENDPOINT_TLS_MOUNT_DIR = "/var/run/gco/tls"
408# The sidecar runs the stdlib-only gco/services/tls_proxy.py source, shipped to
409# the pod in a per-endpoint ``{name}-tls-proxy`` ConfigMap exactly like the PD
410# proxy program. GCO's service images are linux/amd64 only, while managed model
411# pods also land on arm64 (Graviton GPU) nodes, so the sidecar cannot reuse a
412# service image; a plain multi-arch interpreter image plus the shipped source
413# runs identically on both.
414ENDPOINT_TLS_PROXY_SCRIPT_FILENAME = "tls_proxy.py"
415ENDPOINT_TLS_PROXY_SCRIPT_VOLUME = "endpoint-tls-proxy-script"
416ENDPOINT_TLS_PROXY_CONFIG_MOUNT_DIR = "/etc/gco-tls-proxy"
417ENDPOINT_TLS_PROXY_SCRIPT_PATH = (
418 f"{ENDPOINT_TLS_PROXY_CONFIG_MOUNT_DIR}/{ENDPOINT_TLS_PROXY_SCRIPT_FILENAME}"
419)
421# Official Docker Library Python image (ECR Public mirror) for that sidecar.
422# Keep the readable release tag and immutable OCI image-index digest together:
423# both amd64 and arm64 model nodes resolve through this single verified index.
424# The minor version must satisfy tls_proxy.py (it uses Python 3.14 syntax).
425ENDPOINT_TLS_PROXY_IMAGE = (
426 "public.ecr.aws/docker/library/python:3.14.8-slim@"
427 "sha256:f85c5697265c178cc6887276c55fe16cf3d14ca35c3df6a5eab3b360534a55d2"
428)
429ENDPOINT_TLS_PROXY_IMAGE_ENV = "ENDPOINT_TLS_PROXY_IMAGE"
431# The image defaults to root and the sidecar needs no identity or writable
432# path, so it runs as a fixed unprivileged UID. The Secret and ConfigMap
433# volumes are world-readable (0444) because model pods carry no fsGroup: GCO
434# does not own the model image's users.
435ENDPOINT_TLS_PROXY_UID = 65532
437# Model pods stop within the Kubernetes default grace period, stated explicitly
438# so the sidecar's drain budget is derived from it: the sidecar and the model
439# server receive SIGTERM together, and the sidecar keeps relaying accepted
440# streams for ENDPOINT_TLS_PROXY_DRAIN_SECONDS, then exits before the kubelet's
441# SIGKILL. No preStop sleep: the model server stops accepting at SIGTERM too, so
442# holding the TLS listener open longer would only accept doomed connections.
443MODEL_POD_TERMINATION_GRACE_SECONDS = 30
444ENDPOINT_TLS_PROXY_DRAIN_SECONDS = MODEL_POD_TERMINATION_GRACE_SECONDS - 5
446# The pods' service account carries IRSA / Pod Identity credentials; the
447# sidecar needs none, so the credential webhook is told to skip it.
448SKIP_CONTAINERS_ANNOTATION = "eks.amazonaws.com/skip-containers"
450# Digest of the rendered sidecar and its volumes, stamped on the pod template.
451# Reconcile patches a Deployment in place only when the digest differs, which
452# is how endpoints created before the sidecar existed pick it up, and how an
453# image override or model-port change reaches running pods without rewriting
454# unchanged templates on every pass.
455ENDPOINT_TLS_PROXY_DIGEST_ANNOTATION = "gco.io/endpoint-tls-proxy-digest"
457# Digest of the rendered ``mooncake.json``, stamped on every Mooncake role pod
458# template. The connectors read the file once at start-up, so re-rendering the
459# ConfigMap alone (``configure-store``, a transport change) would leave running
460# pods on the old configuration; a differing digest patches the template and
461# rolls the role, the same way the sidecar digest does.
462MOONCAKE_CONFIG_DIGEST_ANNOTATION = "gco.io/mooncake-config-digest"
465@dataclass
466class RegionServicesResolution:
467 """Outcome of resolving the in-region service addresses an endpoint needs.
469 ``render_mooncake_config`` consumes already-resolved values via a
470 ``region_services`` dict; this carries that dict together with the signals
471 a reconcile pass acts on:
473 - ``region_services`` is the resolved dict to render with, or ``None`` when
474 rendering must be skipped.
475 - ``render_skipped`` is set when the store is enabled but the own-region
476 master address is not configured: the existing endpoint configuration is
477 left untouched and ``store_master_unresolved`` records why.
478 - ``cold_tier_unresolved`` is set when the cold tier was requested but the
479 own-region general-purpose bucket could not be resolved; the cold tier is
480 dropped while the hot-path store keeps operating, and ``error`` explains
481 the condition.
482 """
484 region_services: dict[str, Any] | None = None
485 render_skipped: bool = False
486 store_master_unresolved: bool = False
487 cold_tier_unresolved: bool = False
488 error: str | None = None
491@dataclass
492class MasterReadinessGate:
493 """Outcome of gating dependent role-pod creation on the shared master.
495 A store-bearing endpoint must not materialize its role pods until the
496 single shared ``mooncake-master`` reports a Ready replica. This carries the
497 decision a reconcile pass acts on:
499 - ``proceed`` is ``True`` only when the master reports at least one Ready
500 replica; the caller may then create the dependent role pods and advance
501 out of ``creating``. While it is ``False`` the caller materializes no
502 dependent pods.
503 - ``state`` is the endpoint state to report. It is ``"creating"`` whenever
504 creation is deferred (master not ready, still within the wait window,
505 timed out, or could not be created) and ``None`` when the gate is open.
506 - ``error`` records why creation could not advance: the master did not
507 become Ready within the wait window, or the master could not be created.
508 It is ``None`` while the master is simply still coming up within the
509 window, and ``None`` once the gate is open.
510 """
512 proceed: bool = False
513 state: str | None = None
514 error: str | None = None
517@dataclass
518class RegionalScopeResolution:
519 """Outcome of confirming a disaggregated topology stays inside one region.
521 KV cache transfer over RoCE cannot cross a region boundary, so every
522 ``MooncakeConnector`` peer address and the ``master_server_address`` a
523 topology wires to must resolve to the monitor's own region. An endpoint
524 that targets several regions runs one independent topology per region; each
525 region's monitor reconciles only its own topology and confirms that
526 topology's addresses never escape the region.
528 - ``in_region`` is ``True`` only when every resolved address belongs to the
529 monitor's own region. While it is ``True`` the caller may materialize the
530 topology's role Deployments.
531 - ``peer_addresses`` lists the addresses that were resolved and checked, in
532 a stable order, so callers and logs can show exactly what was wired.
533 - ``state`` is the endpoint state to report. It is ``"failed"`` when an
534 out-of-region address is detected and ``None`` when the topology is
535 wholly in-region.
536 - ``error`` describes the cross-region boundary violation — which addresses
537 resolved to which other regions — when one is found, and is ``None``
538 otherwise. When a violation is reported the caller materializes no role
539 Deployments and leaves any previously materialized resources unchanged.
540 """
542 in_region: bool = True
543 peer_addresses: list[str] = field(default_factory=list)
544 state: str | None = None
545 error: str | None = None
548class RegionStatusConditions(TypedDict, total=False):
549 """Optional DynamoDB predicates shared by every regional status write."""
551 expected_lifecycle_id: str
552 expected_region_generation: str
553 expected_deletion_generation: str
556_LIFECYCLE_ANNOTATION = "gco.io/lifecycle-id"
557_REGION_GENERATION_ANNOTATION = "gco.io/region-generation"
558_LEADER_EPOCH_ANNOTATION = "gco.io/leader-epoch"
561class ReconcileFencedError(RuntimeError):
562 """The caller no longer owns the endpoint or leader epoch it read."""
565@dataclass(frozen=True)
566class ReconcileAuthority:
567 """Immutable authority carried by one endpoint reconciliation pass."""
569 endpoint_name: str
570 lifecycle_id: str
571 region_generation: str
572 leader_epoch: str
573 deletion_generation: str | None = None
574 deleting: bool = False
575 region_removed: bool = False
577 @property
578 def annotations(self) -> dict[str, str]:
579 return {
580 _LIFECYCLE_ANNOTATION: self.lifecycle_id,
581 _REGION_GENERATION_ANNOTATION: self.region_generation,
582 _LEADER_EPOCH_ANNOTATION: self.leader_epoch,
583 }
586@dataclass(frozen=True)
587class EndpointResourceInventory:
588 """Deterministic names of every top-level Kubernetes object owned by an endpoint."""
590 deployments: tuple[str, ...]
591 services: tuple[str, ...]
592 horizontal_pod_autoscalers: tuple[str, ...]
593 scaled_objects: tuple[str, ...]
594 config_maps: tuple[str, ...]
595 legacy_ingresses: tuple[str, ...]
596 legacy_http_routes: tuple[str, ...]
597 generated_admin_secret: str
600@dataclass(frozen=True)
601class ResourceCleanupResult:
602 """Observed result of one idempotent endpoint cleanup pass.
604 A successful Kubernetes delete only starts asynchronous deletion. ``pending``
605 therefore contains objects still returned by a read-after-delete, while
606 ``errors`` contains sanitized request failures. Cleanup is complete only
607 when every owned object is independently observed absent and no request
608 failed during the pass.
609 """
611 pending: tuple[str, ...] = ()
612 errors: tuple[str, ...] = ()
613 resources_found: bool = False
615 @property
616 def complete(self) -> bool:
617 return not self.pending and not self.errors
619 @property
620 def error_message(self) -> str | None:
621 if not self.errors:
622 return None
623 return "Endpoint cleanup could not be completed: " + "; ".join(self.errors)
626def _resolved_mooncake_transfer(mooncake: dict[str, Any]) -> tuple[str, str]:
627 """Resolve the persisted transfer intent and optional network device.
629 GCO's spec deliberately uses ``rdma`` as the portable high-performance
630 intent because the mounted Mooncake store configuration accepts
631 ``rdma|tcp``. On AWS, role pods with that intent are placed on EFA nodes,
632 so :func:`build_kv_transfer_config` translates it to vLLM's explicit
633 ``mooncake_protocol=efa`` at the point-to-point connector boundary.
634 """
635 transfer = mooncake.get("transfer", {})
636 if not isinstance(transfer, dict):
637 raise ValueError("mooncake.transfer must be a mapping")
639 protocol = transfer.get("protocol", "rdma")
640 if protocol not in {"rdma", "tcp"}:
641 raise ValueError(
642 f"mooncake.transfer.protocol must be one of {{rdma, tcp}}, got {protocol!r}"
643 )
644 device_name = transfer.get("device_name", "")
645 if not isinstance(device_name, str):
646 raise ValueError(f"mooncake.transfer.device_name must be a string, got {device_name!r}")
647 return protocol, device_name
650def build_kv_transfer_config(mooncake: dict[str, Any], role: str) -> str:
651 """Return the JSON string for vLLM's ``--kv-transfer-config``.
653 Translates a mooncake spec block plus a worker role into the connector
654 configuration vLLM expects:
656 - ``disaggregated`` emits a ``MooncakeConnector``.
657 - ``store`` emits a ``MooncakeStoreConnector``.
658 - ``both`` emits a ``MultiConnector`` wrapping a ``MooncakeConnector``
659 (index 0) followed by a ``MooncakeStoreConnector`` (index 1), both
660 sharing the role's ``kv_role``.
662 Every point-to-point ``MooncakeConnector`` receives explicit
663 ``kv_connector_extra_config``. GCO's default/high-performance ``rdma``
664 intent maps to Mooncake's AWS-specific ``efa`` protocol because the same
665 pod is pinned to the EFA node pool; ``tcp`` remains an explicit fallback.
666 ``device_name`` is forwarded verbatim, with an empty string requesting
667 Mooncake/libfabric auto-detection.
669 The emitted ``kv_role`` is ``kv_producer`` for prefill, ``kv_consumer``
670 for decode, and ``kv_both`` for a single store instance.
672 Args:
673 mooncake: The ``spec["mooncake"]`` block; its ``mode`` selects the
674 connector shape and its optional ``transfer`` block selects the
675 protocol/device.
676 role: One of ``"prefill"``, ``"decode"``, or ``"single"``.
678 Returns:
679 A JSON object string parseable by vLLM.
681 Raises:
682 ValueError: If the ``(mode, role)`` combination or transfer settings
683 are unsupported. No configuration is emitted in that case.
684 """
685 mode = mooncake.get("mode")
686 supported_roles = _WORKER_ROLES_BY_MODE.get(mode) if isinstance(mode, str) else None
687 if supported_roles is None or role not in supported_roles:
688 raise ValueError(f"Unsupported (mode, role) pair: ({mode!r}, {role!r})")
690 kv_role = _KV_ROLE_BY_WORKER_ROLE[role]
692 if mode == "store":
693 return json.dumps({"kv_connector": "MooncakeStoreConnector", "kv_role": kv_role})
695 protocol, device_name = _resolved_mooncake_transfer(mooncake)
696 connector = {
697 "kv_connector": "MooncakeConnector",
698 "kv_role": kv_role,
699 "kv_connector_extra_config": {
700 "mooncake_protocol": "efa" if protocol == "rdma" else "tcp",
701 "device_name": device_name,
702 },
703 }
704 if mode == "disaggregated":
705 return json.dumps(connector)
707 # mode == "both": MultiConnector chains transfer then store.
708 return json.dumps(
709 {
710 "kv_connector": "MultiConnector",
711 "kv_role": kv_role,
712 "kv_connector_extra_config": {
713 "connectors": [
714 connector,
715 {"kv_connector": "MooncakeStoreConnector", "kv_role": kv_role},
716 ]
717 },
718 }
719 )
722def bootstrap_port_for_worker(base_port: int, dp_rank: int, tp_size: int, tp_rank: int) -> int:
723 """Compute the bootstrap port for a ``(dp_rank, tp_rank)`` worker.
725 The port is ``base_port + dp_rank * tp_size + tp_rank``. For a fixed
726 ``base_port`` and ``tp_size`` distinct ``(dp_rank, tp_rank)`` pairs map to
727 distinct ports.
729 Args:
730 base_port: The base bootstrap port for the endpoint.
731 dp_rank: The data-parallel rank of the worker (``>= 0``).
732 tp_size: The tensor-parallel world size (``>= 1``).
733 tp_rank: The tensor-parallel rank within the worker (``0 <= tp_rank < tp_size``).
735 Returns:
736 The TCP port assigned to the worker.
738 Raises:
739 ValueError: If the computed port falls outside the valid range
740 ``1024..65535``. No port is assigned in that case.
741 """
742 port = base_port + dp_rank * tp_size + tp_rank
743 if port < MIN_BOOTSTRAP_PORT or port > MAX_BOOTSTRAP_PORT:
744 raise ValueError(
745 f"Computed bootstrap port {port} is outside the valid range "
746 f"{MIN_BOOTSTRAP_PORT}..{MAX_BOOTSTRAP_PORT}"
747 )
748 return port
751def render_mooncake_config(
752 mooncake: dict[str, Any], region_services: dict[str, Any]
753) -> dict[str, Any]:
754 """Render the ``mooncake.json`` contents mounted into each vLLM pod.
756 The returned dict is written verbatim to a ConfigMap and mounted at the
757 path named by ``MOONCAKE_CONFIG_PATH``. It always carries the metadata
758 server and the RDMA/TCP transport settings (``protocol`` and
759 ``device_name``); the key-value store and its optional cold tier are layered
760 on only when requested.
762 The transport block (``protocol``/``device_name``) describes the hot
763 RDMA/RoCE path. The cold tier is an asynchronous object-store backend keyed
764 separately as ``cold_tier_s3_uri``; it is never wired into the transport
765 block, so cold-tier reads and writes stay off the RDMA hot path.
767 Resolution of in-region addresses is the caller's responsibility: this
768 function consumes already-resolved values from ``region_services`` and
769 performs no lookups of its own. In particular, the cold-tier URI is the
770 monitor-resolved general-purpose regional bucket for the monitor's own
771 region; any cold-tier bucket URI in the user spec is ignored.
773 Args:
774 mooncake: The ``spec["mooncake"]`` block.
775 region_services: Resolved in-region addresses, e.g.::
777 {
778 "metadata_server": "http://mooncake-master:8080/metadata",
779 "master_server_address": "mooncake-master:50051",
780 "cold_tier_s3_uri": "s3://gco-regional-shared-<acct>-<region>/...",
781 }
783 ``master_server_address`` is required when the store is enabled.
784 ``cold_tier_s3_uri`` carries the resolved cold-tier bucket; when the
785 caller could not resolve one (see
786 :class:`RegionServicesResolution.cold_tier_unresolved`) the key is
787 absent and the cold tier is left off.
789 Returns:
790 The ``mooncake.json`` contents as a dict, where:
792 - ``protocol`` and ``device_name`` are always present.
793 - ``master_server_address`` is present only when the store is enabled.
794 - ``cold_tier_s3_uri`` is present only when the cold tier is enabled,
795 which requires ``cold_tier_enabled`` to be the boolean ``True``, and
796 the caller resolved a URI for it; any other ``cold_tier_enabled``
797 value (absent, null, truthy non-bool) or an unresolved URI leaves the
798 cold tier off so the hot-path store keeps operating.
799 """
800 protocol, device_name = _resolved_mooncake_transfer(mooncake)
801 store = mooncake.get("store", {})
802 cfg: dict[str, Any] = {
803 "metadata_server": region_services["metadata_server"],
804 "protocol": protocol,
805 "device_name": device_name,
806 }
807 if store.get("enabled"):
808 cfg["master_server_address"] = region_services["master_server_address"]
809 # The store runs embedded in each vLLM pod (every rank contributes
810 # `global_segment_size` to the shared pool; GCO's per-region
811 # mooncake-master is only the metadata/master coordinator, not a
812 # standalone store that owns the pool). Embedded mode rejects a zero
813 # segment, so default to 4 GiB (the upstream default) when the spec
814 # does not set one; an operator can tune it via configure-store.
815 cfg["global_segment_size"] = store.get("global_segment_size", "4294967296")
816 cfg["local_buffer_size"] = store.get("local_buffer_size", "2147483648")
817 # Only the boolean True enables the cold tier; any other value leaves it
818 # off. The URI is resolved by the caller for the monitor's own region —
819 # never authored by the user — and is an object-store backend kept off
820 # the RDMA transport block above. When the caller could not resolve the
821 # own-region bucket it omits the key: the cold tier is dropped for this
822 # pass while the hot-path store stays configured, instead of failing
823 # the whole endpoint on a missing discovery value.
824 cold_tier_s3_uri = region_services.get("cold_tier_s3_uri")
825 if store.get("cold_tier_enabled") is True and cold_tier_s3_uri:
826 cfg["cold_tier_s3_uri"] = cold_tier_s3_uri
827 return cfg
830def mooncake_config_digest(cfg: dict[str, Any]) -> str:
831 """Return the stable digest of a rendered ``mooncake.json``.
833 Keys are sorted and separators fixed so the same configuration always
834 hashes identically across passes and monitor replicas; the value is what
835 :data:`MOONCAKE_CONFIG_DIGEST_ANNOTATION` carries on role pod templates.
836 """
837 encoded = json.dumps(cfg, sort_keys=True, separators=(",", ":"), default=str)
838 return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
841def apply_efa_scheduling(mooncake: dict[str, Any], pod_spec: client.V1PodSpec) -> None:
842 """Place a role pod on the EFA RDMA fabric when transfer runs over RDMA.
844 KV cache transfer over RoCE only runs on EFA-enabled nodes, which carry a
845 ``vpc.amazonaws.com/efa`` taint, advertise the ``vpc.amazonaws.com/efa``
846 extended resource, and are labelled ``efa=true``. When the transfer
847 protocol is ``rdma`` this mutates ``pod_spec`` in place to:
849 - add a ``vpc.amazonaws.com/efa`` toleration (in addition to any existing
850 tolerations such as the GPU one),
851 - add an ``efa=true`` node selector plus a ``mooncake-efa=true`` node
852 selector (merged with any existing selectors), and
853 - request at least one ``vpc.amazonaws.com/efa`` device on the pod's
854 workload containers (never the TLS sidecar), leaving every existing
855 resource request and limit — including GPU asks — untouched.
857 The ``mooncake-efa=true`` selector pins the pod to the dedicated
858 ``mooncake-efa-pool`` NodePool, which only offers instance families with
859 >=80GB of GPU memory and FP8-capable Hopper/Blackwell GPUs. This keeps role
860 pods off the A100-40GB ``p4d`` family that the shared training EFA pool
861 still offers — that family OOMs on many models and cannot run FP8 KV-cache
862 configs, so Karpenter selecting it for a mooncake pod is a latent failure.
864 When the transfer protocol is explicitly set to anything other than
865 ``rdma`` (for example ``tcp``) the pod is left exactly as it was: no
866 toleration, no node selector, and no device request are added. An unset
867 protocol defaults to ``rdma`` — matching the rest of the Mooncake path — so
868 a disaggregated endpoint lands on EFA by default.
870 Tolerations, selectors, and device requests are applied idempotently, so
871 re-running over an already-scheduled pod produces no duplicates.
873 Args:
874 mooncake: The ``spec["mooncake"]`` block; ``transfer.protocol``
875 (defaulting to ``rdma`` when unset) decides whether EFA scheduling
876 applies.
877 pod_spec: The pod specification to mutate in place.
878 """
879 protocol, _device_name = _resolved_mooncake_transfer(mooncake)
880 if protocol != "rdma":
881 return
883 # Tolerate the EFA taint without disturbing existing tolerations.
884 tolerations = list(pod_spec.tolerations or [])
885 if not any(t.key == EFA_RESOURCE_NAME for t in tolerations):
886 tolerations.append(
887 client.V1Toleration(
888 key=EFA_RESOURCE_NAME,
889 operator="Equal",
890 value="true",
891 effect="NoSchedule",
892 )
893 )
894 pod_spec.tolerations = tolerations
896 # Merge the EFA node selectors with any selectors already in place. The
897 # generic efa=true selector lands the pod on EFA fabric; mooncake-efa=true
898 # narrows that to the dedicated mooncake-efa-pool, which excludes the
899 # A100-40GB p4d family that the shared training EFA pool still offers.
900 node_selector = dict(pod_spec.node_selector or {})
901 node_selector[EFA_NODE_SELECTOR_KEY] = EFA_NODE_SELECTOR_VALUE
902 node_selector[MOONCAKE_EFA_NODE_SELECTOR_KEY] = MOONCAKE_EFA_NODE_SELECTOR_VALUE
903 pod_spec.node_selector = node_selector
905 # Request at least one EFA device, preserving existing requests and limits
906 # (notably the GPU asks). Apply to containers that already request an
907 # accelerator; if none do, apply to every workload container so the pod
908 # still asks for the fabric it needs. The TLS sidecar only relays HTTPS over
909 # loopback and never takes a device.
910 containers = [
911 container
912 for container in pod_spec.containers or []
913 if getattr(container, "name", None) != ENDPOINT_TLS_PROXY_CONTAINER
914 ]
915 accelerator_keys = ("nvidia.com/gpu", "aws.amazon.com/neuron")
917 def _requests_accelerator(container: client.V1Container) -> bool:
918 reqs = container.resources
919 if reqs is None:
920 return False
921 for table in (reqs.requests, reqs.limits):
922 if table and any(key in table for key in accelerator_keys):
923 return True
924 return False
926 targets = [c for c in containers if _requests_accelerator(c)] or list(containers)
927 for container in targets:
928 if container.resources is None:
929 container.resources = client.V1ResourceRequirements()
930 if container.resources.requests is None:
931 container.resources.requests = {}
932 if container.resources.limits is None:
933 container.resources.limits = {}
934 container.resources.requests.setdefault(EFA_RESOURCE_NAME, "1")
935 container.resources.limits.setdefault(EFA_RESOURCE_NAME, "1")
938def build_pd_proxy_config(mooncake: dict[str, Any]) -> dict[str, str]:
939 """Return the environment the prefill-decode proxy runs with.
941 The proxy fronts a disaggregated endpoint and decides, per request, whether
942 to consult the shared store and which backends to dispatch to. Its behavior
943 is fixed by the values returned here so every disaggregated endpoint front
944 behaves identically:
946 - It looks up whether the prompt's KV blocks already reside in the store
947 before sending the prompt to prefill, and that lookup is bounded to
948 ``PD_PROXY_RESIDENCY_TIMEOUT_SECONDS`` seconds.
949 - A miss, or a lookup that does not finish in time, is treated as "not
950 resident": the prompt is sent to a prefill pod and the request is never
951 held waiting on the store.
952 - Decode-phase requests are routed only to decode pods reporting Ready, so a
953 pod that is still starting is skipped.
954 - When no decode pod reports Ready, the proxy refuses the request with a
955 stable status and message rather than streaming any partial output.
957 The residency bound is held constant rather than read from the spec so the
958 responsiveness guarantee cannot be weakened per endpoint. The store address
959 points at the shared in-region master, and the same-role dispatch strategy
960 falls back to round-robin when the spec names none.
962 Args:
963 mooncake: The ``spec["mooncake"]`` block; its optional ``proxy`` section
964 supplies the same-role scheduling strategy.
966 Returns:
967 A mapping of environment variable name to value, ready to attach to the
968 proxy container.
969 """
970 proxy = mooncake.get("proxy", {}) or {}
971 scheduling = proxy.get("scheduling") or PD_PROXY_DEFAULT_SCHEDULING
972 store_address = f"{MOONCAKE_MASTER_SERVICE}:{MOONCAKE_MASTER_RPC_PORT}"
973 return {
974 PD_PROXY_RESIDENCY_TIMEOUT_ENV: str(PD_PROXY_RESIDENCY_TIMEOUT_SECONDS),
975 PD_PROXY_RESIDENCY_BLOCKING_ENV: PD_PROXY_RESIDENCY_BLOCKING,
976 PD_PROXY_RESIDENCY_MISS_TARGET_ENV: PD_PROXY_RESIDENCY_MISS_TARGET,
977 PD_PROXY_DECODE_ROUTING_ENV: PD_PROXY_DECODE_ROUTING_READY_ONLY,
978 PD_PROXY_NO_DECODE_BACKEND_ACTION_ENV: PD_PROXY_NO_DECODE_BACKEND_ACTION_REJECT,
979 PD_PROXY_NO_DECODE_BACKEND_STATUS_ENV: PD_PROXY_NO_DECODE_BACKEND_STATUS,
980 PD_PROXY_NO_DECODE_BACKEND_MESSAGE_ENV: PD_PROXY_NO_DECODE_BACKEND_MESSAGE,
981 PD_PROXY_SCHEDULING_ENV: scheduling,
982 PD_PROXY_STORE_ADDRESS_ENV: store_address,
983 }
986def _role_service_url(name: str, ns: str, role: str) -> str:
987 """Return the verified-HTTPS URL the PD proxy uses for one role Service.
989 The fully qualified Service name is what the ``gco-inference`` wildcard
990 certificate covers, so hostname verification in the router succeeds
991 against exactly the name it dials.
992 """
993 return f"https://{name}-{role}.{ns}.svc.cluster.local:{ENDPOINT_TLS_PORT}"
996def _tls_proxy_config_map_name(name: str) -> str:
997 """Name of the per-endpoint ConfigMap carrying the TLS sidecar program."""
998 return f"{name}-tls-proxy"
1001def _endpoint_tls_proxy_image() -> str:
1002 """Return the TLS sidecar image: the env override or the pinned default."""
1003 override = os.environ.get(ENDPOINT_TLS_PROXY_IMAGE_ENV, "").strip()
1004 return override or ENDPOINT_TLS_PROXY_IMAGE
1007def _merge_skip_containers(existing: object) -> str:
1008 """Add the TLS sidecar to an ``eks.amazonaws.com/skip-containers`` value.
1010 The annotation is a comma-separated container list; entries already present
1011 (including the sidecar itself) are kept in order so the merge is idempotent.
1012 """
1013 names = [part.strip() for part in existing.split(",")] if isinstance(existing, str) else []
1014 merged = [part for part in names if part]
1015 if ENDPOINT_TLS_PROXY_CONTAINER not in merged:
1016 merged.append(ENDPOINT_TLS_PROXY_CONTAINER)
1017 return ",".join(merged)
1020def _vllm_args_name_model(args: list[str]) -> bool:
1021 """True when vLLM arguments already choose the model to serve.
1023 The model is named by ``--model`` / ``--model=...``, by the positional tag
1024 ``vllm serve`` takes first, or inside a ``--config`` file the operator owns.
1025 """
1026 tokens = [str(arg) for arg in args]
1027 if tokens and not tokens[0].startswith("-"):
1028 return True
1029 return any(
1030 token in ("--model", "--config") or token.startswith(("--model=", "--config="))
1031 for token in tokens
1032 )
1035def _reject_tls_port_collision(port: object) -> None:
1036 """Refuse a model port that would collide with the TLS sidecar listener.
1038 The sidecar and the model server share the pod's network namespace, so a
1039 model server on ENDPOINT_TLS_PORT could never start alongside it. Failing
1040 the render names the conflict instead of leaving a crash-looping pod.
1041 """
1042 if port == ENDPOINT_TLS_PORT:
1043 raise ValueError(
1044 f"endpoint port {ENDPOINT_TLS_PORT} is reserved for the model pod's TLS sidecar; "
1045 "serve the model on a different spec.port"
1046 )
1049def _reject_bootstrap_window_collision(base_port: int) -> None:
1050 """Refuse a KV-transfer bootstrap window that contains the TLS sidecar port.
1052 Mooncake workers bind ``base_port + dp_rank * tp_size + tp_rank``, anywhere
1053 in the inclusive window the ``allow-rdma-bootstrap`` rule opens
1054 (``base_port`` through ``base_port + MOONCAKE_BOOTSTRAP_PORT_SPAN``). A
1055 worker landing on ENDPOINT_TLS_PORT would collide with the sidecar in the
1056 shared pod network namespace, so the render names the conflict instead.
1057 """
1058 if base_port <= ENDPOINT_TLS_PORT <= base_port + MOONCAKE_BOOTSTRAP_PORT_SPAN:
1059 lowest_clash = ENDPOINT_TLS_PORT - MOONCAKE_BOOTSTRAP_PORT_SPAN
1060 raise ValueError(
1061 f"mooncake.transfer.bootstrap_base_port {base_port} opens the KV-transfer "
1062 f"bootstrap window {base_port}-{base_port + MOONCAKE_BOOTSTRAP_PORT_SPAN}, which "
1063 f"contains port {ENDPOINT_TLS_PORT}, reserved for the model pod's TLS sidecar; "
1064 f"choose a base port outside {lowest_clash}-{ENDPOINT_TLS_PORT}"
1065 )
1068@dataclass(frozen=True)
1069class EndpointTLSProxy:
1070 """The TLS sidecar a managed model pod carries, with its volumes and digest."""
1072 container: client.V1Container
1073 volumes: tuple[client.V1Volume, ...]
1074 digest: str
1077def build_endpoint_tls_proxy(upstream_port: int, config_map_name: str) -> EndpointTLSProxy:
1078 """Render the ``endpoint-tls-proxy`` sidecar for one model pod.
1080 The sidecar serves the ``gco-inference-tls`` wildcard keypair on
1081 ENDPOINT_TLS_PORT and relays to ``127.0.0.1:upstream_port`` (the model
1082 server, or the PD router). It runs ``tls_proxy.py`` from the endpoint's
1083 ConfigMap on the pinned multi-arch Python image as a fixed unprivileged
1084 UID with a read-only root filesystem and no capabilities. Its probes are
1085 socket checks against its own listener only, so a slow model never restarts
1086 the sidecar and a sidecar fault never restarts the model. It requests a
1087 little CPU and memory (autoscaler Resource metrics need a request on every
1088 container) and is memory-limited but not CPU-limited, so CFS throttling
1089 never adds latency to streamed tokens.
1091 The digest covers everything rendered here plus the pod grace period, so a
1092 changed image, upstream port, or sidecar shape is detected on existing
1093 Deployments; the program's source is deliberately excluded (see
1094 :meth:`InferenceMonitor._ensure_tls_proxy_configmap`).
1095 """
1096 container = client.V1Container(
1097 name=ENDPOINT_TLS_PROXY_CONTAINER,
1098 image=_endpoint_tls_proxy_image(),
1099 image_pull_policy="IfNotPresent",
1100 command=["python3", ENDPOINT_TLS_PROXY_SCRIPT_PATH],
1101 env=[
1102 # The root filesystem is read-only; never try to write bytecode.
1103 client.V1EnvVar(name="PYTHONDONTWRITEBYTECODE", value="1"),
1104 client.V1EnvVar(name="TLS_PROXY_PORT", value=str(ENDPOINT_TLS_PORT)),
1105 client.V1EnvVar(name="TLS_PROXY_UPSTREAM_HOST", value="127.0.0.1"),
1106 client.V1EnvVar(name="TLS_PROXY_UPSTREAM_PORT", value=str(upstream_port)),
1107 client.V1EnvVar(name="GCO_TLS_CERT_FILE", value=f"{ENDPOINT_TLS_MOUNT_DIR}/tls.crt"),
1108 client.V1EnvVar(name="GCO_TLS_KEY_FILE", value=f"{ENDPOINT_TLS_MOUNT_DIR}/tls.key"),
1109 client.V1EnvVar(
1110 name="GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
1111 value=str(ENDPOINT_TLS_PROXY_DRAIN_SECONDS),
1112 ),
1113 ],
1114 ports=[
1115 client.V1ContainerPort(
1116 name=ENDPOINT_TLS_PORT_NAME,
1117 container_port=ENDPOINT_TLS_PORT,
1118 protocol="TCP",
1119 )
1120 ],
1121 resources=client.V1ResourceRequirements(
1122 requests={"cpu": "50m", "memory": "64Mi"},
1123 limits={"memory": "256Mi"},
1124 ),
1125 security_context=client.V1SecurityContext(
1126 allow_privilege_escalation=False,
1127 capabilities=client.V1Capabilities(drop=["ALL"]),
1128 privileged=False,
1129 read_only_root_filesystem=True,
1130 run_as_non_root=True,
1131 run_as_user=ENDPOINT_TLS_PROXY_UID,
1132 run_as_group=ENDPOINT_TLS_PROXY_UID,
1133 seccomp_profile=client.V1SeccompProfile(type="RuntimeDefault"),
1134 ),
1135 readiness_probe=client.V1Probe(
1136 tcp_socket=client.V1TCPSocketAction(port=ENDPOINT_TLS_PORT_NAME),
1137 period_seconds=5,
1138 timeout_seconds=3,
1139 failure_threshold=3,
1140 ),
1141 liveness_probe=client.V1Probe(
1142 tcp_socket=client.V1TCPSocketAction(port=ENDPOINT_TLS_PORT_NAME),
1143 initial_delay_seconds=10,
1144 period_seconds=15,
1145 timeout_seconds=5,
1146 failure_threshold=4,
1147 ),
1148 volume_mounts=[
1149 client.V1VolumeMount(
1150 name=ENDPOINT_TLS_VOLUME,
1151 mount_path=ENDPOINT_TLS_MOUNT_DIR,
1152 read_only=True,
1153 ),
1154 client.V1VolumeMount(
1155 name=ENDPOINT_TLS_PROXY_SCRIPT_VOLUME,
1156 mount_path=ENDPOINT_TLS_PROXY_CONFIG_MOUNT_DIR,
1157 read_only=True,
1158 ),
1159 ],
1160 )
1161 volumes = (
1162 client.V1Volume(
1163 name=ENDPOINT_TLS_VOLUME,
1164 secret=client.V1SecretVolumeSource(
1165 secret_name=ENDPOINT_TLS_SECRET,
1166 default_mode=0o444,
1167 ),
1168 ),
1169 client.V1Volume(
1170 name=ENDPOINT_TLS_PROXY_SCRIPT_VOLUME,
1171 config_map=client.V1ConfigMapVolumeSource(
1172 name=config_map_name,
1173 default_mode=0o444,
1174 ),
1175 ),
1176 )
1177 rendered = {
1178 "container": container.to_dict(),
1179 "volumes": [volume.to_dict() for volume in volumes],
1180 "termination_grace_period_seconds": MODEL_POD_TERMINATION_GRACE_SECONDS,
1181 }
1182 digest = hashlib.sha256(
1183 json.dumps(rendered, sort_keys=True, separators=(",", ":")).encode("utf-8")
1184 ).hexdigest()
1185 return EndpointTLSProxy(container=container, volumes=volumes, digest=digest)
1188def _endpoint_tls_pod_annotations(
1189 existing: Mapping[str, str] | None, sidecar: EndpointTLSProxy
1190) -> dict[str, str]:
1191 """Pod-template annotations for a model pod carrying the TLS sidecar."""
1192 annotations = dict(existing or {})
1193 annotations[SKIP_CONTAINERS_ANNOTATION] = _merge_skip_containers(
1194 annotations.get(SKIP_CONTAINERS_ANNOTATION)
1195 )
1196 annotations[ENDPOINT_TLS_PROXY_DIGEST_ANNOTATION] = sidecar.digest
1197 return annotations
1200def _endpoint_service_ports() -> list[client.V1ServicePort]:
1201 """The only port a model Service publishes: the pods' TLS sidecar."""
1202 return [
1203 client.V1ServicePort(
1204 name=ENDPOINT_TLS_PORT_NAME,
1205 port=ENDPOINT_TLS_PORT,
1206 target_port=ENDPOINT_TLS_PORT_NAME,
1207 protocol="TCP",
1208 )
1209 ]
1212def _service_publishes_only_tls(service: Any) -> bool:
1213 """Whether a live Service already exposes exactly the TLS sidecar port."""
1214 ports = getattr(getattr(service, "spec", None), "ports", None)
1215 if not isinstance(ports, list) or len(ports) != 1:
1216 return False
1217 [port] = ports
1218 return (
1219 getattr(port, "port", None) == ENDPOINT_TLS_PORT
1220 and getattr(port, "name", None) == ENDPOINT_TLS_PORT_NAME
1221 and getattr(port, "target_port", None) == ENDPOINT_TLS_PORT_NAME
1222 and (getattr(port, "protocol", None) or "TCP") == "TCP"
1223 )
1226def _model_server_port(containers: list[Any], fallback: int) -> int:
1227 """Return the port the live model container serves on, else ``fallback``.
1229 In-place reconciliation reads the port from the running template rather
1230 than the spec: only the image and replica count of an existing Deployment
1231 are reconciled, so the model keeps listening where it was created, and the
1232 sidecar must forward there.
1233 """
1234 for container in containers:
1235 if getattr(container, "name", None) != "inference":
1236 continue
1237 ports = getattr(container, "ports", None)
1238 for port in ports if isinstance(ports, list) else []:
1239 value = getattr(port, "container_port", None)
1240 if isinstance(value, int) and not isinstance(value, bool):
1241 return value
1242 return fallback
1245def _pd_proxy_health_probe(*, liveness: bool) -> client.V1Probe:
1246 """Exec probe for the PD router, which listens on pod loopback only.
1248 The kubelet dials probes at the pod IP, so neither an httpGet nor a
1249 tcpSocket check can reach a loopback listener. An exec probe checks the
1250 router process itself; probing through the TLS sidecar instead would let
1251 a sidecar fault restart a healthy router (the sidecar has its own probes).
1252 It follows the repo's lean exec-probe shape: an isolated interpreter
1253 without ``site`` and a bare-socket request instead of ``urllib``, with
1254 every socket wait bounded below the probe timeout. The router image ships
1255 ``python3`` (not ``python``), the same interpreter the pod already runs.
1256 """
1257 command = [
1258 "python3",
1259 "-I",
1260 "-S",
1261 "-c",
1262 "import socket,sys;"
1263 f"s=socket.create_connection(('127.0.0.1',{PD_PROXY_PORT}),3);"
1264 "s.settimeout(3);"
1265 "s.sendall(b'GET /healthz HTTP/1.0\\r\\n\\r\\n');"
1266 "sys.exit(0 if s.makefile('rb').readline().split(b' ')[1:2]==[b'200'] else 1)",
1267 ]
1268 if liveness:
1269 return client.V1Probe(
1270 _exec=client.V1ExecAction(command=command),
1271 initial_delay_seconds=30,
1272 period_seconds=15,
1273 timeout_seconds=10,
1274 failure_threshold=5,
1275 )
1276 return client.V1Probe(
1277 _exec=client.V1ExecAction(command=command),
1278 initial_delay_seconds=10,
1279 period_seconds=10,
1280 timeout_seconds=5,
1281 failure_threshold=3,
1282 )
1285class InferenceMonitor:
1286 """
1287 Reconciliation controller for inference endpoints.
1289 Polls DynamoDB for desired endpoint state and reconciles with
1290 the actual Kubernetes resources in the local cluster.
1291 """
1293 def __init__(
1294 self,
1295 cluster_id: str,
1296 region: str,
1297 store: InferenceEndpointStore,
1298 namespace: str = "gco-inference",
1299 reconcile_interval: int = 15,
1300 ):
1301 self.cluster_id = cluster_id
1302 self.region = region
1303 self.store = store
1304 self.namespace = namespace
1305 self.reconcile_interval = reconcile_interval
1306 self._running = False
1308 # Initialize Kubernetes clients
1309 try:
1310 config.load_incluster_config()
1311 logger.info("Loaded in-cluster Kubernetes configuration")
1312 except config.ConfigException:
1313 try:
1314 config.load_kube_config()
1315 logger.info("Loaded local Kubernetes configuration")
1316 except config.ConfigException as e:
1317 logger.error("Failed to load Kubernetes configuration: %s", e)
1318 raise
1320 self.apps_v1 = client.AppsV1Api()
1321 self.core_v1 = client.CoreV1Api()
1322 self.networking_v1 = client.NetworkingV1Api()
1323 self.discovery_v1 = client.DiscoveryV1Api()
1325 # Timeout for Kubernetes API calls (seconds)
1326 self._k8s_timeout = int(os.environ.get("K8S_API_TIMEOUT", "30"))
1328 # Health watchdog: tracks when each endpoint first became unready.
1329 # Inference traffic enters through the shared ``gco-system/gco-gateway``
1330 # HTTPRoute at ``/inference`` and then ``gco-system/inference-proxy``, so
1331 # model readiness never mutates shared Gateway API resources. Once this
1332 # threshold is exceeded, reconciliation emits an explicit degraded-state
1333 # warning while the proxy continues returning 503 until a replica is ready.
1334 self._unready_since: dict[str, datetime] = {}
1335 self._unhealthy_threshold_seconds = int(
1336 os.environ.get("INFERENCE_UNHEALTHY_THRESHOLD_SECONDS", "300")
1337 ) # 5 minutes default
1339 # Master-readiness gate: tracks when each store-bearing endpoint first
1340 # deferred its role-pod creation because the shared master was not yet
1341 # Ready. The entry is cleared once the master reports a Ready replica so
1342 # a later restart of the master restarts the clock cleanly.
1343 self._master_deferral_since: dict[str, datetime] = {}
1345 # Leader authority is renewed from a dedicated thread while reconcile
1346 # performs synchronous Kubernetes calls. The per-acquisition epoch is
1347 # also stamped on endpoint-owned objects; object UID/resourceVersion
1348 # preconditions remain the hard fence if a process resumes after losing
1349 # the Lease.
1350 self._lease_name: str | None = None
1351 self._lease_holder: str | None = None
1352 self._leader_epoch: str | None = None
1353 self._leadership_lost = threading.Event()
1354 self._active_authority: ReconcileAuthority | None = None
1356 # Metrics
1357 self._reconcile_count = 0
1358 self._errors_count = 0
1359 # Monotonic stamp of the last completed loop iteration (leader pass or
1360 # standby lease check). Exported as seconds_since_last_pass so a loop
1361 # that is wedged — not crashed, which the probes would catch — is
1362 # visible; initialised at construction so the gauge exists before the
1363 # first iteration and grows if the loop never starts.
1364 self._last_loop_completed_at = time.monotonic()
1366 # ------------------------------------------------------------------
1367 # Reconciliation loop
1368 # ------------------------------------------------------------------
1370 async def start(self) -> None:
1371 """Start reconciliation under an independently renewed leader Lease."""
1372 if self._running:
1373 logger.warning("Inference monitor already running")
1374 return
1375 self._running = True
1376 logger.info(
1377 "Starting inference monitor for %s in %s (interval=%ds)",
1378 self.cluster_id,
1379 self.region,
1380 self.reconcile_interval,
1381 )
1383 pod_name = os.environ.get("HOSTNAME", f"monitor-{id(self)}")
1384 lease_name = "inference-monitor-leader"
1385 self._lease_name = lease_name
1386 self._lease_holder = pod_name
1388 while self._running:
1389 try:
1390 if self._try_acquire_lease(lease_name, pod_name):
1391 with self._renewing_leadership():
1392 await self.reconcile()
1393 else:
1394 logger.debug("Not the leader, waiting...")
1395 except ReconcileFencedError as error:
1396 logger.warning("Reconciliation stopped after authority loss: %s", error)
1397 except Exception as error:
1398 logger.error("Reconciliation error: %s", error, exc_info=True)
1399 self._errors_count += 1
1400 self._last_loop_completed_at = time.monotonic()
1401 try:
1402 await asyncio.sleep(self.reconcile_interval)
1403 except Exception as error:
1404 logger.error("Sleep interrupted: %s", error)
1405 break
1407 @staticmethod
1408 def _lease_annotations(lease: Any) -> dict[str, str]:
1409 metadata = getattr(lease, "metadata", None)
1410 annotations = getattr(metadata, "annotations", None)
1411 return dict(annotations) if isinstance(annotations, dict) else {}
1413 def _lease_is_expired(self, lease: Any, now: datetime) -> bool:
1414 renew_time = getattr(getattr(lease, "spec", None), "renew_time", None)
1415 if renew_time is None:
1416 return True
1417 if renew_time.tzinfo is None:
1418 renew_time = renew_time.replace(tzinfo=UTC)
1419 duration = getattr(lease.spec, "lease_duration_seconds", None)
1420 if not isinstance(duration, int) or isinstance(duration, bool) or duration <= 0:
1421 duration = self.reconcile_interval * 3
1422 return bool((now - renew_time).total_seconds() >= int(duration))
1424 def _stamp_lease_epoch(self, lease: Any, epoch: str) -> None:
1425 metadata = getattr(lease, "metadata", None)
1426 if metadata is None:
1427 lease.metadata = client.V1ObjectMeta()
1428 metadata = lease.metadata
1429 annotations = self._lease_annotations(lease)
1430 annotations[_LEADER_EPOCH_ANNOTATION] = epoch
1431 metadata.annotations = annotations
1433 def _try_acquire_lease(self, lease_name: str, holder: str) -> bool:
1434 """Acquire/renew a Lease and establish one stable acquisition epoch."""
1435 coordination_v1 = client.CoordinationV1Api()
1436 now = datetime.now(UTC)
1437 lease_duration = self.reconcile_interval * 3
1438 try:
1439 lease = coordination_v1.read_namespaced_lease(lease_name, self.namespace)
1440 current_holder = lease.spec.holder_identity
1441 expired = self._lease_is_expired(lease, now)
1442 annotations = self._lease_annotations(lease)
1443 observed_epoch = annotations.get(_LEADER_EPOCH_ANNOTATION)
1445 if current_holder == holder and not expired:
1446 # Adopt the persisted epoch after a harmless local restart of
1447 # the loop, but never renew a same-name holder with a different
1448 # in-memory epoch.
1449 if self._leader_epoch is None:
1450 self._leader_epoch = observed_epoch or secrets.token_hex(32)
1451 elif observed_epoch and observed_epoch != self._leader_epoch:
1452 self._leadership_lost.set()
1453 return False
1454 epoch = self._leader_epoch
1455 elif current_holder in (None, "") or expired:
1456 epoch = secrets.token_hex(32)
1457 self._leader_epoch = epoch
1458 lease.spec.holder_identity = holder
1459 lease.spec.acquire_time = now
1460 transitions = getattr(lease.spec, "lease_transitions", None)
1461 lease.spec.lease_transitions = int(transitions or 0) + 1
1462 logger.info("Acquiring leader lease as %s with a new epoch", holder)
1463 else:
1464 return False
1466 lease.spec.renew_time = now
1467 lease.spec.lease_duration_seconds = lease_duration
1468 self._stamp_lease_epoch(lease, epoch)
1469 coordination_v1.replace_namespaced_lease(lease_name, self.namespace, lease)
1470 self._lease_name = lease_name
1471 self._lease_holder = holder
1472 self._leadership_lost.clear()
1473 return True
1474 except ApiException as error:
1475 if error.status == 404:
1476 epoch = secrets.token_hex(32)
1477 lease = client.V1Lease(
1478 metadata=client.V1ObjectMeta(
1479 name=lease_name,
1480 namespace=self.namespace,
1481 annotations={_LEADER_EPOCH_ANNOTATION: epoch},
1482 ),
1483 spec=client.V1LeaseSpec(
1484 holder_identity=holder,
1485 lease_duration_seconds=lease_duration,
1486 acquire_time=now,
1487 renew_time=now,
1488 lease_transitions=0,
1489 ),
1490 )
1491 try:
1492 coordination_v1.create_namespaced_lease(self.namespace, lease)
1493 except ApiException:
1494 return False
1495 self._leader_epoch = epoch
1496 self._lease_name = lease_name
1497 self._lease_holder = holder
1498 self._leadership_lost.clear()
1499 logger.info("Created leader lease as %s", holder)
1500 return True
1501 if error.status == 409:
1502 logger.info("Lost leader Lease optimistic-concurrency race")
1503 self._leadership_lost.set()
1504 return False
1505 logger.warning("Lease check failed: %s", error.reason)
1506 self._leadership_lost.set()
1507 return False
1509 def _renew_current_lease(self) -> bool:
1510 """Renew only the exact holder/epoch currently owned by this process."""
1511 lease_name = self._lease_name
1512 holder = self._lease_holder
1513 epoch = self._leader_epoch
1514 if not all(isinstance(value, str) and value for value in (lease_name, holder, epoch)):
1515 return False
1516 coordination_v1 = client.CoordinationV1Api()
1517 now = datetime.now(UTC)
1518 try:
1519 lease = coordination_v1.read_namespaced_lease(lease_name, self.namespace)
1520 if (
1521 lease.spec.holder_identity != holder
1522 or self._lease_annotations(lease).get(_LEADER_EPOCH_ANNOTATION) != epoch
1523 or self._lease_is_expired(lease, now)
1524 ):
1525 return False
1526 lease.spec.renew_time = now
1527 coordination_v1.replace_namespaced_lease(lease_name, self.namespace, lease)
1528 return True
1529 except Exception:
1530 logger.warning("Leader Lease renewal failed", exc_info=True)
1531 return False
1533 def _lease_renewal_loop(self, stop_event: threading.Event) -> None:
1534 interval = max(1.0, float(self.reconcile_interval))
1535 while not stop_event.wait(interval):
1536 if not self._renew_current_lease():
1537 self._leadership_lost.set()
1538 return
1540 @contextlib.contextmanager
1541 def _renewing_leadership(self) -> Iterator[None]:
1542 """Renew the Lease outside the asyncio loop while one pass is running."""
1543 stop_event = threading.Event()
1544 renewal = threading.Thread(
1545 target=self._lease_renewal_loop,
1546 args=(stop_event,),
1547 name="inference-monitor-lease-renewer",
1548 daemon=True,
1549 )
1550 renewal.start()
1551 try:
1552 yield
1553 finally:
1554 stop_event.set()
1555 renewal.join(timeout=max(1.0, float(self.reconcile_interval)))
1557 def _assert_current_leadership(self) -> None:
1558 """Fail closed before mutation when the persisted Lease epoch changed."""
1559 if self._leadership_lost.is_set():
1560 raise ReconcileFencedError("leader Lease was lost")
1561 # Direct method-level unit tests do not enter start(); production does.
1562 if self._lease_name is None or self._lease_holder is None or self._leader_epoch is None:
1563 return
1564 coordination_v1 = client.CoordinationV1Api()
1565 try:
1566 lease = coordination_v1.read_namespaced_lease(self._lease_name, self.namespace)
1567 except Exception as error:
1568 self._leadership_lost.set()
1569 raise ReconcileFencedError("leader Lease could not be verified") from error
1570 if (
1571 lease.spec.holder_identity != self._lease_holder
1572 or self._lease_annotations(lease).get(_LEADER_EPOCH_ANNOTATION) != self._leader_epoch
1573 or self._lease_is_expired(lease, datetime.now(UTC))
1574 ):
1575 self._leadership_lost.set()
1576 raise ReconcileFencedError("leader Lease holder or epoch changed")
1578 def stop(self) -> None:
1579 """Stop the reconciliation loop and prevent additional mutations."""
1580 self._running = False
1581 self._leadership_lost.set()
1582 logger.info("Inference monitor stopped")
1584 @staticmethod
1585 def _lifecycle_metadata_complete(endpoint: dict[str, Any]) -> bool:
1586 """Return whether an endpoint has complete immutable cleanup metadata."""
1587 lifecycle_id = endpoint.get("lifecycle_id")
1588 cleanup_regions = endpoint.get("cleanup_regions")
1589 region_generations = endpoint.get("region_generations")
1590 return (
1591 isinstance(lifecycle_id, str)
1592 and bool(lifecycle_id)
1593 and isinstance(cleanup_regions, list)
1594 and isinstance(region_generations, dict)
1595 and all(
1596 isinstance(region, str)
1597 and bool(region)
1598 and isinstance(region_generations.get(region), str)
1599 and bool(region_generations[region])
1600 for region in cleanup_regions
1601 )
1602 )
1604 def _status_write_conditions(
1605 self,
1606 endpoint: dict[str, Any],
1607 *,
1608 deleting: bool = False,
1609 ) -> RegionStatusConditions:
1610 """Return lifecycle/generation predicates for one regional status write."""
1611 lifecycle_id = endpoint.get("lifecycle_id")
1612 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1613 return {}
1614 conditions: RegionStatusConditions = {"expected_lifecycle_id": lifecycle_id}
1615 if deleting:
1616 generation = endpoint.get("deletion_generation")
1617 if isinstance(generation, str) and generation:
1618 conditions["expected_deletion_generation"] = generation
1619 return conditions
1620 raw_generations = endpoint.get("region_generations")
1621 generations = raw_generations if isinstance(raw_generations, dict) else {}
1622 region_generation = generations.get(self.region)
1623 if isinstance(region_generation, str) and region_generation:
1624 conditions["expected_region_generation"] = region_generation
1625 return conditions
1627 def _authority_from_endpoint(self, endpoint: dict[str, Any]) -> ReconcileAuthority | None:
1628 """Build object-level provenance for a production endpoint snapshot."""
1629 lifecycle_id = endpoint.get("lifecycle_id")
1630 generations = endpoint.get("region_generations")
1631 region_generation = generations.get(self.region) if isinstance(generations, dict) else None
1632 if not isinstance(lifecycle_id, str) or not lifecycle_id:
1633 if isinstance(endpoint.get("updated_at"), str):
1634 raise ReconcileFencedError("endpoint snapshot has no lifecycle authority")
1635 return None # Compatibility for isolated method-level fixtures only.
1636 if not isinstance(region_generation, str) or not region_generation:
1637 if isinstance(endpoint.get("updated_at"), str):
1638 raise ReconcileFencedError("endpoint snapshot has no Region generation")
1639 return None
1640 desired_state = endpoint.get("desired_state")
1641 targets = endpoint.get("target_regions")
1642 target_regions = targets if isinstance(targets, list) else []
1643 deletion_generation = endpoint.get("deletion_generation")
1644 return ReconcileAuthority(
1645 endpoint_name=str(endpoint.get("endpoint_name", "")),
1646 lifecycle_id=lifecycle_id,
1647 region_generation=region_generation,
1648 leader_epoch=self._leader_epoch or f"direct-{lifecycle_id[:16]}",
1649 deletion_generation=(
1650 deletion_generation
1651 if isinstance(deletion_generation, str) and deletion_generation
1652 else None
1653 ),
1654 deleting=desired_state == "deleted",
1655 region_removed=desired_state != "deleted" and self.region not in target_regions,
1656 )
1658 def _strong_authority_matches(self, authority: ReconcileAuthority) -> bool:
1659 """Re-read DynamoDB before claiming legacy/previous-epoch objects."""
1660 try:
1661 latest = self.store.get_endpoint(authority.endpoint_name, consistent_read=True)
1662 except AttributeError, TypeError:
1663 return getattr(self, "_lease_name", None) is None
1664 if not isinstance(latest, dict):
1665 return getattr(self, "_lease_name", None) is None
1666 if latest.get("lifecycle_id") != authority.lifecycle_id:
1667 return False
1668 if authority.deleting:
1669 return (
1670 latest.get("desired_state") == "deleted"
1671 and latest.get("deletion_generation") == authority.deletion_generation
1672 )
1673 generations = latest.get("region_generations")
1674 return (
1675 latest.get("desired_state") != "deleted"
1676 and isinstance(generations, dict)
1677 and generations.get(self.region) == authority.region_generation
1678 )
1680 @staticmethod
1681 def _object_metadata(resource: Any) -> tuple[Any, dict[str, str], str | None, str | None]:
1682 """Return metadata, annotations, UID, and resourceVersion for typed/dict objects."""
1683 if isinstance(resource, dict):
1684 metadata = resource.get("metadata")
1685 metadata = metadata if isinstance(metadata, dict) else {}
1686 annotations = metadata.get("annotations")
1687 return (
1688 metadata,
1689 dict(annotations) if isinstance(annotations, dict) else {},
1690 metadata.get("uid") if isinstance(metadata.get("uid"), str) else None,
1691 (
1692 metadata.get("resourceVersion")
1693 if isinstance(metadata.get("resourceVersion"), str)
1694 else None
1695 ),
1696 )
1697 metadata = getattr(resource, "metadata", None)
1698 annotations = getattr(metadata, "annotations", None)
1699 uid = getattr(metadata, "uid", None)
1700 resource_version = getattr(metadata, "resource_version", None)
1701 return (
1702 metadata,
1703 dict(annotations) if isinstance(annotations, dict) else {},
1704 uid if isinstance(uid, str) else None,
1705 resource_version if isinstance(resource_version, str) else None,
1706 )
1708 @staticmethod
1709 def _object_labels(resource: Any) -> dict[str, str]:
1710 if isinstance(resource, dict):
1711 metadata = resource.get("metadata")
1712 labels = metadata.get("labels") if isinstance(metadata, dict) else None
1713 else:
1714 metadata = getattr(resource, "metadata", None)
1715 labels = getattr(metadata, "labels", None)
1716 return dict(labels) if isinstance(labels, dict) else {}
1718 @classmethod
1719 def _has_monitor_provenance(cls, resource: Any) -> bool:
1720 labels = cls._object_labels(resource)
1721 return labels.get("project") == "gco" and labels.get("gco.io/type") == "inference"
1723 def _handoff_stale_resource(
1724 self,
1725 resource: Any,
1726 *,
1727 kind: str,
1728 resource_name: str,
1729 delete_resource: Callable[..., Any] | None,
1730 reason: str,
1731 ) -> NoReturn:
1732 """UID-delete a proven stale monitor object only for current DDB authority."""
1733 authority = getattr(self, "_active_authority", None)
1734 if (
1735 authority is None
1736 or delete_resource is None
1737 or not self._has_monitor_provenance(resource)
1738 or not self._strong_authority_matches(authority)
1739 ):
1740 raise ReconcileFencedError(f"{kind} {resource_name} {reason}")
1741 self._assert_current_leadership()
1742 try:
1743 delete_resource(
1744 body=self._delete_options_for(
1745 resource,
1746 kind=kind,
1747 resource_name=resource_name,
1748 )
1749 )
1750 except ApiException as error:
1751 if error.status not in {404, 409}:
1752 raise
1753 raise ReconcileFencedError(
1754 f"{kind} {resource_name} stale authority handoff deletion requested"
1755 )
1757 def _authorize_resource(
1758 self,
1759 resource: Any,
1760 *,
1761 kind: str,
1762 resource_name: str,
1763 patch_metadata: Callable[..., Any] | None = None,
1764 read_resource: Callable[[], Any] | None = None,
1765 delete_resource: Callable[..., Any] | None = None,
1766 allow_region_mismatch: bool = False,
1767 ) -> Any:
1768 """Verify lifecycle provenance and CAS-claim the current leader epoch."""
1769 authority = getattr(self, "_active_authority", None)
1770 if authority is None:
1771 return resource
1772 self._assert_current_leadership()
1773 _metadata, annotations, _uid, resource_version = self._object_metadata(resource)
1774 if getattr(self, "_lease_name", None) is None and resource_version is None:
1775 # Historical method-level fixtures use metadata-less MagicMocks.
1776 # Production reconciliation always has a Lease and real metadata.
1777 return resource
1778 observed_lifecycle = annotations.get(_LIFECYCLE_ANNOTATION)
1779 if observed_lifecycle is None and not self._has_monitor_provenance(resource):
1780 raise ReconcileFencedError(f"{kind} {resource_name} has ambiguous legacy ownership")
1781 if observed_lifecycle not in (None, authority.lifecycle_id):
1782 self._handoff_stale_resource(
1783 resource,
1784 kind=kind,
1785 resource_name=resource_name,
1786 delete_resource=delete_resource,
1787 reason="belongs to another endpoint lifecycle",
1788 )
1789 observed_region = annotations.get(_REGION_GENERATION_ANNOTATION)
1790 if not allow_region_mismatch and observed_region not in (None, authority.region_generation):
1791 self._handoff_stale_resource(
1792 resource,
1793 kind=kind,
1794 resource_name=resource_name,
1795 delete_resource=delete_resource,
1796 reason="belongs to another Region generation",
1797 )
1798 expected = authority.annotations
1799 if all(annotations.get(key) == value for key, value in expected.items()):
1800 return resource
1801 if patch_metadata is None or read_resource is None:
1802 raise ReconcileFencedError(f"{kind} {resource_name} lacks current immutable provenance")
1803 if not isinstance(resource_version, str) or not resource_version:
1804 raise ReconcileFencedError(
1805 f"{kind} {resource_name} has no resourceVersion for authority claim"
1806 )
1807 if not self._strong_authority_matches(authority):
1808 raise ReconcileFencedError("endpoint authority changed before Kubernetes mutation")
1809 self._assert_current_leadership()
1810 merged = dict(annotations)
1811 merged.update(expected)
1812 try:
1813 patch_metadata(
1814 body={
1815 "metadata": {
1816 "resourceVersion": resource_version,
1817 "annotations": merged,
1818 }
1819 }
1820 )
1821 except Exception as error:
1822 raise ReconcileFencedError(
1823 f"{kind} {resource_name} changed during authority claim"
1824 ) from error
1825 claimed = read_resource()
1826 _metadata, claimed_annotations, _claimed_uid, _claimed_version = self._object_metadata(
1827 claimed
1828 )
1829 if not all(claimed_annotations.get(key) == value for key, value in expected.items()):
1830 raise ReconcileFencedError(
1831 f"{kind} {resource_name} authority claim could not be verified"
1832 )
1833 return claimed
1835 def _provenance_annotations(self) -> dict[str, str] | None:
1836 authority = getattr(self, "_active_authority", None)
1837 return dict(authority.annotations) if authority is not None else None
1839 def _assert_mutation_authority(self) -> None:
1840 authority = getattr(self, "_active_authority", None)
1841 if authority is None:
1842 return
1843 self._assert_current_leadership()
1844 if not self._strong_authority_matches(authority):
1845 raise ReconcileFencedError("endpoint authority changed before Kubernetes mutation")
1847 def _confirm_created_resource(
1848 self,
1849 *,
1850 kind: str,
1851 resource_name: str,
1852 read_resource: Callable[[], Any],
1853 delete_resource: Callable[..., Any],
1854 ) -> Any:
1855 """Compensate an exact just-created object if Lease/DDB authority changed."""
1856 authority = getattr(self, "_active_authority", None)
1857 if authority is None:
1858 return None
1859 try:
1860 created = read_resource()
1861 except ApiException as error:
1862 if getattr(self, "_lease_name", None) is None and error.status == 404:
1863 return None
1864 raise
1865 _metadata, annotations, _uid, resource_version = self._object_metadata(created)
1866 if getattr(self, "_lease_name", None) is None and resource_version is None:
1867 return created
1868 if not all(annotations.get(key) == value for key, value in authority.annotations.items()):
1869 raise ReconcileFencedError(f"{kind} {resource_name} post-create provenance changed")
1870 try:
1871 self._assert_mutation_authority()
1872 except ReconcileFencedError:
1873 # This process created the exact annotated UID. Remove only that UID;
1874 # a replacement racing into the name makes the precondition fail and
1875 # is never touched.
1876 try:
1877 delete_resource(
1878 body=self._delete_options_for(
1879 created,
1880 kind=kind,
1881 resource_name=resource_name,
1882 )
1883 )
1884 except ApiException as error:
1885 if error.status not in {404, 409}:
1886 logger.warning(
1887 "Post-create compensation failed for %s/%s: status %s",
1888 kind,
1889 resource_name,
1890 error.status,
1891 )
1892 raise
1893 return created
1895 def _delete_options_for(self, resource: Any, *, kind: str, resource_name: str) -> Any:
1896 """Build UID/resourceVersion delete preconditions for one authorized object."""
1897 _metadata, _annotations, uid, resource_version = self._object_metadata(resource)
1898 if uid and resource_version:
1899 return client.V1DeleteOptions(
1900 propagation_policy="Foreground",
1901 preconditions=client.V1Preconditions(
1902 uid=uid,
1903 resource_version=resource_version,
1904 ),
1905 )
1906 if (
1907 getattr(self, "_active_authority", None) is None
1908 or getattr(self, "_lease_name", None) is None
1909 ):
1910 return client.V1DeleteOptions(
1911 propagation_policy="Foreground",
1912 preconditions=client.V1Preconditions(uid=uid) if uid else None,
1913 )
1914 raise ReconcileFencedError(
1915 f"{kind} {resource_name} lacks UID/resourceVersion delete authority"
1916 )
1918 async def reconcile(self) -> list[dict[str, Any]]:
1919 """Run one reconciliation cycle with generation-fenced deletion."""
1920 self._reconcile_count += 1
1921 actions: list[dict[str, Any]] = []
1922 try:
1923 endpoints = self.store.list_endpoints()
1924 except Exception as e:
1925 logger.error("Failed to list endpoints from DynamoDB: %s", e)
1926 return actions
1928 # Records written before lifecycle fencing are upgraded from their
1929 # DynamoDB snapshot before any Kubernetes mutation. The update is
1930 # conditional on ``updated_at``; a concurrent writer wins and this pass
1931 # simply retries from the next scan. Direct method-level test fixtures
1932 # without a persistence timestamp remain outside this production path.
1933 normalized_endpoints: list[dict[str, Any]] = []
1934 for stored in endpoints:
1935 endpoint = stored
1936 if not self._lifecycle_metadata_complete(stored) and isinstance(
1937 stored.get("updated_at"), str
1938 ):
1939 upgraded = self.store.ensure_lifecycle_metadata(stored)
1940 if not isinstance(upgraded, dict):
1941 continue
1942 endpoint = upgraded
1943 actions.append(
1944 {
1945 "action": "initialize_lifecycle",
1946 "endpoint": endpoint.get("endpoint_name", "unknown"),
1947 }
1948 )
1949 normalized_endpoints.append(endpoint)
1950 endpoints = normalized_endpoints
1952 for endpoint in endpoints:
1953 name = endpoint.get("endpoint_name", "unknown")
1954 try:
1955 action = await self._reconcile_endpoint(endpoint)
1956 if action:
1957 actions.append(action)
1958 except ReconcileFencedError as error:
1959 # Authority loss is not endpoint health. Another leader may
1960 # already have written the terminal acknowledgement; a stale
1961 # error write must never regress that quorum to ``error``.
1962 logger.warning("Stopped stale reconciliation for %s: %s", name, error)
1963 continue
1964 except Exception as e:
1965 logger.error("Failed to reconcile endpoint %s: %s", name, e)
1966 self._errors_count += 1
1967 status_conditions = self._status_write_conditions(
1968 endpoint,
1969 deleting=endpoint.get("desired_state") == "deleted",
1970 )
1971 self.store.update_region_status(
1972 name,
1973 self.region,
1974 "error",
1975 error=str(e),
1976 **status_conditions,
1977 )
1979 # Any monitor may purge, but only from a strong snapshot containing a
1980 # fresh terminal acknowledgement for every immutable deletion member.
1981 # Each acknowledgement itself represents two child-complete inventory
1982 # sweeps, and the delete is conditioned on the same lifecycle,
1983 # generation, and updated_at snapshot.
1984 for endpoint in endpoints:
1985 if endpoint.get("desired_state") != "deleted":
1986 continue
1987 ep_name = endpoint.get("endpoint_name")
1988 if not isinstance(ep_name, str):
1989 continue
1990 try:
1991 latest = self.store.get_endpoint(ep_name, consistent_read=True)
1992 except Exception as e:
1993 logger.warning("Failed to refresh deleted endpoint %s: %s", ep_name, e)
1994 continue
1995 if not isinstance(latest, dict) or latest.get("desired_state") != "deleted":
1996 continue
1997 lifecycle_id = latest.get("lifecycle_id")
1998 generation = latest.get("deletion_generation")
1999 deletion_regions = latest.get("deletion_regions")
2000 updated_at = latest.get("updated_at")
2001 if not all(
2002 isinstance(value, str) and value for value in (lifecycle_id, generation, updated_at)
2003 ):
2004 continue
2005 if not isinstance(deletion_regions, list) or not deletion_regions:
2006 continue
2007 cleanup_regions = {
2008 region for region in deletion_regions if isinstance(region, str) and region
2009 }
2010 if len(cleanup_regions) != len(deletion_regions):
2011 continue
2012 raw_status = latest.get("region_status")
2013 region_status = raw_status if isinstance(raw_status, dict) else {}
2014 if not all(
2015 isinstance(region_status.get(region), dict)
2016 and region_status[region].get("state") == "deleted"
2017 and region_status[region].get("lifecycle_id") == lifecycle_id
2018 and region_status[region].get("deletion_generation") == generation
2019 and region_status[region].get("absence_observations", 0) >= 2
2020 for region in cleanup_regions
2021 ):
2022 continue
2023 try:
2024 deleted = self.store.delete_endpoint(
2025 ep_name,
2026 expected_updated_at=updated_at,
2027 expected_lifecycle_id=lifecycle_id,
2028 expected_deletion_generation=generation,
2029 )
2030 if not deleted:
2031 continue
2032 logger.info(
2033 "Purged endpoint %s lifecycle %s generation %s",
2034 ep_name,
2035 lifecycle_id,
2036 generation,
2037 )
2038 actions.append({"action": "purge", "endpoint": ep_name})
2039 except Exception as e:
2040 logger.warning("Failed to purge endpoint %s: %s", ep_name, e)
2041 return actions
2043 @staticmethod
2044 def _cleanup_ack_is_terminal(endpoint: dict[str, Any], region: str) -> bool:
2045 """Return whether this region already durably acknowledged this generation."""
2046 raw_statuses = endpoint.get("region_status")
2047 statuses = raw_statuses if isinstance(raw_statuses, dict) else {}
2048 status = statuses.get(region)
2049 if not isinstance(status, dict) or status.get("state") != "deleted":
2050 return False
2051 lifecycle_id = endpoint.get("lifecycle_id")
2052 if status.get("lifecycle_id") != lifecycle_id:
2053 return False
2054 if endpoint.get("desired_state") == "deleted":
2055 return (
2056 status.get("deletion_generation") == endpoint.get("deletion_generation")
2057 and status.get("absence_observations", 0) >= 2
2058 )
2059 raw_generations = endpoint.get("region_generations")
2060 generations = raw_generations if isinstance(raw_generations, dict) else {}
2061 return (
2062 status.get("region_generation") == generations.get(region)
2063 and status.get("absence_observations", 0) >= 2
2064 )
2066 def _record_cleanup_observation(
2067 self,
2068 endpoint: dict[str, Any],
2069 cleanup: ResourceCleanupResult,
2070 ) -> tuple[str, bool]:
2071 """Persist one stable-absence observation for the active lifecycle/generation."""
2072 lifecycle_id = endpoint.get("lifecycle_id")
2073 if not isinstance(lifecycle_id, str) or not lifecycle_id:
2074 raise RuntimeError("Endpoint cleanup requires an immutable lifecycle id")
2075 deleting = endpoint.get("desired_state") == "deleted"
2076 generation = endpoint.get("deletion_generation") if deleting else None
2077 raw_region_generations = endpoint.get("region_generations")
2078 region_generations = (
2079 raw_region_generations if isinstance(raw_region_generations, dict) else {}
2080 )
2081 region_generation = region_generations.get(self.region) if not deleting else None
2082 if deleting and (not isinstance(generation, str) or not generation):
2083 raise RuntimeError("Endpoint deletion requires an immutable deletion generation")
2084 if not deleting and (not isinstance(region_generation, str) or not region_generation):
2085 raise RuntimeError("Endpoint cleanup requires a current Region generation")
2087 raw_statuses = endpoint.get("region_status")
2088 statuses = raw_statuses if isinstance(raw_statuses, dict) else {}
2089 previous = statuses.get(self.region)
2090 same_generation = (
2091 isinstance(previous, dict)
2092 and previous.get("lifecycle_id") == lifecycle_id
2093 and (
2094 previous.get("deletion_generation") == generation
2095 if deleting
2096 else previous.get("region_generation") == region_generation
2097 )
2098 )
2099 previous_observations = (
2100 int(previous.get("absence_observations", 0))
2101 if same_generation and isinstance(previous, dict)
2102 else 0
2103 )
2104 observations = previous_observations + 1 if cleanup.complete else 0
2105 state = "deleted" if observations >= 2 else "deleting"
2106 status_kwargs: dict[str, Any] = {
2107 "extra": {"absence_observations": observations},
2108 "expected_lifecycle_id": lifecycle_id,
2109 }
2110 if deleting:
2111 status_kwargs["expected_deletion_generation"] = generation
2112 else:
2113 status_kwargs["expected_region_generation"] = region_generation
2114 if cleanup.error_message:
2115 status_kwargs["error"] = cleanup.error_message
2116 written = self.store.update_region_status(
2117 endpoint["endpoint_name"],
2118 self.region,
2119 state,
2120 **status_kwargs,
2121 )
2122 return state, bool(written)
2124 async def _reconcile_endpoint(self, endpoint: dict[str, Any]) -> dict[str, Any] | None:
2125 """Reconcile one endpoint under immutable Kubernetes provenance."""
2126 authority = self._authority_from_endpoint(endpoint)
2127 previous = self._active_authority
2128 self._active_authority = authority
2129 try:
2130 return await self._reconcile_endpoint_authorized(endpoint)
2131 finally:
2132 self._active_authority = previous
2134 async def _reconcile_endpoint_authorized(
2135 self, endpoint: dict[str, Any]
2136 ) -> dict[str, Any] | None:
2137 """Implementation for one endpoint after authority has been installed."""
2138 name = endpoint["endpoint_name"]
2139 desired_state = endpoint.get("desired_state", "deploying")
2140 target_regions = endpoint.get("target_regions", [])
2141 spec = endpoint.get("spec", {})
2142 ns = endpoint.get("namespace", self.namespace)
2144 lifecycle_value = endpoint.get("lifecycle_id")
2145 lifecycle_id = (
2146 lifecycle_value if isinstance(lifecycle_value, str) and lifecycle_value else None
2147 )
2149 if desired_state == "deleted":
2150 generation = endpoint.get("deletion_generation")
2151 deletion_regions = endpoint.get("deletion_regions")
2152 if not isinstance(generation, str) or not isinstance(deletion_regions, list):
2153 # Upgrade a legacy or interrupted delete transition atomically;
2154 # the next pass sees the persisted immutable snapshot.
2155 self.store.update_desired_state(
2156 name,
2157 "deleted",
2158 expected_lifecycle_id=lifecycle_id,
2159 )
2160 return {"action": "initialize_deletion", "endpoint": name}
2161 if self.region not in deletion_regions:
2162 return None
2163 if self._cleanup_ack_is_terminal(endpoint, self.region):
2164 # Durable current-generation acknowledgement makes completed
2165 # non-target work quiescent: no Kubernetes reads and no DDB write.
2166 return None
2167 return self._reconcile_deleted(endpoint, ns, spec if isinstance(spec, dict) else None)
2169 if self.region not in target_regions:
2170 cleanup_regions = endpoint.get("cleanup_regions")
2171 if not isinstance(cleanup_regions, list) or self.region not in cleanup_regions:
2172 return None
2173 if self._cleanup_ack_is_terminal(endpoint, self.region):
2174 return None
2175 cleanup = self._delete_resources(
2176 name,
2177 ns,
2178 spec if isinstance(spec, dict) else None,
2179 expected_lifecycle_id=lifecycle_id,
2180 )
2181 state, written = self._record_cleanup_observation(endpoint, cleanup)
2182 return {
2183 "action": "cleanup",
2184 "endpoint": name,
2185 "reason": "region_removed",
2186 "cleanup_complete": state == "deleted" and written,
2187 }
2189 if desired_state in ("deploying", "running"):
2190 if not isinstance(spec, dict):
2191 error = "endpoint spec must be a mapping"
2192 elif "mooncake" in spec and "canary" in spec:
2193 error = "endpoint spec cannot combine 'mooncake' and 'canary' blocks"
2194 else:
2195 return await self._reconcile_running(name, ns, spec, endpoint)
2196 logger.error("Rejecting invalid endpoint %s: %s", name, error)
2197 self.store.update_region_status(
2198 name,
2199 self.region,
2200 "failed",
2201 error=error,
2202 **self._status_write_conditions(endpoint),
2203 )
2204 return {"action": "reject", "endpoint": name, "reason": "invalid_spec"}
2205 if desired_state == "stopped":
2206 return self._reconcile_stopped(
2207 name,
2208 ns,
2209 spec if isinstance(spec, dict) else None,
2210 endpoint,
2211 )
2212 return None
2214 async def _reconcile_running(
2215 self,
2216 name: str,
2217 namespace: str,
2218 spec: dict[str, Any],
2219 endpoint: dict[str, Any],
2220 ) -> dict[str, Any] | None:
2221 """Ensure the endpoint is running with the correct spec."""
2222 status_conditions = self._status_write_conditions(endpoint)
2223 # Specs carrying a ``mooncake`` block take the disaggregated path. The
2224 # branch returns ``None`` when no such block is present, so a plain
2225 # endpoint falls through to the single-Deployment path below unchanged.
2226 mooncake_action = await self._reconcile_mooncake(name, namespace, spec, endpoint)
2227 if mooncake_action is not None:
2228 return mooncake_action
2230 deployment = self._get_deployment(name, namespace)
2231 configured_replicas = spec.get("replicas", 1)
2232 raw_autoscaling = spec.get("autoscaling", {})
2233 autoscaling = raw_autoscaling if isinstance(raw_autoscaling, dict) else {}
2234 autoscaling_enabled = bool(autoscaling.get("enabled"))
2236 if deployment is None:
2237 # A missing Deployment is still an ownership handoff boundary. Do
2238 # not recreate it until every obsolete HPA/KEDA owner is absent.
2239 if autoscaling_enabled:
2240 ownership = self._reconcile_classic_autoscaler(
2241 name,
2242 namespace,
2243 spec,
2244 apply_desired=False,
2245 )
2246 else:
2247 ownership = self._delete_autoscalers(
2248 (name,),
2249 (name, f"keda-hpa-{name}"),
2250 namespace,
2251 )
2252 if not ownership.complete:
2253 ownership_status_kwargs: dict[str, Any] = {}
2254 if ownership.error_message:
2255 ownership_status_kwargs["error"] = ownership.error_message
2256 self.store.update_region_status(
2257 name,
2258 self.region,
2259 "updating",
2260 replicas_ready=0,
2261 replicas_desired=0,
2262 **status_conditions,
2263 **ownership_status_kwargs,
2264 )
2265 return {
2266 "action": "reconcile_autoscaler",
2267 "endpoint": name,
2268 "cleanup_complete": False,
2269 }
2271 logger.info("Creating endpoint %s in %s", name, self.region)
2272 self._ensure_tls_proxy_configmap(name, namespace)
2273 self._create_deployment(name, namespace, spec)
2274 self._create_service(name, namespace, spec)
2275 if autoscaling_enabled:
2276 self._create_or_update_hpa(name, namespace, spec)
2277 self.store.update_region_status(
2278 name,
2279 self.region,
2280 "creating",
2281 replicas_desired=(
2282 int(autoscaling.get("min_replicas", 1))
2283 if autoscaling_enabled
2284 else configured_replicas
2285 ),
2286 **status_conditions,
2287 )
2288 return {"action": "create", "endpoint": name}
2290 # Deployment exists — ensure its Service exists. Public traffic follows
2291 # ``gco-system/gco-gateway``'s shared ``/inference`` HTTPRoute to
2292 # ``gco-system/inference-proxy``, which then reaches this endpoint's
2293 # ClusterIP Service on its TLS sidecar port.
2294 self._ensure_service(name, namespace, spec)
2296 # Endpoints created before the TLS sidecar existed gain it here, in
2297 # place; the resulting rollout is observed through ready replicas like
2298 # any other template change, so the pass carries on.
2299 self._ensure_tls_proxy_configmap(name, namespace)
2300 self._converge_endpoint_tls_proxy(
2301 name,
2302 namespace,
2303 deployment,
2304 config_map_name=_tls_proxy_config_map_name(name),
2305 fallback_port=spec.get("port", 8000),
2306 )
2308 # Once enabled, HPA/KEDA is the sole owner of Deployment
2309 # ``spec.replicas``; the static endpoint count must never fight it.
2310 # Controller creation/handoff runs after capturing live replica status
2311 # below so a pending ownership transition can report useful progress.
2313 observed_desired = getattr(deployment.spec, "replicas", None)
2314 live_desired_replicas = (
2315 int(observed_desired) if observed_desired is not None else configured_replicas
2316 )
2317 current_replicas = live_desired_replicas
2318 ready_replicas = int(getattr(deployment.status, "ready_replicas", 0) or 0)
2319 # During enabled -> disabled handoff the old owner still controls this
2320 # observed value. Report it until static reconciliation actually takes
2321 # ownership, rather than publishing an aspirational configured count.
2322 status_desired_replicas = live_desired_replicas
2323 readiness_floor = (
2324 int(autoscaling.get("min_replicas", 1)) if autoscaling_enabled else configured_replicas
2325 )
2327 self._check_health_watchdog(
2328 name,
2329 namespace,
2330 ready_replicas,
2331 status_desired_replicas,
2332 spec,
2333 endpoint,
2334 )
2336 autoscaler_cleanup: ResourceCleanupResult
2337 if autoscaling_enabled:
2338 # Reconcile on every existing-Deployment pass to repair partial
2339 # creation and configuration drift. Ownership handoffs remain
2340 # updating until the obsolete controller is actually absent.
2341 autoscaler_cleanup = self._reconcile_classic_autoscaler(name, namespace, spec)
2342 else:
2343 # Static ownership is explicit even when the autoscaling block was
2344 # removed entirely: stale HPA/KEDA objects must never regain count.
2345 autoscaler_cleanup = self._delete_autoscalers(
2346 (name,),
2347 (name, f"keda-hpa-{name}"),
2348 namespace,
2349 )
2351 if not autoscaler_cleanup.complete:
2352 status_kwargs: dict[str, Any] = {}
2353 if autoscaler_cleanup.error_message:
2354 status_kwargs["error"] = autoscaler_cleanup.error_message
2355 self.store.update_region_status(
2356 name,
2357 self.region,
2358 "updating",
2359 replicas_ready=ready_replicas,
2360 replicas_desired=status_desired_replicas,
2361 **status_conditions,
2362 **status_kwargs,
2363 )
2364 return {
2365 "action": "reconcile_autoscaler",
2366 "endpoint": name,
2367 "cleanup_complete": False,
2368 }
2370 if not autoscaling_enabled and current_replicas != configured_replicas:
2371 logger.info(
2372 "Scaling endpoint %s: %d → %d replicas",
2373 name,
2374 current_replicas,
2375 configured_replicas,
2376 )
2377 self._scale_deployment(name, namespace, configured_replicas)
2378 self.store.update_region_status(
2379 name,
2380 self.region,
2381 "updating",
2382 replicas_ready=ready_replicas,
2383 replicas_desired=configured_replicas,
2384 **status_conditions,
2385 )
2386 return {"action": "scale", "endpoint": name, "replicas": configured_replicas}
2388 # Check if image changed
2389 current_image = self._get_deployment_image(deployment)
2390 desired_image = self._resolve_image_for_region(spec) if spec.get("image") else ""
2391 if current_image and desired_image and current_image != desired_image:
2392 logger.info("Updating endpoint %s image: %s → %s", name, current_image, desired_image)
2393 self._update_deployment_image(name, namespace, desired_image)
2394 self.store.update_region_status(
2395 name,
2396 self.region,
2397 "updating",
2398 replicas_ready=ready_replicas,
2399 replicas_desired=status_desired_replicas,
2400 **status_conditions,
2401 )
2402 return {"action": "update_image", "endpoint": name, "image": desired_image}
2404 # Reconcile canary first and publish only observed readiness. The
2405 # authenticated proxy will not sample canary traffic until this exact
2406 # region reports the matching image fully Ready.
2407 canary = spec.get("canary")
2408 canary_status = None
2409 if isinstance(canary, dict):
2410 canary_status = self._reconcile_canary(name, namespace, spec, canary, endpoint)
2411 else:
2412 self._cleanup_canary(name, namespace)
2414 # For an autoscaled endpoint, readiness means the configured minimum
2415 # serving capacity is available. During ordinary scale-out the live HPA
2416 # target can temporarily exceed Ready pods without making a healthy
2417 # endpoint flap back to "creating". Keep reporting that live target so
2418 # status and watchdog diagnostics remain truthful.
2419 state = "running" if ready_replicas >= readiness_floor else "creating"
2420 self.store.update_region_status(
2421 name,
2422 self.region,
2423 state,
2424 replicas_ready=ready_replicas,
2425 replicas_desired=status_desired_replicas,
2426 extra={"canary": canary_status} if canary_status is not None else None,
2427 **status_conditions,
2428 )
2430 self._promote_deploying_endpoint(name, endpoint, state)
2432 return None
2434 def _promote_deploying_endpoint(
2435 self, name: str, endpoint: dict[str, Any], local_state: str
2436 ) -> bool:
2437 """Promote ``desired_state`` from ``deploying`` to ``running`` once every region is up.
2439 The inference proxy serves an endpoint only while its ``desired_state``
2440 is ``running``, so a deploying endpoint that never promotes is
2441 unreachable even when every pod is Ready. Promotion is decided from live
2442 local readiness (``local_state``, the state this pass just wrote) plus
2443 explicit ``running`` observations for every *other* target region. The
2444 endpoint object may contain a stale local region_status from before this
2445 pass, which is why the local region is never read back from it.
2447 The write is fenced on the lifecycle id and the ``deploying`` state so a
2448 concurrent delete or replacement cannot be resurrected. Both the classic
2449 single-Deployment path and the Mooncake role-keyed path use this.
2451 Returns:
2452 ``True`` when a promotion write was issued, ``False`` otherwise.
2453 """
2454 if local_state != "running" or endpoint.get("desired_state") != "deploying":
2455 return False
2456 stored_statuses = endpoint.get("region_status", {})
2457 target_regions = endpoint.get("target_regions", [])
2458 if not target_regions:
2459 return False
2460 for target_region in target_regions:
2461 if target_region == self.region:
2462 continue
2463 target_status = (
2464 stored_statuses.get(target_region, {}) if isinstance(stored_statuses, dict) else {}
2465 )
2466 if not isinstance(target_status, dict) or target_status.get("state") != "running":
2467 return False
2468 lifecycle_id = endpoint.get("lifecycle_id")
2469 if not isinstance(lifecycle_id, str) or not lifecycle_id:
2470 return False
2471 self.store.update_desired_state(
2472 name,
2473 "running",
2474 expected_lifecycle_id=lifecycle_id,
2475 expected_desired_state="deploying",
2476 )
2477 return True
2479 # ------------------------------------------------------------------
2480 # Mooncake reconciliation branch
2481 # ------------------------------------------------------------------
2483 @staticmethod
2484 def _desired_roles(mode: str | None) -> list[str]:
2485 """Return the worker roles a mode materializes, in a stable order.
2487 Disaggregated and ``both`` modes split work across ``prefill`` then
2488 ``decode``; store mode runs a single ``kv_both`` instance under the
2489 ``single`` role. The order is fixed so role creation and status
2490 reporting are deterministic across passes.
2491 """
2492 roles = _WORKER_ROLES_BY_MODE.get(mode, set()) if isinstance(mode, str) else set()
2493 return [role for role in ("prefill", "decode", "single") if role in roles]
2495 @staticmethod
2496 def _needs_shared_master(mooncake: dict[str, Any]) -> bool:
2497 """Whether the endpoint depends on the shared per-region master.
2499 The store-bearing modes (``store`` and ``both``) always reach the
2500 master for KV metadata, and any endpoint transferring over RDMA reaches
2501 the master's built-in metadata server for the connector handshake. A
2502 disaggregated endpoint transferring over TCP needs no master.
2503 """
2504 mode = mooncake.get("mode")
2505 if mode in ("store", "both"):
2506 return True
2507 transfer = mooncake.get("transfer") or {}
2508 return bool(transfer.get("protocol", "rdma") == "rdma")
2510 def _ensure_mooncake_configmap(self, name: str, ns: str, cfg: dict[str, Any]) -> None:
2511 """Create or update the shared transport ConfigMap for an endpoint.
2513 The rendered transport settings (the dict produced by
2514 :func:`render_mooncake_config`) are written to a ConfigMap named
2515 ``{name}-mooncake`` under the ``mooncake.json`` key, which each role pod
2516 mounts at the configured path. Creation is idempotent: an existing
2517 ConfigMap is patched to the desired contents so a transport change on
2518 the spec propagates on the next pass.
2519 """
2520 cm_name = f"{name}-mooncake"
2521 config_map = client.V1ConfigMap(
2522 metadata=client.V1ObjectMeta(
2523 name=cm_name,
2524 namespace=ns,
2525 labels={"app": name, "project": "gco", "gco.io/type": "inference"},
2526 annotations=self._provenance_annotations(),
2527 ),
2528 data={"mooncake.json": json.dumps(cfg, sort_keys=True)},
2529 )
2530 self._assert_mutation_authority()
2531 try:
2532 self.core_v1.create_namespaced_config_map(
2533 ns, config_map, _request_timeout=self._k8s_timeout
2534 )
2535 self._confirm_created_resource(
2536 kind="configmap",
2537 resource_name=cm_name,
2538 read_resource=partial(
2539 self.core_v1.read_namespaced_config_map,
2540 cm_name,
2541 ns,
2542 _request_timeout=self._k8s_timeout,
2543 ),
2544 delete_resource=partial(
2545 self.core_v1.delete_namespaced_config_map,
2546 cm_name,
2547 ns,
2548 _request_timeout=self._k8s_timeout,
2549 ),
2550 )
2551 logger.info("Created mooncake config map %s/%s", ns, cm_name)
2552 except ApiException as error:
2553 if error.status != 409:
2554 raise
2555 existing = self.core_v1.read_namespaced_config_map(
2556 cm_name, ns, _request_timeout=self._k8s_timeout
2557 )
2558 existing = self._authorize_resource(
2559 existing,
2560 kind="configmap",
2561 resource_name=cm_name,
2562 patch_metadata=partial(
2563 self.core_v1.patch_namespaced_config_map,
2564 cm_name,
2565 ns,
2566 _request_timeout=self._k8s_timeout,
2567 ),
2568 read_resource=lambda: self.core_v1.read_namespaced_config_map(
2569 cm_name, ns, _request_timeout=self._k8s_timeout
2570 ),
2571 delete_resource=partial(
2572 self.core_v1.delete_namespaced_config_map,
2573 cm_name,
2574 ns,
2575 _request_timeout=self._k8s_timeout,
2576 ),
2577 )
2578 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
2579 config_map.metadata.resource_version = resource_version
2580 self._assert_mutation_authority()
2581 self.core_v1.patch_namespaced_config_map(
2582 cm_name, ns, config_map, _request_timeout=self._k8s_timeout
2583 )
2584 logger.info("Updated mooncake config map %s/%s", ns, cm_name)
2586 def _ensure_role_deployment(
2587 self,
2588 name: str,
2589 ns: str,
2590 spec: dict[str, Any],
2591 role: str,
2592 *,
2593 config_digest: str | None = None,
2594 ) -> tuple[int, int, bool]:
2595 """Ensure one role target has its static or restart-seed capacity.
2597 Existing autoscaled targets normally retain controller ownership, but a
2598 manually stopped zero target is seeded to the role minimum before its
2599 autoscaler is reapplied. Static roles always converge to topology.
2601 Args:
2602 config_digest: Digest of the ``mooncake.json`` the role must run
2603 with (see :func:`mooncake_config_digest`). A new Deployment is
2604 stamped with it; an existing one whose pod template carries a
2605 different digest is patched so its pods roll onto the
2606 re-rendered configuration. ``None`` leaves the template alone.
2607 """
2608 mooncake = spec.get("mooncake") or {}
2609 deploy_name = name if role == "single" else f"{name}-{role}"
2610 desired = self._replica_count_for_role(mooncake, role)
2612 deployment = self._get_deployment(deploy_name, ns)
2613 if deployment is None:
2614 self._create_role_deployment(name, ns, spec, role, config_digest=config_digest)
2615 return 0, desired, False
2617 # Role Deployments created before the TLS sidecar existed gain it in
2618 # place (the PD proxy and kube-proxy only reach the role through it).
2619 sidecar_patched = self._converge_endpoint_tls_proxy(
2620 deploy_name,
2621 ns,
2622 deployment,
2623 config_map_name=_tls_proxy_config_map_name(name),
2624 fallback_port=spec.get("port", 8000),
2625 )
2626 # One compare-and-swap per pass: the sidecar patch above already moved
2627 # the resourceVersion, so a config digest change waits for the next pass.
2628 if config_digest is not None and not sidecar_patched:
2629 self._converge_mooncake_config_digest(deploy_name, ns, deployment, config_digest)
2631 autoscaling = mooncake.get("autoscaling") or {}
2632 role_autoscaling = autoscaling.get(role)
2633 autoscaled = (
2634 bool(autoscaling.get("enabled"))
2635 and role in ("prefill", "decode")
2636 and isinstance(role_autoscaling, dict)
2637 )
2638 current = deployment.spec.replicas or 0
2639 restarted = autoscaled and current == 0
2640 if (not autoscaled and current != desired) or restarted:
2641 logger.info(
2642 "Scaling role deployment %s/%s: %d → %d",
2643 ns,
2644 deploy_name,
2645 current,
2646 desired,
2647 )
2648 self._scale_deployment(deploy_name, ns, desired)
2650 status = getattr(deployment, "status", None)
2651 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2652 return ready, desired, restarted
2654 def _report_role_status(
2655 self,
2656 name: str,
2657 ns: str,
2658 mooncake: dict[str, Any],
2659 region_services: dict[str, Any],
2660 endpoint: dict[str, Any] | None = None,
2661 *,
2662 cold_tier_error: str | None = None,
2663 ) -> str:
2664 """Write the role-keyed region status for a Mooncake endpoint.
2666 For split topologies the status carries a ``roles`` map of observed and
2667 desired replica counts per role; for store-bearing endpoints it carries
2668 a ``store`` sub-status with the master's readiness and address. The flat
2669 ``replicas_ready`` / ``replicas_desired`` fields are also populated with
2670 the totals so consumers that only read the flat shape still see motion.
2672 When the endpoint asked for the cold tier, the ``store`` sub-status also
2673 reports whether it is active (``cold_tier.enabled``) and the resolved
2674 ``s3_uri``; an unresolved own-region bucket surfaces as
2675 ``cold_tier.enabled: false`` with ``cold_tier_error`` as the reason, so
2676 ``gco inference status`` shows that the hot-path store is serving
2677 without its S3 tier instead of hiding the downgrade.
2679 Args:
2680 cold_tier_error: Why the cold tier was dropped this pass, or
2681 ``None`` when it resolved (or was not requested).
2683 Returns:
2684 The reported endpoint state: ``"running"`` once every desired role
2685 replica (and, when applicable, the master) is Ready, otherwise
2686 ``"creating"``.
2687 """
2688 mode = mooncake.get("mode")
2689 roles = self._desired_roles(mode)
2690 extra: dict[str, Any] = {}
2691 total_ready = 0
2692 total_desired = 0
2693 all_ready = True
2695 if mode in ("disaggregated", "both"):
2696 roles_block: dict[str, Any] = {}
2697 for role in ("prefill", "decode"):
2698 if role not in roles:
2699 continue
2700 deploy_name = f"{name}-{role}"
2701 configured_floor = self._replica_count_for_role(mooncake, role)
2702 dep = self._get_deployment(deploy_name, ns)
2703 status = getattr(dep, "status", None) if dep else None
2704 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2705 deployment_spec = getattr(dep, "spec", None) if dep else None
2706 observed_desired = getattr(deployment_spec, "replicas", None)
2707 desired = (
2708 int(observed_desired) if observed_desired is not None else configured_floor
2709 )
2710 roles_block[role] = {"ready": ready, "desired": desired}
2711 total_ready += ready
2712 total_desired += desired
2713 if ready < configured_floor:
2714 all_ready = False
2715 extra["roles"] = roles_block
2716 else:
2717 # Store mode runs a single kv_both Deployment under the endpoint name.
2718 configured_floor = self._replica_count_for_role(mooncake, "single")
2719 dep = self._get_deployment(name, ns)
2720 status = getattr(dep, "status", None) if dep else None
2721 ready = int(getattr(status, "ready_replicas", 0) or 0) if status else 0
2722 deployment_spec = getattr(dep, "spec", None) if dep else None
2723 observed_desired = getattr(deployment_spec, "replicas", None)
2724 desired = int(observed_desired) if observed_desired is not None else configured_floor
2725 total_ready += ready
2726 total_desired += desired
2727 if ready < configured_floor:
2728 all_ready = False
2730 store = mooncake.get("store") or {}
2731 if store.get("enabled"):
2732 master_ready = self._mooncake_master_ready_replicas(ns) >= 1
2733 store_status: dict[str, Any] = {
2734 "ready": master_ready,
2735 "master": region_services.get("master_server_address"),
2736 }
2737 if store.get("cold_tier_enabled") is True:
2738 cold_tier_s3_uri = region_services.get("cold_tier_s3_uri")
2739 cold_tier_status: dict[str, Any] = {
2740 "enabled": bool(cold_tier_s3_uri),
2741 "s3_uri": cold_tier_s3_uri,
2742 }
2743 if cold_tier_error:
2744 cold_tier_status["error"] = cold_tier_error
2745 store_status["cold_tier"] = cold_tier_status
2746 extra["store"] = store_status
2747 if not master_ready:
2748 all_ready = False
2750 state = "running" if all_ready and total_desired > 0 else "creating"
2751 status_conditions = self._status_write_conditions(endpoint or {})
2752 self.store.update_region_status(
2753 name,
2754 self.region,
2755 state,
2756 replicas_ready=total_ready,
2757 replicas_desired=total_desired,
2758 extra=extra or None,
2759 **status_conditions,
2760 )
2761 return state
2763 async def _reconcile_mooncake(
2764 self,
2765 name: str,
2766 ns: str,
2767 spec: dict[str, Any],
2768 endpoint: dict[str, Any],
2769 ) -> dict[str, Any] | None:
2770 """Reconcile an endpoint whose spec carries a ``mooncake`` block.
2772 Returns ``None`` when the spec carries no ``mooncake`` block, signalling
2773 the caller to take the single-Deployment path: one Deployment at the
2774 configured replica count and one internal ClusterIP Service, with no role
2775 split, proxy, autoscaler, shared-master dependency, endpoint Ingress,
2776 Gateway, or HTTPRoute. The shared platform route remains unchanged.
2778 With a ``mooncake`` block present, the topology is materialized in
2779 dependency order, and the shared ConfigMap and master are laid down
2780 before any role pod, the roles before the front-end, and the front-end
2781 before status is written:
2783 1. Resolve the in-region addresses (master, metadata, optional cold
2784 tier). When the store is enabled but no own-region master is
2785 configured, nothing further is materialized; the existing
2786 configuration is left unchanged and the endpoint is reported as
2787 still coming up with the unresolved-master reason.
2788 2. Confirm every wired address stays inside the monitor's own region. A
2789 cross-region address fails the endpoint and materializes nothing,
2790 leaving any prior resources in place.
2791 3. Gate dependent pods on the shared per-region master, which also lays
2792 down the intra-namespace allow rules. While the master is not Ready,
2793 or if it could not be created, nothing further is materialized and
2794 the endpoint is reported as still coming up.
2795 4. Render and apply the shared transport ConfigMap.
2796 5. Materialize each role Deployment: prefill and decode for
2797 disaggregated and both modes, a single ``kv_both`` Deployment for
2798 store mode.
2799 6. Materialize each present role's autoscaler when autoscaling is on.
2800 7. Front disaggregated and both modes with the proxy and its internal
2801 ClusterIP Service; give store mode an internal ClusterIP Service.
2802 Public traffic remains on the shared ``gco-system/gco-gateway``
2803 HTTPRoute from ``/inference`` to ``gco-system/inference-proxy``.
2804 8. Write the role-keyed region status.
2806 Returns:
2807 An action record describing what the pass did, or ``None`` when the
2808 spec carries no ``mooncake`` block.
2809 """
2810 mooncake = spec.get("mooncake")
2811 if not mooncake:
2812 return None
2814 status_conditions = self._status_write_conditions(endpoint)
2815 mode = mooncake.get("mode")
2817 # Step 1: resolve in-region addresses. A store without an own-region
2818 # master is left untouched and reported as still coming up.
2819 services = self._resolve_region_services(name, mooncake)
2820 if services.render_skipped:
2821 self.store.update_region_status(
2822 name,
2823 self.region,
2824 "creating",
2825 error=services.error,
2826 **status_conditions,
2827 )
2828 return {
2829 "action": "reconcile_mooncake",
2830 "endpoint": name,
2831 "deferred": "store_master_unresolved",
2832 }
2834 region_services = services.region_services or {}
2836 # Step 2: keep the topology inside its own region.
2837 scope = self._resolve_regional_scope(name, ns, spec, region_services)
2838 if not scope.in_region:
2839 self.store.update_region_status(
2840 name,
2841 self.region,
2842 scope.state or "failed",
2843 error=scope.error,
2844 **status_conditions,
2845 )
2846 return {
2847 "action": "reconcile_mooncake",
2848 "endpoint": name,
2849 "failed": "cross_region_boundary",
2850 }
2852 # Step 3: gate dependent pods on the shared master (and its allow
2853 # rules). This is also where the master itself is created if absent.
2854 if self._needs_shared_master(mooncake):
2855 gate = self._gate_on_mooncake_master(name, ns, spec)
2856 if not gate.proceed:
2857 self.store.update_region_status(
2858 name,
2859 self.region,
2860 gate.state or "creating",
2861 error=gate.error,
2862 **status_conditions,
2863 )
2864 return {
2865 "action": "reconcile_mooncake",
2866 "endpoint": name,
2867 "deferred": "master_not_ready",
2868 }
2870 # Step 4: shared transport ConfigMap, applied once before role pods,
2871 # together with the TLS sidecar program every role and proxy pod runs.
2872 cfg = render_mooncake_config(mooncake, region_services)
2873 self._ensure_mooncake_configmap(name, ns, cfg)
2874 self._ensure_tls_proxy_configmap(name, ns)
2875 # Role pods start with the file they mount, so a re-render must roll
2876 # them (digest stamped on the pod template). A degraded render — the
2877 # cold tier dropped because its bucket did not resolve this pass — is
2878 # not rolled out: that would flap pods on a transient lookup failure.
2879 config_digest = None if services.cold_tier_unresolved else mooncake_config_digest(cfg)
2881 # Steps 5-6: converge ownership before creating/scaling any role
2882 # Deployment. This is the same recreate/handoff barrier used by classic
2883 # endpoints, extended to prefill/decode and stale topology roles.
2884 desired_roles = self._desired_roles(mode)
2885 ownership_results = [
2886 self._delete_autoscalers(
2887 (name,),
2888 (name, f"keda-hpa-{name}"),
2889 ns,
2890 )
2891 ]
2892 for role in ("prefill", "decode"):
2893 role_name = f"{name}-{role}"
2894 if role in desired_roles:
2895 ownership_results.append(
2896 self._reconcile_role_autoscaler(
2897 name,
2898 ns,
2899 spec,
2900 role,
2901 apply_desired=False,
2902 )
2903 )
2904 else:
2905 ownership_results.append(
2906 self._delete_autoscalers(
2907 (role_name,),
2908 (role_name, f"keda-hpa-{role_name}"),
2909 ns,
2910 )
2911 )
2912 ownership = self._merge_cleanup_results(*ownership_results)
2913 if not ownership.complete:
2914 ready_total = 0
2915 desired_total = 0
2916 for role in desired_roles:
2917 deploy_name = name if role == "single" else f"{name}-{role}"
2918 deployment = self._get_deployment(deploy_name, ns)
2919 deployment_status = getattr(deployment, "status", None)
2920 deployment_spec = getattr(deployment, "spec", None)
2921 ready_total += int(getattr(deployment_status, "ready_replicas", 0) or 0)
2922 desired_total += int(getattr(deployment_spec, "replicas", 0) or 0)
2923 status_kwargs: dict[str, Any] = {}
2924 if ownership.error_message:
2925 status_kwargs["error"] = ownership.error_message
2926 self.store.update_region_status(
2927 name,
2928 self.region,
2929 "updating",
2930 replicas_ready=ready_total,
2931 replicas_desired=desired_total,
2932 **status_conditions,
2933 **status_kwargs,
2934 )
2935 return {
2936 "action": "reconcile_mooncake_autoscaler",
2937 "endpoint": name,
2938 "cleanup_complete": False,
2939 }
2941 role_ready_total = 0
2942 role_desired_total = 0
2943 for role in desired_roles:
2944 ready, desired, _restarted = self._ensure_role_deployment(
2945 name, ns, spec, role, config_digest=config_digest
2946 )
2947 role_ready_total += ready
2948 role_desired_total += desired
2950 owner_verifications: list[ResourceCleanupResult] = []
2951 for role in ("prefill", "decode"):
2952 role_config = self._role_autoscaling_config(spec, role)
2953 if role not in desired_roles or role_config is None:
2954 continue
2955 self._create_role_hpa(name, ns, spec, role)
2956 target_name = f"{name}-{role}"
2957 metrics_config = role_config.get("metrics", [{"type": "cpu", "target": 70}])
2958 hpa_name = (
2959 f"keda-hpa-{target_name}"
2960 if self._metrics_require_keda(metrics_config)
2961 else target_name
2962 )
2963 owner_verifications.append(self._verify_hpa_owner(hpa_name, ns, target_name))
2964 verified_owners = self._merge_cleanup_results(*owner_verifications)
2965 if not verified_owners.complete:
2966 verification_status_kwargs: dict[str, Any] = {}
2967 if verified_owners.error_message:
2968 verification_status_kwargs["error"] = verified_owners.error_message
2969 self.store.update_region_status(
2970 name,
2971 self.region,
2972 "updating",
2973 replicas_ready=role_ready_total,
2974 replicas_desired=role_desired_total,
2975 **status_conditions,
2976 **verification_status_kwargs,
2977 )
2978 return {
2979 "action": "reconcile_mooncake_autoscaler",
2980 "endpoint": name,
2981 "cleanup_complete": False,
2982 }
2984 # Step 7: front-end. Disaggregated and both run behind the proxy; store
2985 # exposes its single Deployment directly.
2986 if mode in ("disaggregated", "both"):
2987 # Per-role Services so the proxy can address prefill and decode by
2988 # stable in-cluster DNS. Routing through a Service means kube-proxy
2989 # load-balances across only the Ready pods of each role, which is
2990 # what gives the proxy ready-only decode routing for free. Each
2991 # Service publishes only its pods' TLS sidecar port.
2992 for role in desired_roles:
2993 self._create_role_service(name, ns, role)
2994 try:
2995 self._create_pd_proxy(name, ns, spec, endpoint)
2996 except AdminApiKeySecretError as e:
2997 logger.error("Proxy for endpoint %s in %s not started: %s", name, ns, e)
2998 self.store.update_region_status(
2999 name,
3000 self.region,
3001 "failed",
3002 error=str(e),
3003 **status_conditions,
3004 )
3005 return {
3006 "action": "reconcile_mooncake",
3007 "endpoint": name,
3008 "failed": "admin_api_key",
3009 }
3010 else:
3011 self._create_service(name, ns, spec)
3013 # Step 8: role-keyed status, then the same deploying -> running
3014 # promotion the classic path performs. Without it a Mooncake endpoint
3015 # would sit in ``deploying`` forever and the proxy would refuse every
3016 # request with "not ready in this region" despite Ready role pods.
3017 state = self._report_role_status(
3018 name,
3019 ns,
3020 mooncake,
3021 region_services,
3022 endpoint,
3023 cold_tier_error=services.error if services.cold_tier_unresolved else None,
3024 )
3025 self._promote_deploying_endpoint(name, endpoint, state)
3026 return {"action": "reconcile_mooncake", "endpoint": name, "state": state}
3028 def _reconcile_stopped(
3029 self,
3030 name: str,
3031 namespace: str,
3032 spec: dict[str, Any] | None = None,
3033 endpoint: dict[str, Any] | None = None,
3034 ) -> dict[str, Any] | None:
3035 """Remove every possible autoscaler owner before scaling all roles to zero."""
3036 mooncake_endpoint = isinstance(spec, dict) and isinstance(spec.get("mooncake"), dict)
3037 self._unready_since.pop(name, None)
3038 self._master_deferral_since.pop(name, None)
3039 status_conditions = self._status_write_conditions(endpoint or {})
3040 role_names = (name, f"{name}-prefill", f"{name}-decode")
3041 cleanup = self._delete_autoscalers(
3042 role_names,
3043 (
3044 name,
3045 f"{name}-prefill",
3046 f"{name}-decode",
3047 f"keda-hpa-{name}",
3048 f"keda-hpa-{name}-prefill",
3049 f"keda-hpa-{name}-decode",
3050 ),
3051 namespace,
3052 )
3053 deployment_names = (*role_names, f"{name}-proxy") if mooncake_endpoint else (name,)
3054 deployments = {
3055 deployment_name: self._get_deployment(deployment_name, namespace)
3056 for deployment_name in deployment_names
3057 }
3058 ready_replicas = sum(
3059 int(getattr(getattr(deployment, "status", None), "ready_replicas", 0) or 0)
3060 for deployment in deployments.values()
3061 if deployment is not None
3062 )
3063 desired_replicas = sum(
3064 int(getattr(getattr(deployment, "spec", None), "replicas", 0) or 0)
3065 for deployment in deployments.values()
3066 if deployment is not None
3067 )
3068 if not cleanup.complete:
3069 status_kwargs: dict[str, Any] = {}
3070 if cleanup.error_message:
3071 status_kwargs["error"] = cleanup.error_message
3072 self.store.update_region_status(
3073 name,
3074 self.region,
3075 "stopping",
3076 replicas_ready=ready_replicas,
3077 replicas_desired=desired_replicas,
3078 **status_conditions,
3079 **status_kwargs,
3080 )
3081 return {"action": "stop", "endpoint": name, "cleanup_complete": False}
3083 scaled = False
3084 for deployment_name, deployment in deployments.items():
3085 if deployment is None:
3086 continue
3087 current_replicas = int(getattr(getattr(deployment, "spec", None), "replicas", 0) or 0)
3088 if current_replicas <= 0:
3089 continue
3090 logger.info("Stopping endpoint role %s (scaling to 0)", deployment_name)
3091 self._scale_deployment(deployment_name, namespace, 0)
3092 scaled = True
3094 self.store.update_region_status(
3095 name,
3096 self.region,
3097 "stopped",
3098 replicas_ready=0,
3099 replicas_desired=0,
3100 **status_conditions,
3101 )
3102 if scaled:
3103 return {"action": "stop", "endpoint": name, "cleanup_complete": True}
3104 return None
3106 def _reconcile_deleted(
3107 self,
3108 endpoint: dict[str, Any],
3109 namespace: str,
3110 spec: dict[str, Any] | None = None,
3111 ) -> dict[str, Any]:
3112 """Converge all parents and generated children to stable absence."""
3113 name = endpoint["endpoint_name"]
3114 lifecycle_id = endpoint.get("lifecycle_id")
3115 if not isinstance(lifecycle_id, str) or not lifecycle_id:
3116 raise RuntimeError("Endpoint deletion requires an immutable lifecycle id")
3117 self._unready_since.pop(name, None)
3118 self._master_deferral_since.pop(name, None)
3119 logger.info("Reconciling deletion of endpoint %s from %s", name, self.region)
3120 cleanup = self._delete_resources(
3121 name,
3122 namespace,
3123 spec,
3124 expected_lifecycle_id=lifecycle_id,
3125 )
3126 state, written = self._record_cleanup_observation(endpoint, cleanup)
3127 return {
3128 "action": "delete",
3129 "endpoint": name,
3130 "cleanup_complete": state == "deleted" and written,
3131 }
3133 # ------------------------------------------------------------------
3134 # Kubernetes resource management
3135 # ------------------------------------------------------------------
3137 def _deployment_exists(self, name: str, namespace: str) -> bool:
3138 return self._get_deployment(name, namespace) is not None
3140 def _get_deployment(self, name: str, namespace: str) -> V1Deployment | None:
3141 try:
3142 deployment = self.apps_v1.read_namespaced_deployment(
3143 name, namespace, _request_timeout=self._k8s_timeout
3144 )
3145 except ApiException as error:
3146 if error.status == 404:
3147 return None
3148 raise
3149 return self._authorize_resource(
3150 deployment,
3151 kind="deployment",
3152 resource_name=name,
3153 patch_metadata=partial(
3154 self.apps_v1.patch_namespaced_deployment,
3155 name,
3156 namespace,
3157 _request_timeout=self._k8s_timeout,
3158 ),
3159 read_resource=lambda: self.apps_v1.read_namespaced_deployment(
3160 name, namespace, _request_timeout=self._k8s_timeout
3161 ),
3162 delete_resource=partial(
3163 self.apps_v1.delete_namespaced_deployment,
3164 name,
3165 namespace,
3166 _request_timeout=self._k8s_timeout,
3167 ),
3168 )
3170 def _get_deployment_image(self, deployment: V1Deployment) -> str | None:
3171 """Get the image of the first container in a deployment."""
3172 containers = deployment.spec.template.spec.containers
3173 if containers:
3174 image: str = containers[0].image
3175 return image
3176 return None
3178 def _resolve_image_for_region(self, spec: dict[str, Any]) -> str:
3179 """Pick the image URI this region should pull from.
3181 ``cli.inference.InferenceManager.deploy`` populates
3182 ``spec["region_image_uris"]`` with a per-region map when the
3183 primary image is an ECR URI, so each cluster can pull from its
3184 local replica instead of crossing the WAN. The map is omitted
3185 for non-ECR refs and for deploys with ``rewrite_image=False``,
3186 in which case we fall back to the flat ``spec["image"]`` URI.
3188 When the map is present but lacks an entry for ``self.region``
3189 (a target region was added after the spec was last written),
3190 the flat URI is also used so the deployment doesn't break — the
3191 next reconcile after a fresh deploy picks up the right URI.
3192 """
3193 region_map = spec.get("region_image_uris")
3194 if isinstance(region_map, dict):
3195 uri = region_map.get(self.region)
3196 if isinstance(uri, str) and uri:
3197 return uri
3198 return str(spec["image"])
3200 # ------------------------------------------------------------------
3201 # In-region service resolution (master address + cold-tier bucket)
3202 # ------------------------------------------------------------------
3204 def _resolve_region_services(
3205 self, name: str, mooncake: dict[str, Any]
3206 ) -> RegionServicesResolution:
3207 """Resolve the in-region addresses an endpoint's ``mooncake.json`` needs.
3209 Everything an endpoint wires to is resolved for the monitor's own
3210 region from regional configuration — never from values typed into the
3211 endpoint spec:
3213 - The shared master's RPC address comes from regional configuration. It
3214 is required whenever the store is enabled. When the store is enabled
3215 and no own-region master address is configured, rendering is skipped
3216 and the existing endpoint configuration is left unchanged; the result
3217 records the unresolved-master condition so the caller can report it.
3218 - The metadata server defaults to the master host on the metadata port
3219 unless regional configuration supplies one explicitly.
3220 - When the cold tier is opted in (``store.cold_tier_enabled`` is the
3221 boolean ``True``), the cold-tier object-store URI is resolved to the
3222 own-region general-purpose regional bucket from that region's
3223 ``/name`` discovery value. Any cold-tier bucket URI in the spec is
3224 ignored. Whether the endpoint writes to the cold tier is governed
3225 solely by the per-endpoint flag, independent of the always-on bucket.
3226 When the bucket cannot be resolved (the region's stack is not yet
3227 deployed), the cold tier is dropped and the condition is recorded,
3228 while the hot-path store stays configured.
3230 Args:
3231 name: The endpoint name, used to scope the cold-tier object key.
3232 mooncake: The ``spec["mooncake"]`` block.
3234 Returns:
3235 A :class:`RegionServicesResolution` carrying the resolved
3236 ``region_services`` dict and any skip/unresolved signals.
3237 """
3238 store = mooncake.get("store", {})
3239 store_enabled = bool(store.get("enabled"))
3241 master_address = os.environ.get(MOONCAKE_MASTER_ADDRESS_ENV, "").strip()
3243 # The store needs an own-region master. It is a fixed in-cluster Service
3244 # the monitor itself provisions per region (mooncake-master:50051), so
3245 # when no override is set in the environment, default to that Service
3246 # rather than deferring — the address is known by construction. An
3247 # operator may still override it via MOONCAKE_MASTER_ADDRESS.
3248 if store_enabled and not master_address:
3249 master_address = f"{MOONCAKE_MASTER_SERVICE}:{MOONCAKE_MASTER_RPC_PORT}"
3251 region_services: dict[str, Any] = {
3252 "metadata_server": self._metadata_server_url(master_address),
3253 }
3254 if store_enabled:
3255 region_services["master_server_address"] = master_address
3257 result = RegionServicesResolution(region_services=region_services)
3259 # Cold tier is opt-in per endpoint. The bucket is always resolved for
3260 # the monitor's own region; any URI supplied in the spec is ignored.
3261 if store_enabled and store.get("cold_tier_enabled") is True:
3262 bucket = self._resolve_regional_shared_bucket()
3263 if bucket:
3264 region_services["cold_tier_s3_uri"] = (
3265 f"s3://{bucket}/{MOONCAKE_COLD_TIER_KEY_PREFIX}/{name}/"
3266 )
3267 else:
3268 # The own-region general-purpose bucket is not resolvable yet.
3269 # Drop the cold tier but keep the hot-path store operating.
3270 result.cold_tier_unresolved = True
3271 result.error = (
3272 "general-purpose regional bucket for region "
3273 f"{self.region} could not be resolved; cold tier disabled, "
3274 "hot-path store still active"
3275 )
3277 return result
3279 def _metadata_server_url(self, master_address: str) -> str:
3280 """Return the metadata server URL, deriving it from the master host.
3282 Regional configuration may supply the metadata server URL directly.
3283 When it does not, the URL defaults to the master host on the metadata
3284 port; if no master host is known the conventional in-cluster service
3285 name is used.
3286 """
3287 configured = os.environ.get(MOONCAKE_METADATA_SERVER_ENV, "").strip()
3288 if configured:
3289 return configured
3290 host = master_address.rsplit(":", 1)[0] if master_address else MOONCAKE_MASTER_SERVICE
3291 return f"http://{host}:{MOONCAKE_METADATA_PORT}/metadata"
3293 def _resolve_regional_shared_bucket(self) -> str | None:
3294 """Resolve the own-region general-purpose bucket name, or ``None``.
3296 Reads the monitor's own region's ``/name`` discovery value for the
3297 always-on general-purpose regional bucket. Returns ``None`` when the
3298 value is absent (the region's stack is not yet deployed) or cannot be
3299 read, so the caller can drop the cold tier without disturbing the
3300 hot-path store.
3301 """
3302 from gco.services.aws_ssm import get_ssm_parameter_optional
3304 param_name = f"{_regional_shared_ssm_parameter_prefix()}/name"
3305 try:
3306 bucket = get_ssm_parameter_optional(param_name, region=self.region)
3307 return bucket if isinstance(bucket, str) and bucket else None
3308 except Exception as e: # any read failure means "unresolved"
3309 logger.warning(
3310 "Failed to resolve general-purpose regional bucket for %s: %s",
3311 self.region,
3312 e,
3313 )
3314 return None
3316 # ------------------------------------------------------------------
3317 # Regional scope boundary (intra-region RDMA enforcement)
3318 # ------------------------------------------------------------------
3320 def _region_of_address(self, address: str) -> str:
3321 """Return the address's explicit AWS region or safe local classification.
3323 AWS region tokens embedded in the host are authoritative. Bare Service
3324 names and Kubernetes ``.svc`` names are local by construction. Any
3325 other host without a region token is external and ambiguous, so it is
3326 classified as ``"unknown"`` and rejected by regional-scope checks.
3327 """
3328 candidate = (address or "").strip()
3329 if not candidate:
3330 return "unknown"
3332 try:
3333 parsed = urlsplit(candidate if "://" in candidate else f"//{candidate}")
3334 host = (parsed.hostname or "").rstrip(".").lower()
3335 except ValueError:
3336 return "unknown"
3337 if not host:
3338 return "unknown"
3340 match = _REGION_TOKEN_PATTERN.search(host)
3341 if match:
3342 return match.group(0)
3343 if "." not in host or host.endswith((".svc", ".svc.cluster.local")):
3344 return self.region
3345 return "unknown"
3347 def _resolve_regional_scope(
3348 self,
3349 name: str,
3350 ns: str,
3351 spec: dict[str, Any],
3352 region_services: dict[str, Any] | None,
3353 ) -> RegionalScopeResolution:
3354 """Confirm a disaggregated topology wires only to its own region.
3356 Gathers every address the own-region topology connects to — the
3357 ``MooncakeConnector`` peers (the sibling role Services for prefill and
3358 decode), the shared master's RPC address, and the metadata server — and
3359 confirms each resolves to the monitor's own region. Any explicitly
3360 supplied peer addresses in the spec are checked too, so a misconfigured
3361 endpoint that points a peer or master at another region is caught
3362 before any role pod is materialized.
3364 An endpoint that enumerates two or more target regions runs one
3365 independent topology per region: each region's monitor reconciles only
3366 its own topology (it reconciles only while ``self.region`` is one of the
3367 target regions), and this resolution confirms that topology's addresses
3368 never cross into another region.
3370 Args:
3371 name: The endpoint name; used to derive the in-cluster peer Service
3372 names for the disaggregated roles.
3373 ns: The namespace the topology materializes into.
3374 spec: The endpoint spec being reconciled.
3375 region_services: The resolved in-region addresses (from
3376 :meth:`_resolve_region_services`), or ``None`` when none were
3377 resolved. Supplies the master and metadata addresses to check.
3379 Returns:
3380 A :class:`RegionalScopeResolution`. When every resolved address is
3381 own-region, ``in_region`` is ``True`` and the caller may materialize
3382 the role Deployments. When any address resolves to another region,
3383 ``in_region`` is ``False``, ``state`` is ``"failed"``, and ``error``
3384 names the offending addresses; the caller then materializes no role
3385 Deployments and leaves any prior resources unchanged.
3386 """
3387 mooncake = spec.get("mooncake") or {}
3388 mode = mooncake.get("mode")
3389 roles = _WORKER_ROLES_BY_MODE.get(mode, set()) if isinstance(mode, str) else set()
3391 # MooncakeConnector peers are the sibling role Services within this
3392 # namespace; collect them in a stable order for deterministic reporting.
3393 addresses: list[str] = []
3394 if "prefill" in roles or "decode" in roles:
3395 for role in ("prefill", "decode"):
3396 addresses.append(f"{name}-{role}.{ns}.svc.cluster.local")
3398 # The shared master and metadata server the pods reach. These come from
3399 # the own-region resolution, but are checked here so a foreign address
3400 # supplied through regional configuration is still caught.
3401 if region_services:
3402 master = region_services.get("master_server_address")
3403 metadata = region_services.get("metadata_server")
3404 if master:
3405 addresses.append(str(master))
3406 if metadata:
3407 addresses.append(str(metadata))
3409 # Defensive: honor any explicit peer/master addresses authored on the
3410 # spec so a hand-edited endpoint cannot smuggle in an out-of-region peer.
3411 store = mooncake.get("store") or {}
3412 transfer = mooncake.get("transfer") or {}
3413 for candidate in (
3414 store.get("master_server_address"),
3415 store.get("metadata_server"),
3416 ):
3417 if candidate:
3418 addresses.append(str(candidate))
3419 explicit_peers = transfer.get("peer_addresses")
3420 if isinstance(explicit_peers, list):
3421 addresses.extend(str(peer) for peer in explicit_peers if peer)
3423 # De-duplicate while preserving first-seen order.
3424 seen: set[str] = set()
3425 ordered: list[str] = []
3426 for address in addresses:
3427 if address not in seen:
3428 seen.add(address)
3429 ordered.append(address)
3431 resolved_regions = [(address, self._region_of_address(address)) for address in ordered]
3432 out_of_region = [
3433 (address, region) for address, region in resolved_regions if region != self.region
3434 ]
3436 if out_of_region:
3437 detail = ", ".join(
3438 f"{address!r} resolves to region {region}" for address, region in out_of_region
3439 )
3440 logger.error(
3441 "Cross-region boundary violation for endpoint %s in %s: %s",
3442 name,
3443 self.region,
3444 detail,
3445 )
3446 return RegionalScopeResolution(
3447 in_region=False,
3448 peer_addresses=ordered,
3449 state="failed",
3450 error=(f"cross-region boundary violation: {detail}; expected region {self.region}"),
3451 )
3453 return RegionalScopeResolution(in_region=True, peer_addresses=ordered)
3455 def _ensure_mooncake_store(self, ns: str, spec: dict[str, Any]) -> None:
3456 """Maintain the single shared per-region Mooncake master, idempotently.
3458 The master is region-shared, not per-endpoint: every endpoint that
3459 needs the store reaches the same ``mooncake-master`` StatefulSet and the
3460 headless Service that fronts its RPC and metadata ports. This method
3461 uses create-if-absent semantics, so any number of calls within a region
3462 converge on exactly one StatefulSet with a single replica and one
3463 Service. An already-existing master is left untouched — a conflicting
3464 create is treated as success and never overwrites the running master.
3466 The StatefulSet runs the master daemon with its built-in HTTP metadata
3467 server, exposing RPC on :data:`MOONCAKE_MASTER_RPC_PORT` and the
3468 metadata endpoint on :data:`MOONCAKE_METADATA_PORT`. Both ports are
3469 published on the headless Service so in-namespace pods can resolve them.
3471 Args:
3472 ns: The namespace the master shares with the inference workloads.
3473 spec: The endpoint spec being reconciled. Its ``mooncake`` block may
3474 carry a master image override; otherwise the image is taken from
3475 the in-region deployment's environment.
3476 """
3477 mooncake = spec.get("mooncake", {}) or {}
3478 store = mooncake.get("store", {}) or {}
3479 image = store.get("master_image") or os.environ.get(MOONCAKE_MASTER_IMAGE_ENV, "").strip()
3481 labels = {"app": MOONCAKE_MASTER_SERVICE, "project": "gco"}
3483 # Headless Service exposing both the RPC and metadata ports. A None
3484 # cluster IP keeps it headless so the StatefulSet's stable network
3485 # identity resolves directly.
3486 service = client.V1Service(
3487 metadata=client.V1ObjectMeta(
3488 name=MOONCAKE_MASTER_SERVICE,
3489 namespace=ns,
3490 labels=labels,
3491 ),
3492 spec=client.V1ServiceSpec(
3493 cluster_ip="None",
3494 selector={"app": MOONCAKE_MASTER_SERVICE},
3495 ports=[
3496 client.V1ServicePort(
3497 name="rpc",
3498 port=MOONCAKE_MASTER_RPC_PORT,
3499 target_port="rpc",
3500 protocol="TCP",
3501 ),
3502 client.V1ServicePort(
3503 name="metadata",
3504 port=MOONCAKE_METADATA_PORT,
3505 target_port="metadata",
3506 protocol="TCP",
3507 ),
3508 ],
3509 ),
3510 )
3512 container = client.V1Container(
3513 name=MOONCAKE_MASTER_SERVICE,
3514 image=image,
3515 command=["mooncake_master"],
3516 args=[
3517 f"--port={MOONCAKE_MASTER_RPC_PORT}",
3518 "--enable_http_metadata_server=true",
3519 f"--http_metadata_server_port={MOONCAKE_METADATA_PORT}",
3520 ],
3521 ports=[
3522 client.V1ContainerPort(
3523 name="rpc",
3524 container_port=MOONCAKE_MASTER_RPC_PORT,
3525 protocol="TCP",
3526 ),
3527 client.V1ContainerPort(
3528 name="metadata",
3529 container_port=MOONCAKE_METADATA_PORT,
3530 protocol="TCP",
3531 ),
3532 ],
3533 security_context=client.V1SecurityContext(
3534 allow_privilege_escalation=False,
3535 # The upstream mooncake_master launcher chmods its bundled
3536 # binary on startup, which needs a writable root filesystem (a
3537 # read-only root raised OSError: Read-only file system). The pod
3538 # also runs as root so the chmod of the root-owned binary is
3539 # permitted. Privilege escalation stays disabled and all
3540 # capabilities are dropped, so this is constrained root.
3541 read_only_root_filesystem=False,
3542 capabilities=client.V1Capabilities(drop=["ALL"]),
3543 ),
3544 resources=client.V1ResourceRequirements(
3545 requests={"cpu": "250m", "memory": "512Mi"},
3546 limits={"cpu": "1", "memory": "2Gi"},
3547 ),
3548 startup_probe=client.V1Probe(
3549 tcp_socket=client.V1TCPSocketAction(port="rpc"),
3550 initial_delay_seconds=5,
3551 period_seconds=5,
3552 failure_threshold=30,
3553 ),
3554 liveness_probe=client.V1Probe(
3555 tcp_socket=client.V1TCPSocketAction(port="rpc"),
3556 initial_delay_seconds=15,
3557 period_seconds=30,
3558 ),
3559 readiness_probe=client.V1Probe(
3560 # The HTTP metadata server returns 400 for a bare GET /metadata
3561 # (it expects a ?key=), so an HTTP GET readiness probe never
3562 # passes. Confirm readiness by checking the metadata port is
3563 # accepting connections instead.
3564 tcp_socket=client.V1TCPSocketAction(port="metadata"),
3565 initial_delay_seconds=10,
3566 period_seconds=15,
3567 ),
3568 )
3570 stateful_set = client.V1StatefulSet(
3571 metadata=client.V1ObjectMeta(
3572 name=MOONCAKE_MASTER_SERVICE,
3573 namespace=ns,
3574 labels=labels,
3575 ),
3576 spec=client.V1StatefulSetSpec(
3577 service_name=MOONCAKE_MASTER_SERVICE,
3578 replicas=1,
3579 selector=client.V1LabelSelector(match_labels={"app": MOONCAKE_MASTER_SERVICE}),
3580 template=client.V1PodTemplateSpec(
3581 metadata=client.V1ObjectMeta(
3582 labels=labels,
3583 # Keep Karpenter from consolidating the node out from
3584 # under the master: it is a single-replica, stateful
3585 # control-plane daemon holding KV metadata for every
3586 # in-region endpoint, so an eviction drops that state and
3587 # disrupts inference. On lightly-loaded clusters
3588 # consolidation otherwise evicts it mid-image-pull before
3589 # it can even start.
3590 annotations={"karpenter.sh/do-not-disrupt": "true"},
3591 ),
3592 spec=client.V1PodSpec(
3593 service_account_name="gco-service-account",
3594 automount_service_account_token=False,
3595 # The upstream mooncake_master launcher chmods its
3596 # bundled binary (root-owned in the image) on startup, so
3597 # the master must run as root: a non-root uid cannot
3598 # chmod a root-owned file ("Operation not permitted").
3599 # gco-inference does not enforce restricted Pod Security
3600 # and the no-root rule applies only to user-submitted
3601 # jobs, so a root platform daemon is consistent here. The
3602 # container still drops all Linux capabilities and
3603 # disallows privilege escalation (see its securityContext).
3604 security_context=client.V1PodSecurityContext(
3605 run_as_user=0,
3606 run_as_group=0,
3607 ),
3608 containers=[container],
3609 restart_policy="Always",
3610 ),
3611 ),
3612 ),
3613 )
3615 # Create-if-absent: a 409 means the shared master already exists, which
3616 # is the steady state. Leave it untouched and treat it as success.
3617 self._assert_mutation_authority()
3618 try:
3619 self.core_v1.create_namespaced_service(ns, service, _request_timeout=self._k8s_timeout)
3620 logger.info("Created shared mooncake master service in %s", ns)
3621 except ApiException as e:
3622 if e.status == 409:
3623 logger.info("Shared mooncake master service already exists in %s", ns)
3624 else:
3625 raise
3627 self._assert_mutation_authority()
3628 try:
3629 self.apps_v1.create_namespaced_stateful_set(
3630 ns, stateful_set, _request_timeout=self._k8s_timeout
3631 )
3632 logger.info("Created shared mooncake master statefulset in %s", ns)
3633 except ApiException as e:
3634 if e.status == 409:
3635 logger.info("Shared mooncake master statefulset already exists in %s", ns)
3636 else:
3637 raise
3639 def _mooncake_master_ready_replicas(self, ns: str) -> int:
3640 """Return the shared master's Ready replica count, 0 when absent.
3642 Reads the ``mooncake-master`` StatefulSet status in ``ns``. A missing
3643 StatefulSet (404) reports zero Ready replicas rather than raising, so a
3644 caller gating on readiness simply keeps deferring until it appears.
3646 Args:
3647 ns: The namespace the shared master lives in.
3649 Returns:
3650 The number of Ready replicas the StatefulSet reports, or 0 when it
3651 does not yet exist or reports no Ready replicas.
3652 """
3653 try:
3654 status = self.apps_v1.read_namespaced_stateful_set_status(
3655 MOONCAKE_MASTER_SERVICE, ns, _request_timeout=self._k8s_timeout
3656 )
3657 except ApiException as e:
3658 if e.status == 404:
3659 return 0
3660 raise
3662 ready = getattr(getattr(status, "status", None), "ready_replicas", 0)
3663 return int(ready or 0)
3665 def _gate_on_mooncake_master(
3666 self, name: str, ns: str, spec: dict[str, Any]
3667 ) -> MasterReadinessGate:
3668 """Gate dependent role-pod creation on the shared master's readiness.
3670 Maintains the single shared master (create-if-absent) and then decides
3671 whether the endpoint's dependent role pods may be materialized:
3673 - If maintaining the master fails, no dependent pods are materialized,
3674 any existing master is left unmodified, and the endpoint stays in
3675 ``creating`` carrying a create-failure error.
3676 - While the master reports fewer than 1 Ready replica, creation is
3677 deferred and the endpoint stays in ``creating``. The first deferral
3678 starts a clock; once it exceeds the wait window the endpoint keeps
3679 deferring and stays in ``creating`` but also surfaces a not-ready
3680 error. The master is never deleted or modified on account of the
3681 timeout.
3682 - Once the master reports at least 1 Ready replica, the gate opens: the
3683 clock is cleared and the caller may create the role pods and advance
3684 out of ``creating``.
3686 Args:
3687 name: The endpoint name, used to track its first deferral.
3688 ns: The namespace the master shares with the workloads.
3689 spec: The endpoint spec being reconciled.
3691 Returns:
3692 A :class:`MasterReadinessGate` describing whether to proceed, the
3693 endpoint state to report, and any error to surface.
3694 """
3695 # Maintain the shared master first. A create failure must not produce
3696 # any dependent pods and must leave an existing master untouched.
3697 try:
3698 self._ensure_mooncake_store(ns, spec)
3699 except ApiException as e:
3700 logger.error(
3701 "Could not create shared mooncake master in %s for endpoint %s: %s",
3702 ns,
3703 name,
3704 e,
3705 )
3706 return MasterReadinessGate(
3707 proceed=False,
3708 state="creating",
3709 error="shared master could not be created",
3710 )
3712 # Apply the intra-namespace allow rules before any role pod is created.
3713 # A failure here must fail pod materialization while leaving the
3714 # default-deny posture intact; surface which rule could not be applied.
3715 try:
3716 self._ensure_intra_namespace_network_policies(ns, spec)
3717 except NetworkPolicyApplyError as e:
3718 logger.error(
3719 "Could not apply network policy %s in %s for endpoint %s: %s",
3720 e.rule,
3721 ns,
3722 name,
3723 e.reason,
3724 )
3725 return MasterReadinessGate(
3726 proceed=False,
3727 state="creating",
3728 error=f"network policy {e.rule} could not be applied",
3729 )
3731 ready_replicas = self._mooncake_master_ready_replicas(ns)
3732 if ready_replicas >= 1:
3733 # Master is Ready: open the gate and reset the deferral clock so a
3734 # later master restart restarts the wait window cleanly.
3735 if name in self._master_deferral_since:
3736 logger.info(
3737 "Shared master Ready in %s, resuming creation for endpoint %s",
3738 ns,
3739 name,
3740 )
3741 del self._master_deferral_since[name]
3742 return MasterReadinessGate(proceed=True, state=None, error=None)
3744 # Master not Ready: defer creation and report creating. Start the clock
3745 # on the first deferral.
3746 now = datetime.now(UTC)
3747 first_deferral = self._master_deferral_since.setdefault(name, now)
3748 deferred_for = (now - first_deferral).total_seconds()
3750 if deferred_for >= MOONCAKE_MASTER_READY_TIMEOUT_SECONDS:
3751 logger.error(
3752 "Shared master not Ready in %s after %.0fs, still deferring endpoint %s",
3753 ns,
3754 deferred_for,
3755 name,
3756 )
3757 return MasterReadinessGate(
3758 proceed=False,
3759 state="creating",
3760 error="shared master did not become Ready",
3761 )
3763 logger.info(
3764 "Deferring creation for endpoint %s in %s until shared master is Ready",
3765 name,
3766 ns,
3767 )
3768 return MasterReadinessGate(proceed=False, state="creating", error=None)
3770 def _ensure_intra_namespace_network_policies(self, ns: str, spec: dict[str, Any]) -> None:
3771 """Apply the intra-namespace allow rules disaggregated inference needs.
3773 ``03-network-policies.yaml`` isolates ``gco-inference``: every pod's
3774 egress is limited to DNS, HTTPS and the node's Pod Identity Agent, and
3775 model pods accept ingress only from the inference proxy and, through
3776 the rules below, from each other. The manifest ships these four
3777 widening allow rules too; this re-creates any that are missing
3778 (create-if-absent) and never touches the namespace's other policies:
3780 - ``allow-inference-internal`` — managed inference pods exchange TCP
3781 traffic and may reach the shared master's two fixed ports. This
3782 permits proxy-to-role serving and bootstrap traffic while excluding
3783 unselected sources such as the ALB.
3784 - ``allow-pod-to-master`` — inference pods reach the shared master RPC
3785 port (:data:`MOONCAKE_MASTER_RPC_PORT`).
3786 - ``allow-pod-to-metadata`` — inference pods reach the shared metadata
3787 server (:data:`MOONCAKE_METADATA_PORT`).
3788 - ``allow-rdma-bootstrap`` — inference pods reach each other on the
3789 contiguous KV-transfer bootstrap port window starting at the spec's
3790 ``mooncake.transfer.bootstrap_base_port`` (default
3791 :data:`MOONCAKE_BOOTSTRAP_BASE_PORT`).
3793 Each rule is created independently; an already-present rule (409) is the
3794 steady state and counts as success. No other policy is ever read,
3795 modified, or deleted, so the namespace's isolation holds regardless of
3796 outcome.
3798 Args:
3799 ns: The inference namespace the rules apply to.
3800 spec: The endpoint spec being reconciled; its
3801 ``mooncake.transfer.bootstrap_base_port`` sizes the bootstrap
3802 port window.
3804 Raises:
3805 NetworkPolicyApplyError: If any single rule cannot be created. The
3806 error names the failing rule; rules created before the failure
3807 remain in place and no other policy is touched.
3808 """
3809 transfer = (spec.get("mooncake", {}) or {}).get("transfer", {}) or {}
3810 base_port = transfer.get("bootstrap_base_port", MOONCAKE_BOOTSTRAP_BASE_PORT)
3811 try:
3812 base_port = int(base_port)
3813 except TypeError, ValueError:
3814 base_port = MOONCAKE_BOOTSTRAP_BASE_PORT
3815 end_port = min(base_port + MOONCAKE_BOOTSTRAP_PORT_SPAN, MAX_BOOTSTRAP_PORT)
3817 labels = {"project": "gco"}
3818 master_selector = client.V1LabelSelector(match_labels={"app": MOONCAKE_MASTER_SERVICE})
3819 inference_selector = client.V1LabelSelector(match_labels=INFERENCE_POD_SELECTOR)
3820 inference_peer = [client.V1NetworkPolicyPeer(pod_selector=inference_selector)]
3821 master_peer = [client.V1NetworkPolicyPeer(pod_selector=master_selector)]
3822 all_tcp = [client.V1NetworkPolicyPort(protocol="TCP")]
3824 policies = [
3825 (
3826 NETWORK_POLICY_INFERENCE_INTERNAL,
3827 client.V1NetworkPolicy(
3828 metadata=client.V1ObjectMeta(
3829 name=NETWORK_POLICY_INFERENCE_INTERNAL, namespace=ns, labels=labels
3830 ),
3831 spec=client.V1NetworkPolicySpec(
3832 pod_selector=inference_selector,
3833 policy_types=["Ingress", "Egress"],
3834 ingress=[
3835 client.V1NetworkPolicyIngressRule(
3836 _from=inference_peer,
3837 ports=all_tcp,
3838 )
3839 ],
3840 egress=[
3841 client.V1NetworkPolicyEgressRule(
3842 to=inference_peer,
3843 ports=all_tcp,
3844 ),
3845 client.V1NetworkPolicyEgressRule(
3846 to=master_peer,
3847 ports=[
3848 client.V1NetworkPolicyPort(
3849 protocol="TCP", port=MOONCAKE_MASTER_RPC_PORT
3850 ),
3851 client.V1NetworkPolicyPort(
3852 protocol="TCP", port=MOONCAKE_METADATA_PORT
3853 ),
3854 ],
3855 ),
3856 ],
3857 ),
3858 ),
3859 ),
3860 (
3861 NETWORK_POLICY_POD_TO_MASTER,
3862 client.V1NetworkPolicy(
3863 metadata=client.V1ObjectMeta(
3864 name=NETWORK_POLICY_POD_TO_MASTER, namespace=ns, labels=labels
3865 ),
3866 spec=client.V1NetworkPolicySpec(
3867 pod_selector=master_selector,
3868 policy_types=["Ingress"],
3869 ingress=[
3870 client.V1NetworkPolicyIngressRule(
3871 _from=inference_peer,
3872 ports=[
3873 client.V1NetworkPolicyPort(
3874 protocol="TCP", port=MOONCAKE_MASTER_RPC_PORT
3875 )
3876 ],
3877 )
3878 ],
3879 ),
3880 ),
3881 ),
3882 (
3883 NETWORK_POLICY_POD_TO_METADATA,
3884 client.V1NetworkPolicy(
3885 metadata=client.V1ObjectMeta(
3886 name=NETWORK_POLICY_POD_TO_METADATA, namespace=ns, labels=labels
3887 ),
3888 spec=client.V1NetworkPolicySpec(
3889 pod_selector=master_selector,
3890 policy_types=["Ingress"],
3891 ingress=[
3892 client.V1NetworkPolicyIngressRule(
3893 _from=inference_peer,
3894 ports=[
3895 client.V1NetworkPolicyPort(
3896 protocol="TCP", port=MOONCAKE_METADATA_PORT
3897 )
3898 ],
3899 )
3900 ],
3901 ),
3902 ),
3903 ),
3904 (
3905 NETWORK_POLICY_RDMA_BOOTSTRAP,
3906 client.V1NetworkPolicy(
3907 metadata=client.V1ObjectMeta(
3908 name=NETWORK_POLICY_RDMA_BOOTSTRAP, namespace=ns, labels=labels
3909 ),
3910 spec=client.V1NetworkPolicySpec(
3911 pod_selector=inference_selector,
3912 policy_types=["Ingress"],
3913 ingress=[
3914 client.V1NetworkPolicyIngressRule(
3915 _from=inference_peer,
3916 ports=[
3917 client.V1NetworkPolicyPort(
3918 protocol="TCP",
3919 port=base_port,
3920 end_port=end_port,
3921 )
3922 ],
3923 )
3924 ],
3925 ),
3926 ),
3927 ),
3928 ]
3930 for rule_name, policy in policies:
3931 self._assert_mutation_authority()
3932 try:
3933 self.networking_v1.create_namespaced_network_policy(
3934 ns, policy, _request_timeout=self._k8s_timeout
3935 )
3936 logger.info("Applied network policy %s in %s", rule_name, ns)
3937 except ApiException as e:
3938 if e.status == 409:
3939 # Already present — the steady state. Leave it untouched.
3940 logger.info("Network policy %s already present in %s", rule_name, ns)
3941 continue
3942 logger.error(
3943 "Could not apply network policy %s in %s: %s",
3944 rule_name,
3945 ns,
3946 e,
3947 )
3948 raise NetworkPolicyApplyError(rule_name, e.reason or str(e)) from e
3950 def _create_deployment(
3951 self,
3952 name: str,
3953 namespace: str,
3954 spec: dict[str, Any],
3955 *,
3956 tls_proxy_config_map: str | None = None,
3957 ) -> None:
3958 """Create a Kubernetes Deployment for an inference endpoint.
3960 ``tls_proxy_config_map`` names the ConfigMap carrying the TLS sidecar
3961 program when it is not ``name``'s own: a canary Deployment shares its
3962 endpoint's ConfigMap.
3963 """
3964 replicas = spec.get("replicas", 1)
3965 deployment = self._build_inference_deployment_object(
3966 name=name,
3967 deploy_name=name,
3968 app_label=name,
3969 namespace=namespace,
3970 spec=spec,
3971 replicas=replicas,
3972 tls_proxy_config_map=tls_proxy_config_map,
3973 )
3974 self._assert_mutation_authority()
3975 self.apps_v1.create_namespaced_deployment(
3976 namespace, deployment, _request_timeout=self._k8s_timeout
3977 )
3978 self._confirm_created_resource(
3979 kind="deployment",
3980 resource_name=name,
3981 read_resource=partial(
3982 self.apps_v1.read_namespaced_deployment,
3983 name,
3984 namespace,
3985 _request_timeout=self._k8s_timeout,
3986 ),
3987 delete_resource=partial(
3988 self.apps_v1.delete_namespaced_deployment,
3989 name,
3990 namespace,
3991 _request_timeout=self._k8s_timeout,
3992 ),
3993 )
3994 logger.info("Created deployment %s/%s", namespace, name)
3996 def _build_inference_deployment_object(
3997 self,
3998 name: str,
3999 deploy_name: str,
4000 app_label: str,
4001 namespace: str,
4002 spec: dict[str, Any],
4003 replicas: int,
4004 extra_args: list[str] | None = None,
4005 extra_labels: dict[str, str] | None = None,
4006 *,
4007 tls_proxy_config_map: str | None = None,
4008 ) -> client.V1Deployment:
4009 """Build the ``V1Deployment`` object for an inference workload.
4011 Shared by the single-Deployment path and the role-split prefill/decode/
4012 store paths. ``name`` is the endpoint name (used for the in-cluster
4013 serving prefix and model cache directory), ``deploy_name`` is the
4014 Kubernetes object name, and ``app_label`` is the selector label that
4015 Services and autoscalers target. ``extra_args`` are appended to the
4016 container args (for example the rendered ``--kv-transfer-config``), and
4017 ``extra_labels`` are merged into both the Deployment and pod-template
4018 labels so role pods carry a stable role marker.
4020 The pod always carries the ``endpoint-tls-proxy`` sidecar after the
4021 model container, fronting the model's own port; its program comes from
4022 ``tls_proxy_config_map`` (default: ``name``'s ``{name}-tls-proxy``).
4023 """
4024 image = self._resolve_image_for_region(spec)
4025 port = spec.get("port", 8000)
4026 _reject_tls_port_collision(port)
4027 gpu_count = spec.get("gpu_count", 1)
4028 health_path = spec.get("health_check_path", "/health")
4029 env_vars = spec.get("env", {})
4030 # Stable block hashing across data-parallel ranks: identical prompts
4031 # must hash identically so shared prefix-cache hits are not lost
4032 # between pods. The disaggregated serving image (upstream vLLM) no
4033 # longer bakes this in, so default it here; an explicit spec env wins.
4034 env_vars = {"PYTHONHASHSEED": "0", **env_vars}
4035 resources = spec.get("resources", {})
4036 model_path = spec.get("model_path")
4037 command = spec.get("command")
4038 args = spec.get("args")
4040 # Build container
4041 container_env = [client.V1EnvVar(name=k, value=str(v)) for k, v in env_vars.items()]
4043 # Runtime behavior is persisted explicitly by callers that need a
4044 # strict adapter. Legacy endpoints without the field retain image
4045 # detection. Only vLLM is given --root-path: the authenticated platform
4046 # proxy strips /inference/{name} before forwarding /health, /generate
4047 # or /server_info to the model Service, so SGLang serves its documented
4048 # unprefixed paths.
4049 serving_prefix = f"/inference/{name}"
4050 runtime_framework = spec.get("framework")
4051 if runtime_framework not in ("vllm", "sglang"):
4052 image_lower = image.lower()
4053 if "vllm" in image_lower:
4054 runtime_framework = "vllm"
4055 elif "sglang" in image_lower:
4056 runtime_framework = "sglang"
4057 else:
4058 runtime_framework = None
4059 if not command and runtime_framework == "vllm":
4060 # vLLM never reads MODEL itself, so without this a deploy with
4061 # -e MODEL=... served vLLM's built-in default model while
4062 # ``gco inference invoke`` asked for MODEL by name. The model goes
4063 # on argv as ``--model $(MODEL)`` (the kubelet expands it at
4064 # container start); ``--model`` rather than the positional tag
4065 # works for both the ``vllm serve`` and the older api_server
4066 # entrypoints. Operator arguments that already name a model win.
4067 args = list(args) if args else []
4068 if "MODEL" in env_vars and not _vllm_args_name_model(args):
4069 args = ["--model", "$(MODEL)", *args]
4070 if "--root-path" not in args:
4071 args = [*args, "--root-path", serving_prefix]
4072 if not command and runtime_framework == "sglang":
4073 # The lmsysorg/sglang image has no entrypoint (its CMD is a shell),
4074 # so the renderer supplies the official launcher. The launcher
4075 # binds 127.0.0.1:30000 by default, which the pod's Service could
4076 # never reach, so the host and the spec's port are pinned unless
4077 # the operator passed them explicitly. The model comes from
4078 # ``--model-path`` in the args when given; otherwise from the
4079 # ``MODEL`` environment convention every GCO renderer shares,
4080 # expanded by the kubelet at container start.
4081 command = ["python3", "-m", "sglang.launch_server"]
4082 args = list(args) if args else []
4083 if "--host" not in args:
4084 args.extend(["--host", "0.0.0.0"]) # container listener
4085 if "--port" not in args:
4086 args.extend(["--port", str(port)])
4087 if "--model-path" not in args and "--model" not in args and "MODEL" in env_vars:
4088 args.extend(["--model-path", "$(MODEL)"])
4090 # Append caller-supplied arguments (for example the rendered
4091 # --kv-transfer-config) after any root-path injection so they survive
4092 # alongside user --extra-args.
4093 if extra_args:
4094 args = (list(args) if args else []) + list(extra_args)
4096 resource_reqs = client.V1ResourceRequirements(
4097 requests=resources.get("requests", {"cpu": "1", "memory": "4Gi"}),
4098 limits=resources.get("limits", {"cpu": "4", "memory": "16Gi"}),
4099 )
4100 # Add accelerator resources (GPU or Neuron)
4101 accelerator = spec.get("accelerator", "nvidia")
4102 if gpu_count > 0:
4103 if accelerator == "neuron":
4104 # AWS Trainium/Inferentia — request Neuron devices
4105 if resource_reqs.limits is None:
4106 resource_reqs.limits = {}
4107 resource_reqs.limits["aws.amazon.com/neuron"] = str(gpu_count)
4108 if resource_reqs.requests is None:
4109 resource_reqs.requests = {}
4110 resource_reqs.requests["aws.amazon.com/neuron"] = str(gpu_count)
4111 else:
4112 # NVIDIA GPU (default)
4113 if resource_reqs.limits is None:
4114 resource_reqs.limits = {}
4115 resource_reqs.limits["nvidia.com/gpu"] = str(gpu_count)
4116 if resource_reqs.requests is None:
4117 resource_reqs.requests = {}
4118 resource_reqs.requests["nvidia.com/gpu"] = str(gpu_count)
4120 volume_mounts = []
4121 volumes = []
4122 init_containers = []
4123 model_source = spec.get("model_source")
4125 if model_path or model_source:
4126 volume_mounts.append(
4127 client.V1VolumeMount(
4128 name="model-storage",
4129 mount_path="/models",
4130 )
4131 )
4132 volumes.append(
4133 client.V1Volume(
4134 name="model-storage",
4135 persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
4136 claim_name="efs-claim",
4137 ),
4138 )
4139 )
4141 # Mooncake role pods read the shared transport config (metadata-server
4142 # address, protocol, device) from the per-endpoint ``{name}-mooncake``
4143 # ConfigMap mounted read-only at MOONCAKE_CONFIG_MOUNT_DIR, pointed at by
4144 # MOONCAKE_CONFIG_PATH, and learn the KV-transfer bootstrap base port via
4145 # VLLM_MOONCAKE_BOOTSTRAP_PORT. Plain (non-mooncake) endpoints are
4146 # untouched: no volume, mount, or env is added.
4147 mooncake_block = spec.get("mooncake")
4148 if mooncake_block:
4149 volume_mounts.append(
4150 client.V1VolumeMount(
4151 name="mooncake-config",
4152 mount_path=MOONCAKE_CONFIG_MOUNT_DIR,
4153 read_only=True,
4154 )
4155 )
4156 volumes.append(
4157 client.V1Volume(
4158 name="mooncake-config",
4159 config_map=client.V1ConfigMapVolumeSource(name=f"{name}-mooncake"),
4160 )
4161 )
4162 transfer_block = mooncake_block.get("transfer") or {}
4163 base_port = transfer_block.get("bootstrap_base_port", MOONCAKE_BOOTSTRAP_BASE_PORT)
4164 try:
4165 base_port = int(base_port)
4166 except TypeError, ValueError:
4167 base_port = MOONCAKE_BOOTSTRAP_BASE_PORT
4168 _reject_bootstrap_window_collision(base_port)
4169 container_env.append(
4170 client.V1EnvVar(name=MOONCAKE_CONFIG_PATH_ENV, value=MOONCAKE_CONFIG_FILE_PATH)
4171 )
4172 container_env.append(
4173 client.V1EnvVar(name=VLLM_MOONCAKE_BOOTSTRAP_PORT_ENV, value=str(base_port))
4174 )
4176 # Sync directly with literal argv so a model URI can never become
4177 # shell syntax. ``aws s3 sync`` avoids retransferring unchanged
4178 # objects on reruns while also repairing partial downloads.
4179 if model_source and model_source.startswith("s3://"):
4180 model_dest = f"/models/{name}"
4181 init_containers.append(
4182 client.V1Container(
4183 name="model-sync",
4184 image=AWS_CLI_IMAGE,
4185 command=["aws"],
4186 args=["s3", "sync", model_source, model_dest, "--quiet"],
4187 volume_mounts=[
4188 client.V1VolumeMount(
4189 name="model-storage",
4190 mount_path="/models",
4191 )
4192 ],
4193 resources=client.V1ResourceRequirements(
4194 requests={"cpu": "1", "memory": "2Gi"},
4195 limits={"cpu": "4", "memory": "8Gi"},
4196 ),
4197 )
4198 )
4200 # Probe path depends on whether the server handles the prefix
4201 uses_root_path = args is not None and "--root-path" in args
4202 probe_health = f"{serving_prefix}{health_path}" if uses_root_path else health_path
4204 # Every probe states its timeout; the kubelet's silent 1 s default is
4205 # what killed a healthy SGLang container every ~3 minutes (its /health
4206 # generates a token and polls at one-second steps, so it answers in
4207 # just over a second and never faster). See INFERENCE_PROBE_TIMEOUT_SECONDS.
4208 probe_timeout = INFERENCE_PROBE_TIMEOUT_SECONDS
4210 container = client.V1Container(
4211 name="inference",
4212 image=image,
4213 ports=[client.V1ContainerPort(container_port=port)],
4214 env=container_env if container_env else None,
4215 resources=resource_reqs,
4216 volume_mounts=volume_mounts if volume_mounts else None,
4217 command=command,
4218 args=args,
4219 # A model server answers its health path only once the weights are
4220 # downloaded and loaded: SGLang returns 503 until then, and vLLM
4221 # and Triton don't listen at all. For a multi-gigabyte checkpoint
4222 # that outlasts the liveness budget below (about three minutes),
4223 # and every restart downloads the weights again, so the pod never
4224 # comes up. The startup probe holds liveness off for up to 20
4225 # minutes while that happens, whatever the runtime.
4226 startup_probe=client.V1Probe(
4227 http_get=client.V1HTTPGetAction(path=probe_health, port=port),
4228 period_seconds=15,
4229 failure_threshold=80,
4230 timeout_seconds=probe_timeout,
4231 ),
4232 liveness_probe=client.V1Probe(
4233 http_get=client.V1HTTPGetAction(path=probe_health, port=port),
4234 initial_delay_seconds=120,
4235 period_seconds=15,
4236 failure_threshold=5,
4237 timeout_seconds=probe_timeout,
4238 ),
4239 readiness_probe=client.V1Probe(
4240 http_get=client.V1HTTPGetAction(path=probe_health, port=port),
4241 initial_delay_seconds=30,
4242 period_seconds=10,
4243 timeout_seconds=probe_timeout,
4244 ),
4245 )
4247 # Build tolerations based on accelerator type
4248 if accelerator == "neuron":
4249 tolerations = [
4250 client.V1Toleration(
4251 key="aws.amazon.com/neuron",
4252 operator="Equal",
4253 value="true",
4254 effect="NoSchedule",
4255 )
4256 ]
4257 else:
4258 tolerations = [
4259 client.V1Toleration(
4260 key="nvidia.com/gpu",
4261 operator="Equal",
4262 value="true",
4263 effect="NoSchedule",
4264 )
4265 ]
4267 # Node selector based on accelerator type
4268 node_selector = spec.get("node_selector", {})
4269 if gpu_count > 0 and not node_selector:
4270 if accelerator == "neuron":
4271 node_selector = {"accelerator": "neuron"}
4272 else:
4273 node_selector = {"eks.amazonaws.com/instance-gpu-manufacturer": "nvidia"}
4275 # Apply capacity type preference (spot/on-demand)
4276 capacity_type = spec.get("capacity_type")
4277 if capacity_type in ("spot", "on-demand"):
4278 node_selector["karpenter.sh/capacity-type"] = capacity_type
4280 # SGLang's prebuilt kernels need compute capability >= 8.0; the
4281 # shipped inference NodePool's cheapest fit is a T4 (g4dn), where the
4282 # server crash-loops with "no kernel image is available for execution
4283 # on the device". A required NotIn affinity keeps SGLang pods off
4284 # pre-Ampere GPUs while composing with any operator node selector
4285 # (a plain map cannot express "anything but"). Nodes without the
4286 # label are unaffected, so self-managed node groups keep working.
4287 affinity = None
4288 if runtime_framework == "sglang" and accelerator != "neuron" and gpu_count > 0:
4289 affinity = client.V1Affinity(
4290 node_affinity=client.V1NodeAffinity(
4291 required_during_scheduling_ignored_during_execution=client.V1NodeSelector(
4292 node_selector_terms=[
4293 client.V1NodeSelectorTerm(
4294 match_expressions=[
4295 client.V1NodeSelectorRequirement(
4296 key=GPU_NAME_NODE_LABEL,
4297 operator="NotIn",
4298 values=list(SGLANG_UNSUPPORTED_GPU_NAMES),
4299 )
4300 ]
4301 )
4302 ]
4303 )
4304 )
4305 )
4307 # ``--gpu-type`` is a hint, so it is a preferred node affinity rather
4308 # than a selector: Karpenter first tries to launch exactly this
4309 # instance type and relaxes the preference when it can't, so a type
4310 # that is out of capacity, or excluded by a selector or the SGLang
4311 # affinity above, never leaves the pod unschedulable.
4312 preferred_type = spec.get("gpu_type")
4313 if isinstance(preferred_type, str) and preferred_type.strip():
4314 preference = client.V1PreferredSchedulingTerm(
4315 weight=100,
4316 preference=client.V1NodeSelectorTerm(
4317 match_expressions=[
4318 client.V1NodeSelectorRequirement(
4319 key=INSTANCE_TYPE_NODE_LABEL,
4320 operator="In",
4321 values=[preferred_type.strip()],
4322 )
4323 ]
4324 ),
4325 )
4326 if affinity is None:
4327 affinity = client.V1Affinity(node_affinity=client.V1NodeAffinity())
4328 affinity.node_affinity.preferred_during_scheduling_ignored_during_execution = [
4329 preference
4330 ]
4332 labels = {
4333 "app": app_label,
4334 "project": "gco",
4335 "gco.io/type": "inference",
4336 }
4337 if extra_labels:
4338 labels.update(extra_labels)
4340 # The model container stays first: image reconciliation and every
4341 # consumer that reads ``containers[0]`` address the model server.
4342 sidecar = build_endpoint_tls_proxy(
4343 port, tls_proxy_config_map or _tls_proxy_config_map_name(name)
4344 )
4346 deployment = client.V1Deployment(
4347 metadata=client.V1ObjectMeta(
4348 name=deploy_name,
4349 namespace=namespace,
4350 labels=dict(labels),
4351 annotations=self._provenance_annotations(),
4352 ),
4353 spec=client.V1DeploymentSpec(
4354 replicas=replicas,
4355 selector=client.V1LabelSelector(
4356 match_labels={"app": app_label},
4357 ),
4358 template=client.V1PodTemplateSpec(
4359 metadata=client.V1ObjectMeta(
4360 labels=dict(labels),
4361 annotations=_endpoint_tls_pod_annotations(
4362 self._provenance_annotations(), sidecar
4363 ),
4364 ),
4365 spec=client.V1PodSpec(
4366 service_account_name="gco-service-account",
4367 automount_service_account_token=False,
4368 termination_grace_period_seconds=MODEL_POD_TERMINATION_GRACE_SECONDS,
4369 containers=[container, sidecar.container],
4370 init_containers=init_containers if init_containers else None,
4371 tolerations=tolerations,
4372 node_selector=node_selector if node_selector else None,
4373 affinity=affinity,
4374 volumes=[*volumes, *sidecar.volumes],
4375 ),
4376 ),
4377 ),
4378 )
4380 return deployment
4382 def _replica_count_for_role(self, mooncake: dict[str, Any], role: str) -> int:
4383 """Resolve the materialized replica count for a role.
4385 When per-role autoscaling is enabled and supplies a ``min_replicas``
4386 for the role, the role Deployment is materialized at that lower bound so
4387 the autoscaler owns the count from there. Otherwise the count comes from
4388 the topology: ``topology.prefill`` for prefill and ``topology.decode``
4389 for decode. The single store instance is always one replica.
4390 """
4391 autoscaling = mooncake.get("autoscaling") or {}
4392 if autoscaling.get("enabled") and role in ("prefill", "decode"):
4393 role_cfg = autoscaling.get(role) or {}
4394 min_replicas = role_cfg.get("min_replicas")
4395 if isinstance(min_replicas, int) and not isinstance(min_replicas, bool):
4396 return min_replicas
4398 topology = mooncake.get("topology") or {}
4399 if role == "prefill":
4400 return int(topology.get("prefill", 1))
4401 if role == "decode":
4402 return int(topology.get("decode", 1))
4403 # Single store instance: kv_both runs as one replica.
4404 return 1
4406 def _create_role_deployment(
4407 self,
4408 name: str,
4409 ns: str,
4410 spec: dict[str, Any],
4411 role: str,
4412 *,
4413 config_digest: str | None = None,
4414 ) -> None:
4415 """Materialize one role Deployment for a Mooncake endpoint.
4417 Disaggregated and ``both`` modes split work across ``{name}-prefill``
4418 and ``{name}-decode``; store mode runs a single ``{name}`` instance with
4419 the ``kv_both`` role. The role's ``--kv-transfer-config`` is attached to
4420 the vLLM container, EFA scheduling is applied when transfer runs over
4421 RDMA, and the replica count is taken from the topology (or the
4422 autoscaling lower bound when that is enabled).
4424 Args:
4425 name: The endpoint name.
4426 ns: The namespace to materialize into.
4427 spec: The endpoint spec; ``spec["mooncake"]`` selects the mode and
4428 topology.
4429 role: One of ``"prefill"``, ``"decode"``, or ``"single"``.
4430 config_digest: Digest of the rendered ``mooncake.json`` the pods
4431 start with, stamped on the pod template as
4432 :data:`MOONCAKE_CONFIG_DIGEST_ANNOTATION` so a later
4433 re-render is detected and rolled out.
4434 """
4435 mooncake = spec.get("mooncake") or {}
4437 # The store's single instance keeps the endpoint name; prefill and decode
4438 # are suffixed so Services and autoscalers can target each role.
4439 deploy_name = name if role == "single" else f"{name}-{role}"
4441 kv_transfer_config = build_kv_transfer_config(mooncake, role)
4442 replicas = self._replica_count_for_role(mooncake, role)
4444 deployment = self._build_inference_deployment_object(
4445 name=name,
4446 deploy_name=deploy_name,
4447 app_label=deploy_name,
4448 namespace=ns,
4449 spec=spec,
4450 replicas=replicas,
4451 extra_args=["--kv-transfer-config", kv_transfer_config],
4452 extra_labels={"gco.io/role": role},
4453 )
4455 # Land role pods on the EFA fabric when KV transfer runs over RDMA,
4456 # preserving the GPU asks already built into the pod.
4457 apply_efa_scheduling(mooncake, deployment.spec.template.spec)
4458 if config_digest is not None:
4459 template_metadata = deployment.spec.template.metadata
4460 template_metadata.annotations = {
4461 **(template_metadata.annotations or {}),
4462 MOONCAKE_CONFIG_DIGEST_ANNOTATION: config_digest,
4463 }
4465 self._assert_mutation_authority()
4466 self.apps_v1.create_namespaced_deployment(
4467 ns, deployment, _request_timeout=self._k8s_timeout
4468 )
4469 self._confirm_created_resource(
4470 kind="deployment",
4471 resource_name=deploy_name,
4472 read_resource=partial(
4473 self.apps_v1.read_namespaced_deployment,
4474 deploy_name,
4475 ns,
4476 _request_timeout=self._k8s_timeout,
4477 ),
4478 delete_resource=partial(
4479 self.apps_v1.delete_namespaced_deployment,
4480 deploy_name,
4481 ns,
4482 _request_timeout=self._k8s_timeout,
4483 ),
4484 )
4485 logger.info("Created role deployment %s/%s (role=%s)", ns, deploy_name, role)
4487 def _verify_admin_api_key_secret(self, proxy: dict[str, Any], ns: str) -> str:
4488 """Confirm the proxy admin key Secret exists and carries a key value.
4490 The proxy guards a privileged admin path and must never run without a
4491 usable ``ADMIN_API_KEY``. This reads the Secret named by
4492 ``proxy.admin_api_key_secret`` and confirms it holds a non-empty
4493 ``ADMIN_API_KEY`` value, so the value itself never has to be carried on
4494 the spec or a command argument.
4496 Args:
4497 proxy: The ``spec["mooncake"]["proxy"]`` block; ``admin_api_key_secret``
4498 names the backing Secret.
4499 ns: The namespace the Secret lives in.
4501 Returns:
4502 The verified Secret name, suitable for a Secret reference.
4504 Raises:
4505 AdminApiKeySecretError: If the spec names no Secret, the Secret is
4506 absent, or its ``ADMIN_API_KEY`` value is empty or missing.
4507 """
4508 secret_name = proxy.get("admin_api_key_secret")
4509 if not isinstance(secret_name, str) or not secret_name:
4510 raise AdminApiKeySecretError(None, "no admin API key Secret was named")
4512 try:
4513 secret = self.core_v1.read_namespaced_secret(
4514 secret_name, ns, _request_timeout=self._k8s_timeout
4515 )
4516 except ApiException as e:
4517 if e.status == 404:
4518 raise AdminApiKeySecretError(secret_name, "Secret not found") from e
4519 raise
4521 if not self._secret_has_admin_api_key(secret):
4522 raise AdminApiKeySecretError(
4523 secret_name,
4524 f"{ADMIN_API_KEY_SECRET_DATA_KEY} value is empty or missing",
4525 )
4527 return secret_name
4529 @staticmethod
4530 def _secret_has_admin_api_key(secret: client.V1Secret) -> bool:
4531 """Return whether ``secret`` carries a non-empty ``ADMIN_API_KEY``.
4533 Both the base64 ``data`` and the plaintext ``string_data`` views are
4534 considered, and a value is treated as present only when it decodes to a
4535 non-empty string.
4536 """
4537 string_data = secret.string_data or {}
4538 plain = string_data.get(ADMIN_API_KEY_SECRET_DATA_KEY)
4539 if plain:
4540 return True
4542 data = secret.data or {}
4543 encoded = data.get(ADMIN_API_KEY_SECRET_DATA_KEY)
4544 if not encoded:
4545 return False
4546 try:
4547 return bool(base64.b64decode(encoded))
4548 except ValueError, TypeError:
4549 # A value that cannot be decoded is unusable as an admin key.
4550 return False
4552 def _ensure_admin_api_key_secret(
4553 self,
4554 name: str,
4555 proxy: dict[str, Any],
4556 ns: str,
4557 lifecycle_id: str,
4558 ) -> str:
4559 """Return the proxy admin-key Secret name, provisioning one if needed.
4561 The prefill-decode proxy guards a privileged admin path and must never
4562 run without a usable ``ADMIN_API_KEY``. Two paths satisfy that:
4564 - **Bring-your-own**: when the proxy block names a Secret, that Secret
4565 must already exist and carry a non-empty ``ADMIN_API_KEY``; otherwise
4566 the deployment is rejected, so a typo or a missing pre-created Secret
4567 fails fast. The named Secret is only read, never created or mutated.
4568 - **Auto-managed**: when the proxy names no Secret, a per-endpoint
4569 ``{name}-admin`` Secret is provisioned create-if-absent with a
4570 generated key, so a split deploy needs no manual Secret. The generated
4571 key only ever lives in the cluster — it is never written to the
4572 endpoint spec, a command argument, or a log line.
4574 Args:
4575 name: The endpoint name, used to derive the auto-managed Secret name.
4576 proxy: The ``spec["mooncake"]["proxy"]`` block.
4577 ns: The namespace the Secret lives in.
4579 Returns:
4580 The Secret name to reference from the proxy container.
4582 Raises:
4583 AdminApiKeySecretError: Only on the bring-your-own path, when the
4584 named Secret is absent or its ``ADMIN_API_KEY`` is empty. The
4585 auto-managed path never raises this.
4586 """
4587 named = proxy.get("admin_api_key_secret")
4588 if isinstance(named, str) and named:
4589 return self._verify_admin_api_key_secret(proxy, ns)
4590 return self._provision_admin_api_key_secret(f"{name}-admin", ns, lifecycle_id)
4592 def _provision_admin_api_key_secret(
4593 self,
4594 secret_name: str,
4595 ns: str,
4596 lifecycle_id: str,
4597 ) -> str:
4598 """Create the auto-managed proxy admin-key Secret if absent.
4600 Uses create-if-absent semantics so the key stays stable across reconcile
4601 passes: an existing Secret (the steady state, or one a prior pass
4602 created) is left untouched, and a concurrent create (409) is treated as
4603 success. A freshly created Secret carries a cryptographically strong
4604 64-character hex ``ADMIN_API_KEY`` from :func:`secrets.token_hex`, which
4605 reaches the proxy only through a Secret reference — the value is never
4606 logged or written to the spec.
4608 Args:
4609 secret_name: The Secret to ensure exists (``{endpoint}-admin``).
4610 ns: The namespace to create it in.
4612 Returns:
4613 The Secret name, ready for a Secret reference.
4614 """
4615 try:
4616 existing = self.core_v1.read_namespaced_secret(
4617 secret_name,
4618 ns,
4619 _request_timeout=self._k8s_timeout,
4620 )
4621 except ApiException as e:
4622 if e.status != 404:
4623 raise
4624 else:
4625 existing = self._authorize_resource(
4626 existing,
4627 kind="secret",
4628 resource_name=secret_name,
4629 patch_metadata=partial(
4630 self.core_v1.patch_namespaced_secret,
4631 secret_name,
4632 ns,
4633 _request_timeout=self._k8s_timeout,
4634 ),
4635 read_resource=lambda: self.core_v1.read_namespaced_secret(
4636 secret_name, ns, _request_timeout=self._k8s_timeout
4637 ),
4638 delete_resource=partial(
4639 self.core_v1.delete_namespaced_secret,
4640 secret_name,
4641 ns,
4642 _request_timeout=self._k8s_timeout,
4643 ),
4644 )
4645 self._require_owned_admin_secret(
4646 existing,
4647 secret_name,
4648 ns,
4649 lifecycle_id,
4650 )
4651 return secret_name
4653 ownership = self._generated_admin_secret_labels(secret_name)
4654 annotations = self._provenance_annotations() or {_LIFECYCLE_ANNOTATION: lifecycle_id}
4655 secret = client.V1Secret(
4656 metadata=client.V1ObjectMeta(
4657 name=secret_name,
4658 namespace=ns,
4659 labels=ownership,
4660 annotations=annotations,
4661 ),
4662 string_data={ADMIN_API_KEY_SECRET_DATA_KEY: secrets.token_hex(32)},
4663 type="Opaque",
4664 )
4665 # The two logger.info calls below carry a bare `# nosemgrep`: the
4666 # logger-credential-disclosure rule matches the literal word "Secret" in
4667 # the message, but only the Secret's name and namespace (%s/%s) are
4668 # logged here — never the generated key value set above in string_data.
4669 self._assert_mutation_authority()
4670 try:
4671 self.core_v1.create_namespaced_secret(ns, secret, _request_timeout=self._k8s_timeout)
4672 self._confirm_created_resource(
4673 kind="secret",
4674 resource_name=secret_name,
4675 read_resource=partial(
4676 self.core_v1.read_namespaced_secret,
4677 secret_name,
4678 ns,
4679 _request_timeout=self._k8s_timeout,
4680 ),
4681 delete_resource=partial(
4682 self.core_v1.delete_namespaced_secret,
4683 secret_name,
4684 ns,
4685 _request_timeout=self._k8s_timeout,
4686 ),
4687 )
4688 logger.info("Provisioned proxy admin-key Secret %s/%s", ns, secret_name) # nosemgrep
4689 except ApiException as e:
4690 if e.status != 409:
4691 raise
4692 existing = self.core_v1.read_namespaced_secret(
4693 secret_name,
4694 ns,
4695 _request_timeout=self._k8s_timeout,
4696 )
4697 existing = self._authorize_resource(
4698 existing,
4699 kind="secret",
4700 resource_name=secret_name,
4701 patch_metadata=partial(
4702 self.core_v1.patch_namespaced_secret,
4703 secret_name,
4704 ns,
4705 _request_timeout=self._k8s_timeout,
4706 ),
4707 read_resource=lambda: self.core_v1.read_namespaced_secret(
4708 secret_name, ns, _request_timeout=self._k8s_timeout
4709 ),
4710 delete_resource=partial(
4711 self.core_v1.delete_namespaced_secret,
4712 secret_name,
4713 ns,
4714 _request_timeout=self._k8s_timeout,
4715 ),
4716 )
4717 try:
4718 self._require_owned_admin_secret(
4719 existing,
4720 secret_name,
4721 ns,
4722 lifecycle_id,
4723 )
4724 except AdminApiKeySecretError as error:
4725 raise AdminApiKeySecretError(
4726 secret_name,
4727 "a concurrent conventional Secret lacks matching lifecycle provenance",
4728 ) from error
4729 logger.info("Proxy admin-key Secret %s/%s exists", ns, secret_name) # nosemgrep
4730 return secret_name
4732 def _create_pd_proxy(
4733 self, name: str, ns: str, spec: dict[str, Any], endpoint: dict[str, Any]
4734 ) -> None:
4735 """Materialize the prefill-decode proxy front for a disaggregated endpoint.
4737 Disaggregated and ``both`` modes are fronted by a lightweight proxy that
4738 runs the residency check and dispatches each request to the prefill and
4739 decode pods. It materializes a ConfigMap, a proxy Deployment with at
4740 least one replica, and a Service whose selector matches only the proxy
4741 pods. The shared HTTPRoute attached to ``gco-system/gco-gateway`` sends
4742 ``/inference`` to ``gco-system/inference-proxy``; that authenticated
4743 platform proxy then reaches this internal ClusterIP Service.
4744 Endpoint-specific Ingresses are removed as an unsafe legacy path.
4746 Every hop is verified HTTPS: the router binds pod loopback behind the
4747 pod's ``endpoint-tls-proxy`` sidecar (the Service publishes only that
4748 sidecar's port), and dials the role Services' sidecars by their fully
4749 qualified names, trusting only the GCO internal CA projected from the
4750 endpoint wildcard Secret.
4752 Before those resources are created, a user-named
4753 ``mooncake.proxy.admin_api_key_secret`` is verified to contain a usable
4754 ``ADMIN_API_KEY``. When no Secret is named, the monitor auto-provisions
4755 a generated ``{name}-admin`` Secret instead. A missing or empty named
4756 Secret rejects the proxy; the key itself reaches the container only as
4757 a Secret reference at pod start and is never written to the spec or a
4758 command argument.
4760 Creation is idempotent at the API boundary: an already-present
4761 Deployment is merge-patched to the desired shape and an already-present
4762 ClusterIP Service is moved to the TLS port if it predates it, while
4763 historical direct Ingresses are deleted if present. No endpoint Gateway
4764 or HTTPRoute is created.
4766 Args:
4767 name: The endpoint name.
4768 ns: The namespace to materialize into.
4769 spec: The endpoint spec; ``spec["mooncake"]`` supplies the proxy
4770 image and behavior.
4771 endpoint: The endpoint record. Legacy per-endpoint routing metadata
4772 is ignored because the shared platform HTTPRoute owns the prefix.
4774 Raises:
4775 AdminApiKeySecretError: If the admin key Secret is missing, names no
4776 Secret, or holds an empty ``ADMIN_API_KEY``. No proxy resource
4777 is created in that case.
4778 """
4779 mooncake = spec.get("mooncake") or {}
4780 proxy = mooncake.get("proxy") or {}
4781 proxy_name = f"{name}-proxy"
4782 lifecycle_id = endpoint.get("lifecycle_id")
4783 if not isinstance(lifecycle_id, str) or not lifecycle_id:
4784 raise AdminApiKeySecretError(
4785 None,
4786 "endpoint record has no immutable lifecycle identity",
4787 )
4789 # The proxy fronts a privileged admin path, so it never starts without a
4790 # usable admin key. When the spec names a Secret it must already exist
4791 # and be non-empty (the deployment is rejected otherwise); when it names
4792 # none, a per-endpoint admin-key Secret is auto-provisioned with a
4793 # generated key. Either way the key reaches the container only by Secret
4794 # reference.
4795 admin_secret_name = self._ensure_admin_api_key_secret(
4796 name,
4797 proxy,
4798 ns,
4799 lifecycle_id,
4800 )
4802 proxy_env = build_pd_proxy_config(mooncake)
4803 container_env = [client.V1EnvVar(name=k, value=v) for k, v in proxy_env.items()]
4804 # Deliver the admin key by Secret reference only — its value is never
4805 # placed on the spec or a command argument.
4806 container_env.append(
4807 client.V1EnvVar(
4808 name=PD_PROXY_ADMIN_API_KEY_ENV,
4809 value_from=client.V1EnvVarSource(
4810 secret_key_ref=client.V1SecretKeySelector(
4811 name=admin_secret_name,
4812 key=ADMIN_API_KEY_SECRET_DATA_KEY,
4813 )
4814 ),
4815 )
4816 )
4818 # The proxy fronts at least one replica; a spec may ask for more.
4819 replicas = proxy.get("replicas", 1)
4820 if not isinstance(replicas, int) or isinstance(replicas, bool) or replicas < 1:
4821 replicas = 1
4823 labels = {
4824 "app": proxy_name,
4825 "project": "gco",
4826 "gco.io/type": "inference",
4827 "gco.io/role": PD_PROXY_ROLE_LABEL,
4828 }
4830 # The proxy reaches prefill and decode through their per-role Services
4831 # over verified HTTPS (their TLS sidecars, trusting only the GCO
4832 # internal CA) and listens on loopback PD_PROXY_PORT behind its own
4833 # sidecar for requests from the authenticated API proxy. Routing via
4834 # Services means only Ready role pods receive traffic.
4835 container_env.extend(
4836 [
4837 client.V1EnvVar(name=PD_PROXY_HOST_ENV, value=PD_PROXY_BIND_HOST),
4838 client.V1EnvVar(name=PD_PROXY_PORT_ENV, value=str(PD_PROXY_PORT)),
4839 client.V1EnvVar(
4840 name=PD_PROXY_PREFILL_URL_ENV, value=_role_service_url(name, ns, "prefill")
4841 ),
4842 client.V1EnvVar(
4843 name=PD_PROXY_DECODE_URL_ENV, value=_role_service_url(name, ns, "decode")
4844 ),
4845 client.V1EnvVar(name=PD_PROXY_CA_FILE_ENV, value=PD_PROXY_CA_FILE),
4846 ]
4847 )
4849 # Ship the proxy program to the pod as a ConfigMap and run it from there.
4850 self._ensure_pd_proxy_configmap(name, ns)
4851 proxy_volume_name = "pd-proxy-script"
4853 container = client.V1Container(
4854 name="proxy",
4855 image=proxy.get("image"),
4856 command=["python3", PD_PROXY_SCRIPT_PATH],
4857 ports=[client.V1ContainerPort(container_port=PD_PROXY_PORT)],
4858 env=container_env if container_env else None,
4859 resources=client.V1ResourceRequirements(
4860 requests={"cpu": "250m", "memory": "256Mi"},
4861 limits={"cpu": "1", "memory": "1Gi"},
4862 ),
4863 volume_mounts=[
4864 client.V1VolumeMount(
4865 name=proxy_volume_name,
4866 mount_path=PD_PROXY_CONFIG_MOUNT_DIR,
4867 read_only=True,
4868 ),
4869 client.V1VolumeMount(
4870 name=INTERNAL_CA_VOLUME,
4871 mount_path=INTERNAL_CA_MOUNT_DIR,
4872 read_only=True,
4873 ),
4874 ],
4875 readiness_probe=_pd_proxy_health_probe(liveness=False),
4876 liveness_probe=_pd_proxy_health_probe(liveness=True),
4877 )
4878 sidecar = build_endpoint_tls_proxy(PD_PROXY_PORT, _tls_proxy_config_map_name(name))
4880 deployment = client.V1Deployment(
4881 metadata=client.V1ObjectMeta(
4882 name=proxy_name,
4883 namespace=ns,
4884 labels=dict(labels),
4885 annotations=self._provenance_annotations(),
4886 ),
4887 spec=client.V1DeploymentSpec(
4888 replicas=replicas,
4889 selector=client.V1LabelSelector(match_labels={"app": proxy_name}),
4890 template=client.V1PodTemplateSpec(
4891 metadata=client.V1ObjectMeta(
4892 labels=dict(labels),
4893 annotations=_endpoint_tls_pod_annotations(
4894 self._provenance_annotations(), sidecar
4895 ),
4896 ),
4897 spec=client.V1PodSpec(
4898 service_account_name="gco-service-account",
4899 automount_service_account_token=False,
4900 termination_grace_period_seconds=MODEL_POD_TERMINATION_GRACE_SECONDS,
4901 containers=[container, sidecar.container],
4902 volumes=[
4903 client.V1Volume(
4904 name=proxy_volume_name,
4905 config_map=client.V1ConfigMapVolumeSource(
4906 name=f"{name}-pd-proxy",
4907 default_mode=0o555,
4908 ),
4909 ),
4910 # A separate projection of only the CA certificate
4911 # from the wildcard Secret the sidecar mounts whole.
4912 client.V1Volume(
4913 name=INTERNAL_CA_VOLUME,
4914 secret=client.V1SecretVolumeSource(
4915 secret_name=ENDPOINT_TLS_SECRET,
4916 items=[
4917 client.V1KeyToPath(
4918 key=INTERNAL_CA_KEY,
4919 path=INTERNAL_CA_KEY,
4920 )
4921 ],
4922 default_mode=0o444,
4923 ),
4924 ),
4925 *sidecar.volumes,
4926 ],
4927 ),
4928 ),
4929 ),
4930 )
4932 self._assert_mutation_authority()
4933 try:
4934 self.apps_v1.create_namespaced_deployment(
4935 ns, deployment, _request_timeout=self._k8s_timeout
4936 )
4937 self._confirm_created_resource(
4938 kind="deployment",
4939 resource_name=proxy_name,
4940 read_resource=partial(
4941 self.apps_v1.read_namespaced_deployment,
4942 proxy_name,
4943 ns,
4944 _request_timeout=self._k8s_timeout,
4945 ),
4946 delete_resource=partial(
4947 self.apps_v1.delete_namespaced_deployment,
4948 proxy_name,
4949 ns,
4950 _request_timeout=self._k8s_timeout,
4951 ),
4952 )
4953 logger.info("Created proxy deployment %s/%s", ns, proxy_name)
4954 except ApiException as error:
4955 if error.status != 409:
4956 raise
4957 existing = self._get_deployment(proxy_name, ns)
4958 if existing is None:
4959 raise ReconcileFencedError(
4960 "proxy deployment disappeared during reconciliation"
4961 ) from error
4962 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
4963 deployment.metadata.resource_version = resource_version
4964 self._assert_mutation_authority()
4965 # A JSON merge patch replaces the containers and volumes lists
4966 # wholesale (a strategic merge would key them by name and merge each
4967 # entry, keeping a stale tcpSocket beside the new exec handler,
4968 # which the API rejects). Maps still merge, so annotations such as
4969 # a rollout-restart stamp survive.
4970 self.apps_v1.patch_namespaced_deployment(
4971 proxy_name,
4972 ns,
4973 body=deployment,
4974 _content_type="application/merge-patch+json",
4975 _request_timeout=self._k8s_timeout,
4976 )
4977 logger.info("Reconciled proxy deployment %s/%s", ns, proxy_name)
4979 self._create_proxy_service(proxy_name, ns)
4981 def _authorize_existing_service(self, name: str, namespace: str) -> Any:
4982 service = self.core_v1.read_namespaced_service(
4983 name, namespace, _request_timeout=self._k8s_timeout
4984 )
4985 return self._authorize_resource(
4986 service,
4987 kind="service",
4988 resource_name=name,
4989 patch_metadata=partial(
4990 self.core_v1.patch_namespaced_service,
4991 name,
4992 namespace,
4993 _request_timeout=self._k8s_timeout,
4994 ),
4995 read_resource=lambda: self.core_v1.read_namespaced_service(
4996 name, namespace, _request_timeout=self._k8s_timeout
4997 ),
4998 delete_resource=partial(
4999 self.core_v1.delete_namespaced_service,
5000 name,
5001 namespace,
5002 _request_timeout=self._k8s_timeout,
5003 ),
5004 )
5006 def _converge_service_ports(self, name: str, namespace: str, service: Any) -> bool:
5007 """Move an authorized Service that predates the TLS sidecar onto it.
5009 A Service created before the sidecar existed publishes plain HTTP (port
5010 80 or 8000 straight to the model). It is patched in place, so its
5011 ClusterIP and DNS name never change, to publish only the sidecar port.
5012 A JSON merge patch replaces the whole ``ports`` list: a strategic merge
5013 keys ports by number and would keep the old unnamed port beside the new
5014 one, which the API rejects (a multi-port Service must name every port).
5015 The observed resourceVersion makes the patch a compare-and-swap.
5017 Returns:
5018 ``True`` when the Service was patched.
5019 """
5020 if _service_publishes_only_tls(service):
5021 return False
5022 _metadata, _annotations, _uid, resource_version = self._object_metadata(service)
5023 body: dict[str, Any] = {"spec": {"ports": _endpoint_service_ports()}}
5024 if resource_version:
5025 body["metadata"] = {"resourceVersion": resource_version}
5026 self._assert_mutation_authority()
5027 self.core_v1.patch_namespaced_service(
5028 name,
5029 namespace,
5030 body=body,
5031 _content_type="application/merge-patch+json",
5032 _request_timeout=self._k8s_timeout,
5033 )
5034 logger.info(
5035 "Moved service %s/%s to the TLS sidecar port %d", namespace, name, ENDPOINT_TLS_PORT
5036 )
5037 return True
5039 def _create_role_service(self, name: str, ns: str, role: str, port: int = 8000) -> None:
5040 """Create the ClusterIP Service that fronts one role's pods.
5042 Named ``{name}-{role}`` and selecting that role Deployment's app label,
5043 so the PD proxy can address prefill or decode by stable in-cluster DNS.
5044 Routing through a Service means kube-proxy load-balances across only the
5045 role's Ready pods, which is what gives the proxy ready-only decode
5046 routing without watching the Kubernetes API. Idempotent at the API
5047 boundary: an already-present Service is kept and moved onto the TLS
5048 port if it predates it.
5050 The Service publishes only the role pods' TLS sidecar port. ``port`` is
5051 the role pods' model-server port, which the Service deliberately no
5052 longer exposes (the sidecar forwards to it over pod loopback); it is
5053 accepted and ignored so existing callers keep their signature.
5054 """
5055 del port
5056 deploy_name = f"{name}-{role}"
5057 service = client.V1Service(
5058 metadata=client.V1ObjectMeta(
5059 name=deploy_name,
5060 namespace=ns,
5061 labels={
5062 "app": deploy_name,
5063 "project": "gco",
5064 "gco.io/type": "inference",
5065 "gco.io/role": role,
5066 },
5067 annotations=self._provenance_annotations(),
5068 ),
5069 spec=client.V1ServiceSpec(
5070 selector=_inference_service_selector(deploy_name),
5071 ports=_endpoint_service_ports(),
5072 type="ClusterIP",
5073 ),
5074 )
5075 self._assert_mutation_authority()
5076 try:
5077 self.core_v1.create_namespaced_service(ns, service, _request_timeout=self._k8s_timeout)
5078 self._confirm_created_resource(
5079 kind="service",
5080 resource_name=deploy_name,
5081 read_resource=partial(
5082 self.core_v1.read_namespaced_service,
5083 deploy_name,
5084 ns,
5085 _request_timeout=self._k8s_timeout,
5086 ),
5087 delete_resource=partial(
5088 self.core_v1.delete_namespaced_service,
5089 deploy_name,
5090 ns,
5091 _request_timeout=self._k8s_timeout,
5092 ),
5093 )
5094 logger.info("Created role service %s/%s", ns, deploy_name)
5095 except ApiException as error:
5096 if error.status != 409:
5097 raise
5098 existing = self._authorize_existing_service(deploy_name, ns)
5099 self._converge_service_ports(deploy_name, ns, existing)
5100 logger.info("Role service %s/%s already exists", ns, deploy_name)
5102 def _ensure_pd_proxy_configmap(self, name: str, ns: str) -> None:
5103 """Publish the PD proxy program to the pod as a ConfigMap.
5105 The proxy program (``mooncake_pd_proxy.py``) ships in this image
5106 alongside the monitor; its source is read here and mounted into the
5107 ``{name}-proxy`` pod, which runs it with ``python3`` from
5108 ``PD_PROXY_SCRIPT_PATH``. The ConfigMap is patched on conflict so the
5109 program tracks the running monitor build.
5110 """
5111 script = (Path(__file__).resolve().parent / PD_PROXY_SCRIPT_FILENAME).read_text(
5112 encoding="utf-8"
5113 )
5114 cm_name = f"{name}-pd-proxy"
5115 body = client.V1ConfigMap(
5116 metadata=client.V1ObjectMeta(
5117 name=cm_name,
5118 namespace=ns,
5119 labels={
5120 "app": f"{name}-proxy",
5121 "project": "gco",
5122 "gco.io/type": "inference",
5123 "gco.io/role": PD_PROXY_ROLE_LABEL,
5124 },
5125 annotations=self._provenance_annotations(),
5126 ),
5127 data={PD_PROXY_SCRIPT_FILENAME: script},
5128 )
5129 self._assert_mutation_authority()
5130 try:
5131 self.core_v1.create_namespaced_config_map(ns, body, _request_timeout=self._k8s_timeout)
5132 self._confirm_created_resource(
5133 kind="configmap",
5134 resource_name=cm_name,
5135 read_resource=partial(
5136 self.core_v1.read_namespaced_config_map,
5137 cm_name,
5138 ns,
5139 _request_timeout=self._k8s_timeout,
5140 ),
5141 delete_resource=partial(
5142 self.core_v1.delete_namespaced_config_map,
5143 cm_name,
5144 ns,
5145 _request_timeout=self._k8s_timeout,
5146 ),
5147 )
5148 logger.info("Created PD proxy ConfigMap %s/%s", ns, cm_name)
5149 except ApiException as error:
5150 if error.status != 409:
5151 raise
5152 existing = self.core_v1.read_namespaced_config_map(
5153 cm_name, ns, _request_timeout=self._k8s_timeout
5154 )
5155 existing = self._authorize_resource(
5156 existing,
5157 kind="configmap",
5158 resource_name=cm_name,
5159 patch_metadata=partial(
5160 self.core_v1.patch_namespaced_config_map,
5161 cm_name,
5162 ns,
5163 _request_timeout=self._k8s_timeout,
5164 ),
5165 read_resource=lambda: self.core_v1.read_namespaced_config_map(
5166 cm_name, ns, _request_timeout=self._k8s_timeout
5167 ),
5168 delete_resource=partial(
5169 self.core_v1.delete_namespaced_config_map,
5170 cm_name,
5171 ns,
5172 _request_timeout=self._k8s_timeout,
5173 ),
5174 )
5175 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
5176 body.metadata.resource_version = resource_version
5177 self._assert_mutation_authority()
5178 self.core_v1.patch_namespaced_config_map(
5179 cm_name, ns, body, _request_timeout=self._k8s_timeout
5180 )
5181 logger.info("Updated PD proxy ConfigMap %s/%s", ns, cm_name)
5183 def _ensure_tls_proxy_configmap(self, name: str, ns: str) -> None:
5184 """Publish the TLS sidecar program for every pod of one endpoint.
5186 ``gco/services/tls_proxy.py`` ships in this image; its source is read
5187 here on every call (never copied) and published as the endpoint's
5188 ``{name}-tls-proxy`` ConfigMap, which each of the endpoint's model pods
5189 mounts at ``ENDPOINT_TLS_PROXY_CONFIG_MOUNT_DIR`` and runs with
5190 ``python3``. The ConfigMap is read first and written only when absent or
5191 stale, so the steady state costs one read per pass.
5193 A changed program reaches running sidecars on their next start: the
5194 source is deliberately not part of the pod-template digest, because a
5195 monitor upgrade must not roll every GPU model pod (each needs spare GPU
5196 capacity to surge and minutes to reload its model) for a sidecar change.
5197 """
5198 script = (Path(__file__).resolve().parent / ENDPOINT_TLS_PROXY_SCRIPT_FILENAME).read_text(
5199 encoding="utf-8"
5200 )
5201 cm_name = _tls_proxy_config_map_name(name)
5202 body = client.V1ConfigMap(
5203 metadata=client.V1ObjectMeta(
5204 name=cm_name,
5205 namespace=ns,
5206 labels={"app": name, "project": "gco", "gco.io/type": "inference"},
5207 annotations=self._provenance_annotations(),
5208 ),
5209 data={ENDPOINT_TLS_PROXY_SCRIPT_FILENAME: script},
5210 )
5211 read_config_map = partial(
5212 self.core_v1.read_namespaced_config_map,
5213 cm_name,
5214 ns,
5215 _request_timeout=self._k8s_timeout,
5216 )
5217 delete_config_map = partial(
5218 self.core_v1.delete_namespaced_config_map,
5219 cm_name,
5220 ns,
5221 _request_timeout=self._k8s_timeout,
5222 )
5223 try:
5224 existing = read_config_map()
5225 except ApiException as error:
5226 if error.status != 404:
5227 raise
5228 self._assert_mutation_authority()
5229 try:
5230 self.core_v1.create_namespaced_config_map(
5231 ns, body, _request_timeout=self._k8s_timeout
5232 )
5233 except ApiException as create_error:
5234 if create_error.status != 409:
5235 raise
5236 # A concurrent writer created it between the read and the
5237 # create; converge on whatever it wrote below.
5238 existing = read_config_map()
5239 else:
5240 self._confirm_created_resource(
5241 kind="configmap",
5242 resource_name=cm_name,
5243 read_resource=read_config_map,
5244 delete_resource=delete_config_map,
5245 )
5246 logger.info("Created TLS proxy ConfigMap %s/%s", ns, cm_name)
5247 return
5249 existing = self._authorize_resource(
5250 existing,
5251 kind="configmap",
5252 resource_name=cm_name,
5253 patch_metadata=partial(
5254 self.core_v1.patch_namespaced_config_map,
5255 cm_name,
5256 ns,
5257 _request_timeout=self._k8s_timeout,
5258 ),
5259 read_resource=read_config_map,
5260 delete_resource=delete_config_map,
5261 )
5262 if getattr(existing, "data", None) == body.data:
5263 return
5264 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
5265 body.metadata.resource_version = resource_version
5266 self._assert_mutation_authority()
5267 self.core_v1.patch_namespaced_config_map(
5268 cm_name, ns, body, _request_timeout=self._k8s_timeout
5269 )
5270 logger.info("Updated TLS proxy ConfigMap %s/%s", ns, cm_name)
5272 def _converge_endpoint_tls_proxy(
5273 self,
5274 deploy_name: str,
5275 namespace: str,
5276 deployment: Any,
5277 *,
5278 config_map_name: str,
5279 fallback_port: int,
5280 ) -> bool:
5281 """Give an existing model Deployment the current TLS sidecar, in place.
5283 Deployments created before the sidecar existed (or rendered with a
5284 different sidecar image or upstream port) are patched when the live pod
5285 template lacks the sidecar or carries a different
5286 ``ENDPOINT_TLS_PROXY_DIGEST_ANNOTATION``. A strategic merge adds the
5287 sidecar container and its volumes by name, merges the annotations, and
5288 leaves the model container untouched: its image, port, and probes stay
5289 exactly as reconciled. The observed resourceVersion makes the patch a
5290 compare-and-swap; the resulting rollout replaces pods like any other
5291 template change.
5293 Args:
5294 deploy_name: The Deployment to converge.
5295 namespace: Its namespace.
5296 deployment: The authorized live Deployment.
5297 config_map_name: The endpoint's TLS program ConfigMap.
5298 fallback_port: The spec port, used when the live template does not
5299 show the model container's port.
5301 Returns:
5302 ``True`` when the Deployment was patched.
5304 Raises:
5305 ValueError: If the model serves on the sidecar's own port.
5306 """
5307 template = getattr(getattr(deployment, "spec", None), "template", None)
5308 raw_annotations = getattr(getattr(template, "metadata", None), "annotations", None)
5309 annotations = dict(raw_annotations) if isinstance(raw_annotations, dict) else {}
5310 raw_containers = getattr(getattr(template, "spec", None), "containers", None)
5311 containers = list(raw_containers) if isinstance(raw_containers, list) else []
5313 upstream_port = _model_server_port(containers, fallback_port)
5314 _reject_tls_port_collision(upstream_port)
5315 sidecar = build_endpoint_tls_proxy(upstream_port, config_map_name)
5316 has_sidecar = any(
5317 getattr(container, "name", None) == ENDPOINT_TLS_PROXY_CONTAINER
5318 for container in containers
5319 )
5320 if has_sidecar and annotations.get(ENDPOINT_TLS_PROXY_DIGEST_ANNOTATION) == sidecar.digest:
5321 return False
5323 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
5324 body: dict[str, Any] = {
5325 "spec": {
5326 "template": {
5327 "metadata": {
5328 "annotations": {
5329 SKIP_CONTAINERS_ANNOTATION: _merge_skip_containers(
5330 annotations.get(SKIP_CONTAINERS_ANNOTATION)
5331 ),
5332 ENDPOINT_TLS_PROXY_DIGEST_ANNOTATION: sidecar.digest,
5333 }
5334 },
5335 "spec": {
5336 "terminationGracePeriodSeconds": MODEL_POD_TERMINATION_GRACE_SECONDS,
5337 "containers": [sidecar.container],
5338 "volumes": list(sidecar.volumes),
5339 },
5340 }
5341 }
5342 }
5343 if resource_version:
5344 body["metadata"] = {"resourceVersion": resource_version}
5345 self._assert_mutation_authority()
5346 self.apps_v1.patch_namespaced_deployment(
5347 deploy_name,
5348 namespace,
5349 body=body,
5350 _request_timeout=self._k8s_timeout,
5351 )
5352 logger.info("Rolled the TLS sidecar into deployment %s/%s", namespace, deploy_name)
5353 return True
5355 def _converge_mooncake_config_digest(
5356 self,
5357 deploy_name: str,
5358 namespace: str,
5359 deployment: Any,
5360 config_digest: str,
5361 ) -> bool:
5362 """Roll a Mooncake role Deployment onto a re-rendered ``mooncake.json``.
5364 The connectors read the mounted configuration once at start-up, so a
5365 ConfigMap re-render after ``configure-store`` (or any transport change)
5366 would otherwise never reach running pods. When the pod template's
5367 :data:`MOONCAKE_CONFIG_DIGEST_ANNOTATION` differs from ``config_digest``
5368 (including Deployments created before the annotation existed) the
5369 template annotation is patched, which starts an ordinary rolling update
5370 onto pods that mount the current file. The observed resourceVersion makes
5371 the patch a compare-and-swap.
5373 Returns:
5374 ``True`` when the Deployment was patched.
5375 """
5376 template = getattr(getattr(deployment, "spec", None), "template", None)
5377 raw_annotations = getattr(getattr(template, "metadata", None), "annotations", None)
5378 annotations = dict(raw_annotations) if isinstance(raw_annotations, dict) else {}
5379 if annotations.get(MOONCAKE_CONFIG_DIGEST_ANNOTATION) == config_digest:
5380 return False
5381 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
5382 body: dict[str, Any] = {
5383 "spec": {
5384 "template": {
5385 "metadata": {"annotations": {MOONCAKE_CONFIG_DIGEST_ANNOTATION: config_digest}}
5386 }
5387 }
5388 }
5389 if resource_version:
5390 body["metadata"] = {"resourceVersion": resource_version}
5391 self._assert_mutation_authority()
5392 self.apps_v1.patch_namespaced_deployment(
5393 deploy_name,
5394 namespace,
5395 body=body,
5396 _request_timeout=self._k8s_timeout,
5397 )
5398 logger.info(
5399 "Rolled deployment %s/%s onto re-rendered mooncake.json (%s)",
5400 namespace,
5401 deploy_name,
5402 config_digest[:12],
5403 )
5404 return True
5406 def _create_proxy_service(self, proxy_name: str, namespace: str) -> None:
5407 """Create the Service that fronts only the proxy pods.
5409 The selector is the ``{name}-proxy`` app label together with the proxy
5410 role marker, so the Service resolves exclusively to proxy pods and never
5411 to the prefill or decode role pods that share the namespace. It
5412 publishes only the proxy pods' TLS sidecar port.
5413 """
5414 service = client.V1Service(
5415 metadata=client.V1ObjectMeta(
5416 name=proxy_name,
5417 namespace=namespace,
5418 labels={
5419 "app": proxy_name,
5420 "project": "gco",
5421 "gco.io/type": "inference",
5422 "gco.io/role": PD_PROXY_ROLE_LABEL,
5423 },
5424 annotations=self._provenance_annotations(),
5425 ),
5426 spec=client.V1ServiceSpec(
5427 selector=_inference_service_selector(
5428 proxy_name, **{"gco.io/role": PD_PROXY_ROLE_LABEL}
5429 ),
5430 ports=_endpoint_service_ports(),
5431 type="ClusterIP",
5432 ),
5433 )
5435 self._assert_mutation_authority()
5436 try:
5437 self.core_v1.create_namespaced_service(
5438 namespace, service, _request_timeout=self._k8s_timeout
5439 )
5440 self._confirm_created_resource(
5441 kind="service",
5442 resource_name=proxy_name,
5443 read_resource=partial(
5444 self.core_v1.read_namespaced_service,
5445 proxy_name,
5446 namespace,
5447 _request_timeout=self._k8s_timeout,
5448 ),
5449 delete_resource=partial(
5450 self.core_v1.delete_namespaced_service,
5451 proxy_name,
5452 namespace,
5453 _request_timeout=self._k8s_timeout,
5454 ),
5455 )
5456 logger.info("Created proxy service %s/%s", namespace, proxy_name)
5457 except ApiException as error:
5458 if error.status != 409:
5459 raise
5460 existing = self._authorize_existing_service(proxy_name, namespace)
5461 self._converge_service_ports(proxy_name, namespace, existing)
5462 logger.info("Proxy service %s/%s already exists", namespace, proxy_name)
5464 def _create_service(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
5465 """Create the internal ClusterIP Service for an inference endpoint.
5467 The Service publishes only the pods' TLS sidecar port, whatever port
5468 the model serves on, so ``spec`` no longer shapes it; the parameter is
5469 kept so every caller shares one signature with :meth:`_ensure_service`.
5470 """
5471 del spec
5473 service = client.V1Service(
5474 metadata=client.V1ObjectMeta(
5475 name=name,
5476 namespace=namespace,
5477 labels={
5478 "app": name,
5479 "project": "gco",
5480 "gco.io/type": "inference",
5481 },
5482 annotations=self._provenance_annotations(),
5483 ),
5484 spec=client.V1ServiceSpec(
5485 selector=_inference_service_selector(name),
5486 ports=_endpoint_service_ports(),
5487 type="ClusterIP",
5488 ),
5489 )
5491 self._assert_mutation_authority()
5492 try:
5493 self.core_v1.create_namespaced_service(
5494 namespace, service, _request_timeout=self._k8s_timeout
5495 )
5496 self._confirm_created_resource(
5497 kind="service",
5498 resource_name=name,
5499 read_resource=partial(
5500 self.core_v1.read_namespaced_service,
5501 name,
5502 namespace,
5503 _request_timeout=self._k8s_timeout,
5504 ),
5505 delete_resource=partial(
5506 self.core_v1.delete_namespaced_service,
5507 name,
5508 namespace,
5509 _request_timeout=self._k8s_timeout,
5510 ),
5511 )
5512 logger.info("Created service %s/%s", namespace, name)
5513 except ApiException as error:
5514 if error.status != 409:
5515 raise
5516 existing = self._authorize_existing_service(name, namespace)
5517 self._converge_service_ports(name, namespace, existing)
5518 logger.info("Service %s/%s already exists", namespace, name)
5520 def _ensure_service(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
5521 """Ensure an owned endpoint Service exists on the TLS port.
5523 An absent Service is recreated; one that predates the TLS sidecar is
5524 moved onto it in place.
5525 """
5526 try:
5527 existing = self._authorize_existing_service(name, namespace)
5528 except ApiException as error:
5529 if error.status == 404:
5530 logger.warning("Service %s/%s missing, recreating", namespace, name)
5531 self._create_service(name, namespace, spec)
5532 return
5533 raise
5534 self._converge_service_ports(name, namespace, existing)
5536 def _check_health_watchdog(
5537 self,
5538 name: str,
5539 namespace: str,
5540 ready_replicas: int,
5541 desired_replicas: int,
5542 spec: dict[str, Any],
5543 endpoint: dict[str, Any],
5544 ) -> bool:
5545 """Track prolonged unavailability without changing shared routing.
5547 ``gco-system/gco-gateway`` owns the shared ``/inference`` HTTPRoute to
5548 ``gco-system/inference-proxy``. Individual models are reached through
5549 internal ClusterIP Services, so their readiness never changes the shared
5550 Gateway or HTTPRoute. The threshold still drives degraded-state logging.
5551 """
5552 del namespace, spec, endpoint
5553 if ready_replicas > 0:
5554 if name in self._unready_since:
5555 logger.info("Endpoint %s recovered", name)
5556 del self._unready_since[name]
5557 return False
5559 now = datetime.now(UTC)
5560 if name not in self._unready_since:
5561 self._unready_since[name] = now
5562 logger.warning(
5563 "Endpoint %s has 0/%d ready replicas, starting health watchdog timer",
5564 name,
5565 desired_replicas,
5566 )
5567 return False
5569 unready_duration = (now - self._unready_since[name]).total_seconds()
5570 threshold_exceeded = unready_duration >= self._unhealthy_threshold_seconds
5571 if threshold_exceeded:
5572 logger.warning(
5573 "WATCHDOG: Endpoint %s has been unavailable for %ds (threshold %ds); "
5574 "the authenticated proxy will return 503 until it recovers",
5575 name,
5576 int(unready_duration),
5577 self._unhealthy_threshold_seconds,
5578 )
5579 return threshold_exceeded
5581 def _scale_deployment(self, name: str, namespace: str, replicas: int) -> None:
5582 """Scale only the exact authorized Deployment resourceVersion."""
5583 deployment = self._get_deployment(name, namespace)
5584 if deployment is None:
5585 raise ReconcileFencedError(f"deployment {name} disappeared before scaling")
5586 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
5587 self._assert_mutation_authority()
5588 body: dict[str, Any] = {"spec": {"replicas": replicas}}
5589 if resource_version:
5590 body["metadata"] = {"resourceVersion": resource_version}
5591 self.apps_v1.patch_namespaced_deployment(
5592 name,
5593 namespace,
5594 body=body,
5595 _request_timeout=self._k8s_timeout,
5596 )
5598 def _update_deployment_image(self, name: str, namespace: str, image: str) -> None:
5599 """Update only the exact authorized Deployment resourceVersion."""
5600 deployment = self._get_deployment(name, namespace)
5601 if deployment is None:
5602 raise ReconcileFencedError(f"deployment {name} disappeared before image update")
5603 _metadata, _annotations, _uid, resource_version = self._object_metadata(deployment)
5604 self._assert_mutation_authority()
5605 body: dict[str, Any] = {
5606 "spec": {"template": {"spec": {"containers": [{"name": "inference", "image": image}]}}}
5607 }
5608 if resource_version:
5609 body["metadata"] = {"resourceVersion": resource_version}
5610 self.apps_v1.patch_namespaced_deployment(
5611 name,
5612 namespace,
5613 body=body,
5614 _request_timeout=self._k8s_timeout,
5615 )
5617 def _reconcile_canary(
5618 self,
5619 name: str,
5620 namespace: str,
5621 spec: dict[str, Any],
5622 canary: dict[str, Any],
5623 endpoint: dict[str, Any],
5624 ) -> dict[str, Any]:
5625 """Reconcile a classic canary and return observed readiness for routing."""
5626 canary_image_value = canary.get("image")
5627 if not isinstance(canary_image_value, str) or not canary_image_value.strip():
5628 raise ValueError("canary.image must be a non-empty string")
5629 canary_image = canary_image_value.strip()
5631 canary_replicas = canary.get("replicas", 1)
5632 if (
5633 not isinstance(canary_replicas, int)
5634 or isinstance(canary_replicas, bool)
5635 or canary_replicas < 1
5636 ):
5637 raise ValueError("canary.replicas must be a positive integer")
5639 canary_weight = canary.get("weight", 10)
5640 if (
5641 not isinstance(canary_weight, int)
5642 or isinstance(canary_weight, bool)
5643 or not 1 <= canary_weight <= 99
5644 ):
5645 raise ValueError("canary.weight must be an integer between 1 and 99")
5647 canary_name = f"{name}-canary"
5648 del endpoint # Per-endpoint routing metadata is legacy and intentionally ignored.
5650 canary_spec = dict(spec)
5651 canary_spec["image"] = canary_image
5652 canary_spec["replicas"] = canary_replicas
5653 canary_spec.pop("canary", None)
5654 # A canary image is explicit and global. Retaining the primary's
5655 # region_image_uris would silently deploy the old regional image.
5656 canary_spec.pop("region_image_uris", None)
5658 deployment = self._get_deployment(canary_name, namespace)
5659 state = "creating"
5660 ready_replicas = 0
5661 # The canary runs its endpoint's TLS sidecar program; it has no
5662 # ConfigMap of its own to create or clean up.
5663 tls_proxy_config_map = _tls_proxy_config_map_name(name)
5664 if deployment is None:
5665 logger.info("Creating canary deployment %s with image %s", canary_name, canary_image)
5666 self._create_deployment(
5667 canary_name,
5668 namespace,
5669 canary_spec,
5670 tls_proxy_config_map=tls_proxy_config_map,
5671 )
5672 self._create_service(canary_name, namespace, canary_spec)
5673 else:
5674 self._ensure_service(canary_name, namespace, canary_spec)
5675 self._converge_endpoint_tls_proxy(
5676 canary_name,
5677 namespace,
5678 deployment,
5679 config_map_name=tls_proxy_config_map,
5680 fallback_port=canary_spec.get("port", 8000),
5681 )
5682 current_image = self._get_deployment_image(deployment)
5683 current_replicas = deployment.spec.replicas or 1
5684 ready_replicas = deployment.status.ready_replicas or 0
5685 if current_image != canary_image:
5686 self._update_deployment_image(canary_name, namespace, canary_image)
5687 ready_replicas = 0
5688 state = "updating"
5689 elif current_replicas != canary_replicas:
5690 self._scale_deployment(canary_name, namespace, canary_replicas)
5691 ready_replicas = min(ready_replicas, canary_replicas)
5692 state = "updating"
5693 elif ready_replicas >= canary_replicas:
5694 state = "running"
5696 # Canary selection happens behind ``gco-system/inference-proxy``; the
5697 # shared ``gco-system/gco-gateway`` HTTPRoute is never changed per
5698 # endpoint. The shared inference proxy consumes the observed canary
5699 # status returned here.
5700 return {
5701 "state": state,
5702 "image": canary_image,
5703 "weight": canary_weight,
5704 "replicas_ready": ready_replicas,
5705 "replicas_desired": canary_replicas,
5706 }
5708 def _cleanup_canary(self, name: str, namespace: str) -> None:
5709 """Remove only canary objects carrying this reconciliation authority."""
5710 canary_name = f"{name}-canary"
5711 deployment = self._get_deployment(canary_name, namespace)
5712 if deployment is not None:
5713 self._assert_mutation_authority()
5714 try:
5715 self.apps_v1.delete_namespaced_deployment(
5716 canary_name,
5717 namespace,
5718 body=self._delete_options_for(
5719 deployment, kind="deployment", resource_name=canary_name
5720 ),
5721 _request_timeout=self._k8s_timeout,
5722 )
5723 logger.info("Deleted canary deployment %s", canary_name)
5724 except ApiException as error:
5725 if error.status != 404:
5726 logger.error("Failed to delete canary deployment %s: %s", canary_name, error)
5727 try:
5728 service = self._authorize_existing_service(canary_name, namespace)
5729 except ApiException as error:
5730 if error.status != 404:
5731 logger.error("Failed to read canary service %s: %s", canary_name, error)
5732 else:
5733 self._assert_mutation_authority()
5734 try:
5735 self.core_v1.delete_namespaced_service(
5736 canary_name,
5737 namespace,
5738 body=self._delete_options_for(
5739 service, kind="service", resource_name=canary_name
5740 ),
5741 _request_timeout=self._k8s_timeout,
5742 )
5743 logger.info("Deleted canary service %s", canary_name)
5744 except ApiException as error:
5745 if error.status != 404:
5746 logger.error("Failed to delete canary service %s: %s", canary_name, error)
5748 @staticmethod
5749 def _endpoint_resource_inventory(name: str) -> EndpointResourceInventory:
5750 """Return every deterministic endpoint-owned resource name.
5752 The shared ``mooncake-master`` Service/StatefulSet and any Secret named
5753 by the user are deliberately excluded.
5754 """
5755 return EndpointResourceInventory(
5756 deployments=(
5757 name,
5758 f"{name}-canary",
5759 f"{name}-prefill",
5760 f"{name}-decode",
5761 f"{name}-proxy",
5762 ),
5763 services=(
5764 name,
5765 f"{name}-canary",
5766 f"{name}-prefill",
5767 f"{name}-decode",
5768 f"{name}-proxy",
5769 ),
5770 horizontal_pod_autoscalers=(
5771 name,
5772 f"{name}-prefill",
5773 f"{name}-decode",
5774 f"keda-hpa-{name}",
5775 f"keda-hpa-{name}-prefill",
5776 f"keda-hpa-{name}-decode",
5777 ),
5778 scaled_objects=(
5779 name,
5780 f"{name}-prefill",
5781 f"{name}-decode",
5782 ),
5783 config_maps=(
5784 f"{name}-mooncake",
5785 f"{name}-pd-proxy",
5786 _tls_proxy_config_map_name(name),
5787 ),
5788 legacy_ingresses=(name, f"{name}-canary", f"{name}-proxy"),
5789 legacy_http_routes=(name, f"{name}-canary", f"{name}-proxy"),
5790 generated_admin_secret=f"{name}-admin",
5791 )
5793 @staticmethod
5794 def _cleanup_error(
5795 operation: str,
5796 kind: str,
5797 resource_name: str,
5798 error: Exception,
5799 ) -> str:
5800 """Build a bounded error containing only an endpoint-owned name."""
5801 status_value = getattr(error, "status", None)
5802 status = f" (status {status_value})" if status_value is not None else ""
5803 return f"{operation} {kind} {resource_name} failed{status}"
5805 def _delete_and_confirm(
5806 self,
5807 *,
5808 kind: str,
5809 resource_name: str,
5810 delete_call: Callable[..., Any],
5811 read_call: Callable[[], Any],
5812 patch_metadata: Callable[..., Any] | None,
5813 pending: list[str],
5814 errors: list[str],
5815 observed_resource: Any | None = None,
5816 ) -> bool:
5817 """Read-authorize-delete one exact UID/resourceVersion, then re-observe."""
5818 resource_id = f"{kind}/{resource_name}"
5819 if observed_resource is not None:
5820 observed = observed_resource
5821 else:
5822 try:
5823 observed = read_call()
5824 except ApiException as error:
5825 if error.status == 404:
5826 return False
5827 errors.append(self._cleanup_error("read", kind, resource_name, error))
5828 return False
5829 except Exception as error:
5830 errors.append(self._cleanup_error("read", kind, resource_name, error))
5831 return False
5833 observed = self._authorize_resource(
5834 observed,
5835 kind=kind,
5836 resource_name=resource_name,
5837 patch_metadata=patch_metadata,
5838 read_resource=read_call,
5839 delete_resource=delete_call,
5840 allow_region_mismatch=bool(
5841 self._active_authority and self._active_authority.region_removed
5842 ),
5843 )
5844 _metadata, _annotations, observed_uid, _resource_version = self._object_metadata(observed)
5845 self._assert_mutation_authority()
5846 try:
5847 delete_call(
5848 body=self._delete_options_for(observed, kind=kind, resource_name=resource_name)
5849 )
5850 except ApiException as error:
5851 if error.status != 404:
5852 errors.append(self._cleanup_error("delete", kind, resource_name, error))
5853 except Exception as error:
5854 errors.append(self._cleanup_error("delete", kind, resource_name, error))
5856 try:
5857 remaining = read_call()
5858 except ApiException as error:
5859 if error.status == 404:
5860 return True
5861 errors.append(self._cleanup_error("read", kind, resource_name, error))
5862 return True
5863 except Exception as error:
5864 errors.append(self._cleanup_error("read", kind, resource_name, error))
5865 return True
5867 _metadata, _annotations, remaining_uid, _remaining_version = self._object_metadata(
5868 remaining
5869 )
5870 # A replacement with a different UID is never deleted by this stale
5871 # observation. Its continued presence intentionally blocks cleanup.
5872 if observed_uid and remaining_uid and remaining_uid != observed_uid:
5873 pending.append(f"{resource_id}:replacement")
5874 else:
5875 pending.append(resource_id)
5876 return True
5878 @staticmethod
5879 def _merge_cleanup_results(*results: ResourceCleanupResult) -> ResourceCleanupResult:
5880 """Combine independently attempted cleanup groups without losing failures."""
5881 return ResourceCleanupResult(
5882 pending=tuple(item for result in results for item in result.pending),
5883 errors=tuple(item for result in results for item in result.errors),
5884 resources_found=any(result.resources_found for result in results),
5885 )
5887 def _delete_scaled_objects(
5888 self,
5889 names: tuple[str, ...],
5890 namespace: str,
5891 ) -> ResourceCleanupResult:
5892 """Delete and verify the requested KEDA ScaledObjects."""
5893 pending: list[str] = []
5894 errors: list[str] = []
5895 resources_found = False
5896 custom_objects = client.CustomObjectsApi()
5897 for autoscaler_name in names:
5898 resources_found |= self._delete_and_confirm(
5899 kind="scaledobject",
5900 resource_name=autoscaler_name,
5901 delete_call=partial(
5902 custom_objects.delete_namespaced_custom_object,
5903 group=KEDA_API_GROUP,
5904 version=KEDA_API_VERSION,
5905 namespace=namespace,
5906 plural=KEDA_SCALEDOBJECT_PLURAL,
5907 name=autoscaler_name,
5908 _request_timeout=self._k8s_timeout,
5909 ),
5910 read_call=partial(
5911 custom_objects.get_namespaced_custom_object,
5912 group=KEDA_API_GROUP,
5913 version=KEDA_API_VERSION,
5914 namespace=namespace,
5915 plural=KEDA_SCALEDOBJECT_PLURAL,
5916 name=autoscaler_name,
5917 _request_timeout=self._k8s_timeout,
5918 ),
5919 patch_metadata=partial(
5920 custom_objects.patch_namespaced_custom_object,
5921 group=KEDA_API_GROUP,
5922 version=KEDA_API_VERSION,
5923 namespace=namespace,
5924 plural=KEDA_SCALEDOBJECT_PLURAL,
5925 name=autoscaler_name,
5926 _request_timeout=self._k8s_timeout,
5927 ),
5928 pending=pending,
5929 errors=errors,
5930 )
5931 return ResourceCleanupResult(
5932 pending=tuple(pending),
5933 errors=tuple(errors),
5934 resources_found=resources_found,
5935 )
5937 def _delete_hpas(
5938 self,
5939 names: tuple[str, ...],
5940 namespace: str,
5941 ) -> ResourceCleanupResult:
5942 """Delete and verify the requested native or KEDA-generated HPAs."""
5943 pending: list[str] = []
5944 errors: list[str] = []
5945 resources_found = False
5946 autoscaling_v2 = client.AutoscalingV2Api()
5947 for autoscaler_name in names:
5948 resources_found |= self._delete_and_confirm(
5949 kind="hpa",
5950 resource_name=autoscaler_name,
5951 delete_call=partial(
5952 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
5953 autoscaler_name,
5954 namespace,
5955 _request_timeout=self._k8s_timeout,
5956 ),
5957 read_call=partial(
5958 autoscaling_v2.read_namespaced_horizontal_pod_autoscaler,
5959 autoscaler_name,
5960 namespace,
5961 _request_timeout=self._k8s_timeout,
5962 ),
5963 patch_metadata=partial(
5964 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
5965 autoscaler_name,
5966 namespace,
5967 _request_timeout=self._k8s_timeout,
5968 ),
5969 pending=pending,
5970 errors=errors,
5971 )
5972 return ResourceCleanupResult(
5973 pending=tuple(pending),
5974 errors=tuple(errors),
5975 resources_found=resources_found,
5976 )
5978 def _delete_autoscalers(
5979 self,
5980 scaled_object_names: tuple[str, ...],
5981 hpa_names: tuple[str, ...],
5982 namespace: str,
5983 ) -> ResourceCleanupResult:
5984 """Delete every autoscaler owner and wait for all of them to disappear."""
5985 # KEDA must stop first; only then is it safe to remove its generated HPA
5986 # alongside any native HPA that may remain from a prior configuration.
5987 scaled_objects = self._delete_scaled_objects(scaled_object_names, namespace)
5988 hpas = self._delete_hpas(hpa_names, namespace)
5989 return self._merge_cleanup_results(scaled_objects, hpas)
5991 @staticmethod
5992 def _generated_admin_secret_labels(expected_name: str) -> dict[str, str]:
5993 """Return the schema-valid static provenance used by generated Secrets."""
5994 return {
5995 "app": expected_name,
5996 "project": "gco",
5997 "gco.io/type": "inference",
5998 }
6000 @classmethod
6001 def _is_monitor_owned_admin_secret(
6002 cls,
6003 secret: Any,
6004 expected_name: str,
6005 expected_lifecycle_id: str,
6006 ) -> bool:
6007 """Return whether a generated Secret is bound to this endpoint lifecycle."""
6008 metadata = getattr(secret, "metadata", None)
6009 labels = getattr(metadata, "labels", None)
6010 annotations = getattr(metadata, "annotations", None)
6011 return (
6012 labels == cls._generated_admin_secret_labels(expected_name)
6013 and isinstance(annotations, dict)
6014 and annotations.get("gco.io/lifecycle-id") == expected_lifecycle_id
6015 )
6017 @classmethod
6018 def _is_legacy_monitor_admin_secret(cls, secret: Any, expected_name: str) -> bool:
6019 """Recognize only the exact pre-v7 monitor-generated Secret shape."""
6020 metadata = getattr(secret, "metadata", None)
6021 labels = getattr(metadata, "labels", None)
6022 annotations = getattr(metadata, "annotations", None)
6023 lifecycle_annotation = (
6024 annotations.get("gco.io/lifecycle-id") if isinstance(annotations, dict) else None
6025 )
6026 secret_type = getattr(secret, "type", None)
6027 return (
6028 labels == cls._generated_admin_secret_labels(expected_name)
6029 and lifecycle_annotation is None
6030 and secret_type == "Opaque" # nosec B105 # Kubernetes Secret type, not a credential
6031 and cls._secret_has_admin_api_key(secret)
6032 )
6034 def _adopt_legacy_admin_secret(
6035 self,
6036 secret: Any,
6037 secret_name: str,
6038 namespace: str,
6039 lifecycle_id: str,
6040 ) -> Any:
6041 """Patch exact legacy provenance and verify the persisted lifecycle annotation."""
6042 if not self._is_legacy_monitor_admin_secret(secret, secret_name):
6043 raise AdminApiKeySecretError(
6044 secret_name,
6045 "the conventional generated Secret has ambiguous ownership",
6046 )
6047 metadata = getattr(secret, "metadata", None)
6048 resource_version = getattr(metadata, "resource_version", None)
6049 if not isinstance(resource_version, str) or not resource_version:
6050 raise AdminApiKeySecretError(
6051 secret_name,
6052 "legacy generated Secret has no resource version for safe migration",
6053 )
6054 annotations = self._provenance_annotations() or {_LIFECYCLE_ANNOTATION: lifecycle_id}
6055 self._assert_mutation_authority()
6056 try:
6057 self.core_v1.patch_namespaced_secret(
6058 secret_name,
6059 namespace,
6060 body={
6061 "metadata": {
6062 "resourceVersion": resource_version,
6063 "annotations": annotations,
6064 }
6065 },
6066 _request_timeout=self._k8s_timeout,
6067 )
6068 except Exception as error:
6069 raise AdminApiKeySecretError(
6070 secret_name,
6071 "legacy generated Secret changed during lifecycle migration",
6072 ) from error
6073 adopted = self.core_v1.read_namespaced_secret(
6074 secret_name,
6075 namespace,
6076 _request_timeout=self._k8s_timeout,
6077 )
6078 if not self._is_monitor_owned_admin_secret(
6079 adopted,
6080 secret_name,
6081 lifecycle_id,
6082 ) or not self._secret_has_admin_api_key(adopted):
6083 raise AdminApiKeySecretError(
6084 secret_name,
6085 "legacy generated Secret migration could not be verified",
6086 )
6087 return adopted
6089 def _require_owned_admin_secret(
6090 self,
6091 secret: Any,
6092 secret_name: str,
6093 namespace: str,
6094 lifecycle_id: str,
6095 ) -> Any:
6096 """Accept current provenance or safely migrate one exact legacy shape."""
6097 if self._is_monitor_owned_admin_secret(
6098 secret,
6099 secret_name,
6100 lifecycle_id,
6101 ) and self._secret_has_admin_api_key(secret):
6102 return secret
6103 if self._is_legacy_monitor_admin_secret(secret, secret_name):
6104 return self._adopt_legacy_admin_secret(
6105 secret,
6106 secret_name,
6107 namespace,
6108 lifecycle_id,
6109 )
6110 raise AdminApiKeySecretError(
6111 secret_name,
6112 "the conventional generated Secret exists without matching lifecycle provenance",
6113 )
6115 @staticmethod
6116 def _generated_child_matches(
6117 item: Any,
6118 kind: str,
6119 deployment_names: tuple[str, ...],
6120 service_names: tuple[str, ...],
6121 replica_set_names: tuple[str, ...] = (),
6122 ) -> bool:
6123 """Match generated children only through exact parent identity."""
6124 metadata = getattr(item, "metadata", None)
6125 child_name = getattr(metadata, "name", None)
6126 labels = getattr(metadata, "labels", None)
6127 labels = labels if isinstance(labels, dict) else {}
6128 owner_references = getattr(metadata, "owner_references", None)
6129 owners = owner_references if isinstance(owner_references, (list, tuple)) else ()
6131 if kind in {"replicaset", "pod"}:
6132 app_name = labels.get("app")
6133 if app_name not in deployment_names:
6134 return False
6135 if labels.get("project") != "gco" or labels.get("gco.io/type") != "inference":
6136 return False
6137 if not owners:
6138 return True
6139 for owner in owners:
6140 owner_kind = getattr(owner, "kind", None)
6141 owner_name = getattr(owner, "name", None)
6142 if owner_kind == "Deployment" and owner_name in deployment_names:
6143 return True
6144 if kind == "pod" and owner_kind == "ReplicaSet" and owner_name in replica_set_names:
6145 return True
6146 return False
6148 if kind == "endpoints":
6149 return isinstance(child_name, str) and child_name in service_names
6150 if kind == "endpointslice":
6151 service_name = labels.get("kubernetes.io/service-name")
6152 return isinstance(service_name, str) and service_name in service_names
6153 return False
6155 def _observe_generated_children(
6156 self,
6157 name: str,
6158 namespace: str,
6159 inventory: EndpointResourceInventory,
6160 pending: list[str],
6161 errors: list[str],
6162 ) -> bool:
6163 """Inventory exact endpoint children once per kind and dependency order."""
6164 resources_found = False
6165 matched_replica_sets: list[str] = []
6166 list_calls: tuple[tuple[str, Callable[[], Any]], ...] = (
6167 (
6168 "replicaset",
6169 partial(
6170 self.apps_v1.list_namespaced_replica_set,
6171 namespace,
6172 _request_timeout=self._k8s_timeout,
6173 ),
6174 ),
6175 (
6176 "pod",
6177 partial(
6178 self.core_v1.list_namespaced_pod,
6179 namespace,
6180 _request_timeout=self._k8s_timeout,
6181 ),
6182 ),
6183 (
6184 "endpoints",
6185 partial(
6186 self.core_v1.list_namespaced_endpoints,
6187 namespace,
6188 _request_timeout=self._k8s_timeout,
6189 ),
6190 ),
6191 (
6192 "endpointslice",
6193 partial(
6194 self.discovery_v1.list_namespaced_endpoint_slice,
6195 namespace,
6196 _request_timeout=self._k8s_timeout,
6197 ),
6198 ),
6199 )
6200 for kind, list_call in list_calls:
6201 try:
6202 response = list_call()
6203 except ApiException as error:
6204 errors.append(self._cleanup_error("list", kind, name, error))
6205 continue
6206 except Exception as error:
6207 errors.append(self._cleanup_error("list", kind, name, error))
6208 continue
6209 items = getattr(response, "items", None)
6210 if not isinstance(items, (list, tuple)):
6211 items = ()
6212 for item in items:
6213 if not self._generated_child_matches(
6214 item,
6215 kind,
6216 inventory.deployments,
6217 inventory.services,
6218 tuple(matched_replica_sets),
6219 ):
6220 continue
6221 metadata = getattr(item, "metadata", None)
6222 child_name = getattr(metadata, "name", "unknown")
6223 if kind == "replicaset" and isinstance(child_name, str):
6224 matched_replica_sets.append(child_name)
6225 pending.append(f"{kind}/{child_name}")
6226 resources_found = True
6227 return resources_found
6229 def _delete_resources(
6230 self,
6231 name: str,
6232 namespace: str,
6233 spec: dict[str, Any] | None = None,
6234 *,
6235 expected_lifecycle_id: str | None = None,
6236 ) -> ResourceCleanupResult:
6237 """Delete parents and prove all top-level/generated children absent."""
6238 inventory = self._endpoint_resource_inventory(name)
6239 autoscaler_cleanup = self._delete_autoscalers(
6240 inventory.scaled_objects,
6241 inventory.horizontal_pod_autoscalers,
6242 namespace,
6243 )
6244 pending = list(autoscaler_cleanup.pending)
6245 errors = list(autoscaler_cleanup.errors)
6246 resources_found = autoscaler_cleanup.resources_found
6248 for deployment_name in inventory.deployments:
6249 resources_found |= self._delete_and_confirm(
6250 kind="deployment",
6251 resource_name=deployment_name,
6252 delete_call=partial(
6253 self.apps_v1.delete_namespaced_deployment,
6254 deployment_name,
6255 namespace,
6256 _request_timeout=self._k8s_timeout,
6257 ),
6258 read_call=partial(
6259 self.apps_v1.read_namespaced_deployment,
6260 deployment_name,
6261 namespace,
6262 _request_timeout=self._k8s_timeout,
6263 ),
6264 patch_metadata=partial(
6265 self.apps_v1.patch_namespaced_deployment,
6266 deployment_name,
6267 namespace,
6268 _request_timeout=self._k8s_timeout,
6269 ),
6270 pending=pending,
6271 errors=errors,
6272 )
6274 for service_name in inventory.services:
6275 resources_found |= self._delete_and_confirm(
6276 kind="service",
6277 resource_name=service_name,
6278 delete_call=partial(
6279 self.core_v1.delete_namespaced_service,
6280 service_name,
6281 namespace,
6282 _request_timeout=self._k8s_timeout,
6283 ),
6284 read_call=partial(
6285 self.core_v1.read_namespaced_service,
6286 service_name,
6287 namespace,
6288 _request_timeout=self._k8s_timeout,
6289 ),
6290 patch_metadata=partial(
6291 self.core_v1.patch_namespaced_service,
6292 service_name,
6293 namespace,
6294 _request_timeout=self._k8s_timeout,
6295 ),
6296 pending=pending,
6297 errors=errors,
6298 )
6300 for config_map_name in inventory.config_maps:
6301 resources_found |= self._delete_and_confirm(
6302 kind="configmap",
6303 resource_name=config_map_name,
6304 delete_call=partial(
6305 self.core_v1.delete_namespaced_config_map,
6306 config_map_name,
6307 namespace,
6308 _request_timeout=self._k8s_timeout,
6309 ),
6310 read_call=partial(
6311 self.core_v1.read_namespaced_config_map,
6312 config_map_name,
6313 namespace,
6314 _request_timeout=self._k8s_timeout,
6315 ),
6316 patch_metadata=partial(
6317 self.core_v1.patch_namespaced_config_map,
6318 config_map_name,
6319 namespace,
6320 _request_timeout=self._k8s_timeout,
6321 ),
6322 pending=pending,
6323 errors=errors,
6324 )
6326 for ingress_name in inventory.legacy_ingresses:
6327 resources_found |= self._delete_and_confirm(
6328 kind="ingress",
6329 resource_name=ingress_name,
6330 delete_call=partial(
6331 self.networking_v1.delete_namespaced_ingress,
6332 ingress_name,
6333 namespace,
6334 _request_timeout=self._k8s_timeout,
6335 ),
6336 read_call=partial(
6337 self.networking_v1.read_namespaced_ingress,
6338 ingress_name,
6339 namespace,
6340 _request_timeout=self._k8s_timeout,
6341 ),
6342 patch_metadata=partial(
6343 self.networking_v1.patch_namespaced_ingress,
6344 ingress_name,
6345 namespace,
6346 _request_timeout=self._k8s_timeout,
6347 ),
6348 pending=pending,
6349 errors=errors,
6350 )
6352 custom_objects = client.CustomObjectsApi()
6353 for route_name in inventory.legacy_http_routes:
6354 resources_found |= self._delete_and_confirm(
6355 kind="httproute",
6356 resource_name=route_name,
6357 delete_call=partial(
6358 custom_objects.delete_namespaced_custom_object,
6359 group="gateway.networking.k8s.io",
6360 version="v1",
6361 namespace=namespace,
6362 plural="httproutes",
6363 name=route_name,
6364 _request_timeout=self._k8s_timeout,
6365 ),
6366 read_call=partial(
6367 custom_objects.get_namespaced_custom_object,
6368 group="gateway.networking.k8s.io",
6369 version="v1",
6370 namespace=namespace,
6371 plural="httproutes",
6372 name=route_name,
6373 _request_timeout=self._k8s_timeout,
6374 ),
6375 patch_metadata=partial(
6376 custom_objects.patch_namespaced_custom_object,
6377 group="gateway.networking.k8s.io",
6378 version="v1",
6379 namespace=namespace,
6380 plural="httproutes",
6381 name=route_name,
6382 _request_timeout=self._k8s_timeout,
6383 ),
6384 pending=pending,
6385 errors=errors,
6386 )
6388 resources_found |= self._observe_generated_children(
6389 name,
6390 namespace,
6391 inventory,
6392 pending,
6393 errors,
6394 )
6396 # A user-named Secret is external and survives. The conventional name
6397 # is auto-managed only when both labels and immutable provenance match;
6398 # an ambiguous same-name Secret blocks terminal cleanup.
6399 mooncake = spec.get("mooncake") if isinstance(spec, dict) else None
6400 proxy = mooncake.get("proxy") if isinstance(mooncake, dict) else None
6401 named_secret = proxy.get("admin_api_key_secret") if isinstance(proxy, dict) else None
6402 generated_secret = inventory.generated_admin_secret
6403 if named_secret != generated_secret:
6404 try:
6405 secret = self.core_v1.read_namespaced_secret(
6406 generated_secret,
6407 namespace,
6408 _request_timeout=self._k8s_timeout,
6409 )
6410 except ApiException as error:
6411 if error.status != 404:
6412 errors.append(self._cleanup_error("read", "secret", generated_secret, error))
6413 except Exception as error:
6414 errors.append(self._cleanup_error("read", "secret", generated_secret, error))
6415 else:
6416 resources_found = True
6417 if isinstance(expected_lifecycle_id, str) and expected_lifecycle_id:
6418 try:
6419 owned_secret = self._require_owned_admin_secret(
6420 secret,
6421 generated_secret,
6422 namespace,
6423 expected_lifecycle_id,
6424 )
6425 except AdminApiKeySecretError:
6426 errors.append(
6427 f"ambiguous secret {generated_secret} is not owned by lifecycle "
6428 f"{expected_lifecycle_id}"
6429 )
6430 else:
6431 resources_found |= self._delete_and_confirm(
6432 kind="secret",
6433 resource_name=generated_secret,
6434 delete_call=partial(
6435 self.core_v1.delete_namespaced_secret,
6436 generated_secret,
6437 namespace,
6438 _request_timeout=self._k8s_timeout,
6439 ),
6440 read_call=partial(
6441 self.core_v1.read_namespaced_secret,
6442 generated_secret,
6443 namespace,
6444 _request_timeout=self._k8s_timeout,
6445 ),
6446 patch_metadata=partial(
6447 self.core_v1.patch_namespaced_secret,
6448 generated_secret,
6449 namespace,
6450 _request_timeout=self._k8s_timeout,
6451 ),
6452 pending=pending,
6453 errors=errors,
6454 observed_resource=owned_secret,
6455 )
6456 else:
6457 errors.append(
6458 f"ambiguous secret {generated_secret} is not owned by lifecycle unknown"
6459 )
6461 return ResourceCleanupResult(
6462 pending=tuple(sorted(set(pending))),
6463 errors=tuple(errors),
6464 resources_found=resources_found,
6465 )
6467 def _build_hpa_metrics(self, metrics_config: list[dict[str, Any]]) -> list[Any]:
6468 """Translate a metrics config list into autoscaler metric specs.
6470 Each entry names a resource (``cpu`` or ``memory``) and a target
6471 average utilization. Unrecognized entries are skipped, and when nothing
6472 recognizable remains the autoscaler falls back to scaling on CPU at 70%
6473 so a Deployment is never left without a scaling signal.
6474 """
6475 hpa_metrics = []
6476 for m in metrics_config:
6477 metric_type = m.get("type", "cpu")
6478 target_value = m.get("target", 70)
6480 if metric_type == "cpu":
6481 hpa_metrics.append(
6482 client.V2MetricSpec(
6483 type="Resource",
6484 resource=client.V2ResourceMetricSource(
6485 name="cpu",
6486 target=client.V2MetricTarget(
6487 type="Utilization",
6488 average_utilization=target_value,
6489 ),
6490 ),
6491 )
6492 )
6493 elif metric_type == "memory":
6494 hpa_metrics.append(
6495 client.V2MetricSpec(
6496 type="Resource",
6497 resource=client.V2ResourceMetricSource(
6498 name="memory",
6499 target=client.V2MetricTarget(
6500 type="Utilization",
6501 average_utilization=target_value,
6502 ),
6503 ),
6504 )
6505 )
6507 if not hpa_metrics:
6508 # Default to CPU if no recognized metrics
6509 hpa_metrics.append(
6510 client.V2MetricSpec(
6511 type="Resource",
6512 resource=client.V2ResourceMetricSource(
6513 name="cpu",
6514 target=client.V2MetricTarget(
6515 type="Utilization",
6516 average_utilization=70,
6517 ),
6518 ),
6519 )
6520 )
6522 return hpa_metrics
6524 @staticmethod
6525 def _metrics_require_keda(metrics_config: list[dict[str, Any]]) -> bool:
6526 """Return True when any metric can only be scaled via KEDA/CloudWatch.
6528 GPU metrics are not Kubernetes Resource metrics, so a native HPA cannot
6529 consume them. Their presence forces the whole autoscaler onto the KEDA
6530 ScaledObject path, where cpu/memory targets become native KEDA triggers
6531 alongside the aws-cloudwatch GPU trigger.
6532 """
6533 return any(m.get("type") in _CLOUDWATCH_METRIC_BY_TYPE for m in metrics_config)
6535 def _build_keda_triggers(
6536 self,
6537 metrics_config: list[dict[str, Any]],
6538 target_name: str,
6539 namespace: str,
6540 ) -> list[dict[str, Any]]:
6541 """Translate a metrics config list into KEDA ScaledObject triggers.
6543 ``cpu`` and ``memory`` map to KEDA's native resource triggers (the same
6544 utilization signal a plain HPA would use). ``gpu``/``gpu_memory`` map to
6545 an ``aws-cloudwatch`` trigger reading the matching ContainerInsights
6546 metric for this Deployment, identified by the
6547 ClusterName/Namespace/PodName dimension triple. Unrecognized entries are
6548 skipped; when nothing recognizable remains the autoscaler falls back to
6549 CPU at 70% so a Deployment is never left without a scaling signal.
6550 """
6551 triggers: list[dict[str, Any]] = []
6552 for m in metrics_config:
6553 metric_type = m.get("type", "cpu")
6554 target_value = m.get("target", 70)
6556 if metric_type in ("cpu", "memory"):
6557 triggers.append(
6558 {
6559 "type": metric_type,
6560 "metricType": "Utilization",
6561 "metadata": {"value": str(target_value)},
6562 }
6563 )
6564 elif metric_type in _CLOUDWATCH_METRIC_BY_TYPE:
6565 triggers.append(
6566 {
6567 "type": "aws-cloudwatch",
6568 "metadata": {
6569 "namespace": GPU_METRIC_NAMESPACE,
6570 "metricName": _CLOUDWATCH_METRIC_BY_TYPE[metric_type],
6571 "dimensionName": "ClusterName;Namespace;PodName",
6572 "dimensionValue": f"{self.cluster_id};{namespace};{target_name}",
6573 "targetMetricValue": str(target_value),
6574 "minMetricValue": "0",
6575 "metricStat": "Average",
6576 "awsRegion": self.region,
6577 "identityOwner": "operator",
6578 },
6579 }
6580 )
6582 if not triggers:
6583 triggers.append(
6584 {
6585 "type": "cpu",
6586 "metricType": "Utilization",
6587 "metadata": {"value": "70"},
6588 }
6589 )
6591 return triggers
6593 def _apply_scaled_object(
6594 self,
6595 name: str,
6596 namespace: str,
6597 target_name: str,
6598 min_replicas: int,
6599 max_replicas: int,
6600 metrics_config: list[dict[str, Any]],
6601 ) -> None:
6602 """Create or patch a KEDA ScaledObject targeting one Deployment.
6604 Used whenever the metric set includes a GPU signal (see
6605 :meth:`_metrics_require_keda`). KEDA owns the backing HPA and reads GPU
6606 utilization from CloudWatch via the keda-operator's IRSA role, scaling
6607 ``target_name`` between ``min_replicas`` and ``max_replicas``. An
6608 already-present ScaledObject of the same name is merge-patched rather
6609 than duplicated.
6610 """
6611 body: dict[str, Any] = {
6612 "apiVersion": f"{KEDA_API_GROUP}/{KEDA_API_VERSION}",
6613 "kind": "ScaledObject",
6614 "metadata": {
6615 "name": name,
6616 "namespace": namespace,
6617 "labels": {
6618 "app": name,
6619 "project": "gco",
6620 "gco.io/type": "inference",
6621 },
6622 "annotations": self._provenance_annotations() or {},
6623 },
6624 "spec": {
6625 "scaleTargetRef": {"name": target_name},
6626 "minReplicaCount": min_replicas,
6627 "maxReplicaCount": max_replicas,
6628 "triggers": self._build_keda_triggers(metrics_config, target_name, namespace),
6629 },
6630 }
6632 custom = client.CustomObjectsApi()
6633 self._assert_mutation_authority()
6634 try:
6635 custom.create_namespaced_custom_object(
6636 group=KEDA_API_GROUP,
6637 version=KEDA_API_VERSION,
6638 namespace=namespace,
6639 plural=KEDA_SCALEDOBJECT_PLURAL,
6640 body=body,
6641 _request_timeout=self._k8s_timeout,
6642 )
6643 self._confirm_created_resource(
6644 kind="scaledobject",
6645 resource_name=name,
6646 read_resource=partial(
6647 custom.get_namespaced_custom_object,
6648 group=KEDA_API_GROUP,
6649 version=KEDA_API_VERSION,
6650 namespace=namespace,
6651 plural=KEDA_SCALEDOBJECT_PLURAL,
6652 name=name,
6653 _request_timeout=self._k8s_timeout,
6654 ),
6655 delete_resource=partial(
6656 custom.delete_namespaced_custom_object,
6657 group=KEDA_API_GROUP,
6658 version=KEDA_API_VERSION,
6659 namespace=namespace,
6660 plural=KEDA_SCALEDOBJECT_PLURAL,
6661 name=name,
6662 _request_timeout=self._k8s_timeout,
6663 ),
6664 )
6665 logger.info(
6666 "Created KEDA ScaledObject %s targeting %s (min=%d, max=%d)",
6667 name,
6668 target_name,
6669 min_replicas,
6670 max_replicas,
6671 )
6672 except ApiException as error:
6673 if error.status != 409:
6674 raise
6675 existing = custom.get_namespaced_custom_object(
6676 group=KEDA_API_GROUP,
6677 version=KEDA_API_VERSION,
6678 namespace=namespace,
6679 plural=KEDA_SCALEDOBJECT_PLURAL,
6680 name=name,
6681 _request_timeout=self._k8s_timeout,
6682 )
6683 existing = self._authorize_resource(
6684 existing,
6685 kind="scaledobject",
6686 resource_name=name,
6687 patch_metadata=partial(
6688 custom.patch_namespaced_custom_object,
6689 group=KEDA_API_GROUP,
6690 version=KEDA_API_VERSION,
6691 namespace=namespace,
6692 plural=KEDA_SCALEDOBJECT_PLURAL,
6693 name=name,
6694 _request_timeout=self._k8s_timeout,
6695 ),
6696 read_resource=lambda: custom.get_namespaced_custom_object(
6697 group=KEDA_API_GROUP,
6698 version=KEDA_API_VERSION,
6699 namespace=namespace,
6700 plural=KEDA_SCALEDOBJECT_PLURAL,
6701 name=name,
6702 _request_timeout=self._k8s_timeout,
6703 ),
6704 delete_resource=partial(
6705 custom.delete_namespaced_custom_object,
6706 group=KEDA_API_GROUP,
6707 version=KEDA_API_VERSION,
6708 namespace=namespace,
6709 plural=KEDA_SCALEDOBJECT_PLURAL,
6710 name=name,
6711 _request_timeout=self._k8s_timeout,
6712 ),
6713 )
6714 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
6715 body["metadata"]["resourceVersion"] = resource_version
6716 self._assert_mutation_authority()
6717 custom.patch_namespaced_custom_object(
6718 group=KEDA_API_GROUP,
6719 version=KEDA_API_VERSION,
6720 namespace=namespace,
6721 plural=KEDA_SCALEDOBJECT_PLURAL,
6722 name=name,
6723 body=body,
6724 _request_timeout=self._k8s_timeout,
6725 )
6726 logger.info("Updated KEDA ScaledObject %s", name)
6728 def _apply_hpa(
6729 self,
6730 hpa_name: str,
6731 namespace: str,
6732 target_name: str,
6733 min_replicas: int,
6734 max_replicas: int,
6735 metrics_config: list[dict[str, Any]],
6736 ) -> None:
6737 """Create or patch a single autoscaler targeting one Deployment.
6739 Builds a V2 autoscaler that scales ``target_name`` between
6740 ``min_replicas`` and ``max_replicas`` on the given metrics, then creates
6741 it. An already-present autoscaler of the same name is patched in place
6742 rather than duplicated. When the metric set includes a GPU signal the
6743 autoscaler is materialized as a KEDA ScaledObject instead (native HPA
6744 Resource metrics cannot read GPU utilization).
6745 """
6746 if self._metrics_require_keda(metrics_config):
6747 self._apply_scaled_object(
6748 name=hpa_name,
6749 namespace=namespace,
6750 target_name=target_name,
6751 min_replicas=min_replicas,
6752 max_replicas=max_replicas,
6753 metrics_config=metrics_config,
6754 )
6755 return
6757 hpa = client.V2HorizontalPodAutoscaler(
6758 metadata=client.V1ObjectMeta(
6759 name=hpa_name,
6760 namespace=namespace,
6761 labels={
6762 "app": hpa_name,
6763 "project": "gco",
6764 "gco.io/type": "inference",
6765 },
6766 annotations=self._provenance_annotations(),
6767 ),
6768 spec=client.V2HorizontalPodAutoscalerSpec(
6769 scale_target_ref=client.V2CrossVersionObjectReference(
6770 api_version="apps/v1",
6771 kind="Deployment",
6772 name=target_name,
6773 ),
6774 min_replicas=min_replicas,
6775 max_replicas=max_replicas,
6776 metrics=self._build_hpa_metrics(metrics_config),
6777 ),
6778 )
6780 autoscaling_v2 = client.AutoscalingV2Api()
6781 self._assert_mutation_authority()
6782 try:
6783 autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(namespace, hpa)
6784 self._confirm_created_resource(
6785 kind="hpa",
6786 resource_name=hpa_name,
6787 read_resource=partial(
6788 autoscaling_v2.read_namespaced_horizontal_pod_autoscaler,
6789 hpa_name,
6790 namespace,
6791 _request_timeout=self._k8s_timeout,
6792 ),
6793 delete_resource=partial(
6794 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
6795 hpa_name,
6796 namespace,
6797 _request_timeout=self._k8s_timeout,
6798 ),
6799 )
6800 logger.info(
6801 "Created HPA %s targeting %s (min=%d, max=%d)",
6802 hpa_name,
6803 target_name,
6804 min_replicas,
6805 max_replicas,
6806 )
6807 except ApiException as error:
6808 if error.status != 409:
6809 raise
6810 existing = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
6811 hpa_name, namespace, _request_timeout=self._k8s_timeout
6812 )
6813 existing = self._authorize_resource(
6814 existing,
6815 kind="hpa",
6816 resource_name=hpa_name,
6817 patch_metadata=partial(
6818 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
6819 hpa_name,
6820 namespace,
6821 _request_timeout=self._k8s_timeout,
6822 ),
6823 read_resource=lambda: autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
6824 hpa_name, namespace, _request_timeout=self._k8s_timeout
6825 ),
6826 delete_resource=partial(
6827 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
6828 hpa_name,
6829 namespace,
6830 _request_timeout=self._k8s_timeout,
6831 ),
6832 )
6833 _metadata, _annotations, _uid, resource_version = self._object_metadata(existing)
6834 hpa.metadata.resource_version = resource_version
6835 self._assert_mutation_authority()
6836 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler(
6837 hpa_name,
6838 namespace,
6839 hpa,
6840 _request_timeout=self._k8s_timeout,
6841 )
6842 logger.info("Updated HPA %s", hpa_name)
6844 def _verify_hpa_owner(
6845 self,
6846 hpa_name: str,
6847 namespace: str,
6848 target_name: str,
6849 ) -> ResourceCleanupResult:
6850 """Verify that the exact native/KEDA HPA owns the expected Deployment."""
6851 autoscaling_v2 = client.AutoscalingV2Api()
6852 try:
6853 hpa = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
6854 hpa_name,
6855 namespace,
6856 )
6857 except ApiException as error:
6858 if error.status == 404:
6859 return ResourceCleanupResult(pending=(f"hpa/{hpa_name}",))
6860 return ResourceCleanupResult(
6861 errors=(self._cleanup_error("read", "hpa", hpa_name, error),)
6862 )
6863 except Exception as error:
6864 return ResourceCleanupResult(
6865 errors=(self._cleanup_error("read", "hpa", hpa_name, error),)
6866 )
6867 try:
6868 hpa = self._authorize_resource(
6869 hpa,
6870 kind="hpa",
6871 resource_name=hpa_name,
6872 patch_metadata=partial(
6873 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler,
6874 hpa_name,
6875 namespace,
6876 _request_timeout=self._k8s_timeout,
6877 ),
6878 read_resource=lambda: autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(
6879 hpa_name, namespace, _request_timeout=self._k8s_timeout
6880 ),
6881 delete_resource=partial(
6882 autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler,
6883 hpa_name,
6884 namespace,
6885 _request_timeout=self._k8s_timeout,
6886 ),
6887 )
6888 except ReconcileFencedError:
6889 raise
6890 hpa_spec = getattr(hpa, "spec", None)
6891 target_ref = getattr(hpa_spec, "scale_target_ref", None)
6892 observed_api_version = getattr(target_ref, "api_version", None)
6893 observed_kind = getattr(target_ref, "kind", None)
6894 observed_target = getattr(target_ref, "name", None)
6895 if (
6896 observed_api_version != "apps/v1"
6897 or observed_kind != "Deployment"
6898 or observed_target != target_name
6899 ):
6900 return ResourceCleanupResult(
6901 pending=(f"hpa/{hpa_name}",),
6902 resources_found=True,
6903 )
6904 return ResourceCleanupResult(resources_found=True)
6906 def _reconcile_classic_autoscaler(
6907 self,
6908 name: str,
6909 namespace: str,
6910 spec: dict[str, Any],
6911 *,
6912 apply_desired: bool = True,
6913 ) -> ResourceCleanupResult:
6914 """Converge a classic endpoint to exactly one autoscaler owner.
6916 A stopped target is first restored to a positive minimum, then the
6917 desired native/KEDA owner is applied and read back before reconciliation
6918 may proceed. Obsolete owners are always confirmed absent first.
6919 """
6920 autoscaling = spec.get("autoscaling", {})
6921 metrics_config = autoscaling.get("metrics", [{"type": "cpu", "target": 70}])
6922 keda = self._metrics_require_keda(metrics_config)
6923 if keda:
6924 cleanup = self._delete_hpas((name,), namespace)
6925 else:
6926 cleanup = self._merge_cleanup_results(
6927 self._delete_scaled_objects((name,), namespace),
6928 self._delete_hpas((f"keda-hpa-{name}",), namespace),
6929 )
6931 if cleanup.complete and apply_desired:
6932 target = self._get_deployment(name, namespace)
6933 current = getattr(getattr(target, "spec", None), "replicas", None)
6934 if target is not None and current == 0:
6935 minimum = max(1, int(autoscaling.get("min_replicas", 1)))
6936 self._scale_deployment(name, namespace, minimum)
6937 self._create_or_update_hpa(name, namespace, spec)
6938 desired_hpa = f"keda-hpa-{name}" if keda else name
6939 return self._merge_cleanup_results(
6940 cleanup,
6941 self._verify_hpa_owner(desired_hpa, namespace, name),
6942 )
6943 return cleanup
6945 def _create_or_update_hpa(self, name: str, namespace: str, spec: dict[str, Any]) -> None:
6946 """Create or update a Horizontal Pod Autoscaler for an inference endpoint."""
6947 autoscaling_config = spec.get("autoscaling", {})
6948 if not autoscaling_config.get("enabled"):
6949 return
6951 min_replicas = autoscaling_config.get("min_replicas", 1)
6952 max_replicas = autoscaling_config.get("max_replicas", 10)
6953 metrics_config = autoscaling_config.get("metrics", [{"type": "cpu", "target": 70}])
6955 self._apply_hpa(
6956 hpa_name=name,
6957 namespace=namespace,
6958 target_name=name,
6959 min_replicas=min_replicas,
6960 max_replicas=max_replicas,
6961 metrics_config=metrics_config,
6962 )
6964 @staticmethod
6965 def _role_autoscaling_config(
6966 spec: dict[str, Any],
6967 role: str,
6968 ) -> dict[str, Any] | None:
6969 """Return an enabled role config, or ``None`` for static ownership."""
6970 mooncake = spec.get("mooncake")
6971 if not isinstance(mooncake, dict):
6972 return None
6973 autoscaling = mooncake.get("autoscaling")
6974 if not isinstance(autoscaling, dict) or autoscaling.get("enabled") is not True:
6975 return None
6976 role_config = autoscaling.get(role)
6977 return role_config if isinstance(role_config, dict) else None
6979 def _reconcile_role_autoscaler(
6980 self,
6981 name: str,
6982 namespace: str,
6983 spec: dict[str, Any],
6984 role: str,
6985 *,
6986 apply_desired: bool = True,
6987 ) -> ResourceCleanupResult:
6988 """Converge one Mooncake role to exactly one or zero replica owners."""
6989 target_name = f"{name}-{role}"
6990 role_config = self._role_autoscaling_config(spec, role)
6991 if role_config is None:
6992 return self._delete_autoscalers(
6993 (target_name,),
6994 (target_name, f"keda-hpa-{target_name}"),
6995 namespace,
6996 )
6998 metrics_config = role_config.get("metrics", [{"type": "cpu", "target": 70}])
6999 if self._metrics_require_keda(metrics_config):
7000 cleanup = self._delete_hpas((target_name,), namespace)
7001 else:
7002 cleanup = self._merge_cleanup_results(
7003 self._delete_scaled_objects((target_name,), namespace),
7004 self._delete_hpas((f"keda-hpa-{target_name}",), namespace),
7005 )
7006 if cleanup.complete and apply_desired:
7007 self._create_role_hpa(name, namespace, spec, role)
7008 return cleanup
7010 def _create_role_hpa(self, name: str, ns: str, spec: dict[str, Any], role: str) -> None:
7011 """Create or update one autoscaler for a single Mooncake role.
7013 When the endpoint's ``mooncake.autoscaling`` block is enabled and
7014 carries a config for this role, this materializes exactly one autoscaler
7015 named ``{name}-{role}`` that scales the matching ``{name}-{role}``
7016 Deployment between the role's ``min_replicas`` and ``max_replicas``. The
7017 role Deployment itself is already materialized at ``min_replicas`` by
7018 :meth:`_replica_count_for_role`, so the autoscaler owns the count from
7019 that lower bound. When autoscaling is absent or disabled, or when the
7020 role carries no config, no autoscaler is created and the role's replicas
7021 stay at their topology value.
7023 Args:
7024 name: The endpoint name.
7025 ns: The namespace the role Deployment lives in.
7026 spec: The endpoint spec; ``spec["mooncake"]["autoscaling"]`` drives
7027 the bounds and metrics.
7028 role: One of ``"prefill"`` or ``"decode"``.
7029 """
7030 role_cfg = self._role_autoscaling_config(spec, role)
7031 if role_cfg is None:
7032 return
7034 min_replicas = role_cfg.get("min_replicas", 1)
7035 max_replicas = role_cfg.get("max_replicas", 10)
7036 metrics_config = role_cfg.get("metrics", [{"type": "cpu", "target": 70}])
7038 target_name = f"{name}-{role}"
7039 self._apply_hpa(
7040 hpa_name=target_name,
7041 namespace=ns,
7042 target_name=target_name,
7043 min_replicas=min_replicas,
7044 max_replicas=max_replicas,
7045 metrics_config=metrics_config,
7046 )
7048 # ------------------------------------------------------------------
7049 # Metrics
7050 # ------------------------------------------------------------------
7052 def get_metrics(self) -> dict[str, Any]:
7053 return {
7054 "cluster_id": self.cluster_id,
7055 "region": self.region,
7056 "running": self._running,
7057 "reconcile_count": self._reconcile_count,
7058 "errors_count": self._errors_count,
7059 "seconds_since_last_pass": time.monotonic() - self._last_loop_completed_at,
7060 }
7063def create_inference_monitor_from_env() -> InferenceMonitor:
7064 """Create an InferenceMonitor from environment variables."""
7065 cluster_id = os.getenv("CLUSTER_NAME", "unknown-cluster")
7066 region = os.getenv("REGION", "unknown-region")
7067 namespace = os.getenv("INFERENCE_NAMESPACE", "gco-inference")
7068 interval = int(os.getenv("RECONCILE_INTERVAL_SECONDS", "15"))
7070 # Enable structured JSON logging for CloudWatch Insights
7071 configure_structured_logging(
7072 service_name="inference-monitor",
7073 cluster_id=cluster_id,
7074 region=region,
7075 )
7077 store = InferenceEndpointStore() # Uses DYNAMODB_REGION env var, falls back to REGION
7079 return InferenceMonitor(
7080 cluster_id=cluster_id,
7081 region=region,
7082 store=store,
7083 namespace=namespace,
7084 reconcile_interval=interval,
7085 )
7088async def main() -> None:
7089 """Entry point for the inference monitor."""
7090 monitor = create_inference_monitor_from_env()
7091 logger.info("Inference monitor initialized: %s", monitor.get_metrics())
7093 # Expose Prometheus metrics on a dedicated port for the in-cluster
7094 # observability scrape. A scrape-time collector reflects the monitor's live
7095 # counters (reconcile_count, errors_count, running), so no push from the
7096 # reconcile loop is needed. The deployment binds it to pod loopback
7097 # (METRICS_HOST=127.0.0.1) behind a TLS sidecar that Prometheus scrapes.
7098 from gco.services.service_metrics import start_metrics_server
7100 metrics_port = int(os.getenv("METRICS_PORT", "9090"))
7101 metrics_host = os.getenv("METRICS_HOST", "0.0.0.0") # container listener
7102 start_metrics_server(metrics_port, "inference-monitor", monitor.get_metrics, host=metrics_host)
7104 # Kubernetes stops pods with SIGTERM. This process is PID 1 in its
7105 # container, and PID 1 receives no kernel-default signal handling — so
7106 # without an explicit handler SIGTERM was silently ignored, every pod
7107 # rotation burned the full terminationGracePeriodSeconds, and the kubelet
7108 # SIGKILLed the monitor mid-reconcile (exit 137). The handler flips the
7109 # same stop flag the reconcile loop already honors, so shutdown waits for
7110 # the in-flight cycle and exits 0 within one reconcile interval.
7111 shutdown_requested = asyncio.Event()
7112 loop = asyncio.get_running_loop()
7114 def _handle_sigterm() -> None:
7115 logger.info("SIGTERM received; stopping inference monitor after current cycle")
7116 shutdown_requested.set()
7117 monitor.stop()
7119 try:
7120 loop.add_signal_handler(signal.SIGTERM, _handle_sigterm)
7121 except NotImplementedError:
7122 # Non-POSIX event loops (Windows dev environments running the unit
7123 # suite) don't support loop signal handlers; in the Linux container
7124 # this always succeeds.
7125 logger.debug("Event loop does not support signal handlers; skipping SIGTERM hook")
7127 try:
7128 while not shutdown_requested.is_set():
7129 try:
7130 await monitor.start()
7131 except KeyboardInterrupt:
7132 logger.info("Shutting down inference monitor")
7133 monitor.stop()
7134 break
7135 except Exception as e:
7136 logger.error("Monitor crashed, restarting in 10s: %s", e, exc_info=True)
7137 monitor.stop()
7138 monitor._running = False
7139 await asyncio.sleep(10)
7140 logger.info("Inference monitor exited cleanly")
7141 finally:
7142 with contextlib.suppress(NotImplementedError, ValueError):
7143 loop.remove_signal_handler(signal.SIGTERM)
7146if __name__ == "__main__":
7147 asyncio.run(main())