Coverage for gco / models / inference_models.py: 100.00%
117 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Data models for inference endpoint management.
4Defines the schema for inference endpoints stored in DynamoDB and
5used by the inference_monitor reconciliation loop.
6"""
8from __future__ import annotations
10from dataclasses import dataclass, field
11from enum import StrEnum
12from typing import Any
15class EndpointState(StrEnum):
16 """Desired state for an inference endpoint."""
18 DEPLOYING = "deploying"
19 RUNNING = "running"
20 STOPPED = "stopped"
21 DELETED = "deleted"
24class RegionSyncState(StrEnum):
25 """Sync state of an endpoint in a specific region."""
27 PENDING = "pending"
28 CREATING = "creating"
29 RUNNING = "running"
30 UPDATING = "updating"
31 STOPPING = "stopping"
32 STOPPED = "stopped"
33 DELETING = "deleting"
34 DELETED = "deleted"
35 ERROR = "error"
38#: Node labels EKS Auto Mode stamps on every accelerated node, naming the GPU
39#: and the instance family it was provisioned from.
40GPU_NAME_NODE_LABEL = "eks.amazonaws.com/instance-gpu-name"
41INSTANCE_FAMILY_NODE_LABEL = "eks.amazonaws.com/instance-family"
42#: The well-known Kubernetes label naming a node's EC2 instance type; the
43#: renderer prefers it for an endpoint's ``gpu_type`` hint.
44INSTANCE_TYPE_NODE_LABEL = "node.kubernetes.io/instance-type"
46#: NVIDIA GPUs older than Ampere, by their ``GPU_NAME_NODE_LABEL`` value.
47#: SGLang's prebuilt ``sgl-kernel`` / FlashInfer binaries target compute
48#: capability 8.0 and newer; on these GPUs the server crash-loops at startup
49#: with "no kernel image is available for execution on the device". The
50#: shipped inference NodePool admits g4dn (T4) as its cheapest family, so a
51#: plain ``gco inference deploy --framework sglang`` landed there and looped
52#: for the whole readiness window (caught by the live release validation,
53#: 2026-09-18). vLLM is unaffected: it ships sm_70+ kernels. The renderer
54#: keeps SGLang pods off these GPUs and the CLI refuses a node selector that
55#: pins one.
56SGLANG_UNSUPPORTED_GPU_NAMES: tuple[str, ...] = ("k80", "m60", "v100", "t4")
57#: The EC2 families that carry those GPUs, for ``INSTANCE_FAMILY_NODE_LABEL``
58#: selectors.
59SGLANG_UNSUPPORTED_GPU_FAMILIES: tuple[str, ...] = ("p2", "g3", "g3s", "p3", "p3dn", "g4dn")
60#: Ampere-or-newer examples for the operator-facing refusal message.
61SGLANG_SUPPORTED_GPU_EXAMPLES = "A10G (g5), L4 (g6/g6f/gr6), L40S (g6e), A100 (p4d), H100 (p5)"
63#: ``timeoutSeconds`` for every probe the renderer puts on an inference
64#: container. Left unset, the kubelet applies 1 s — invisible in the rendered
65#: manifest and the same default that has already crash-looped the platform
66#: workloads on a contended node (see tests/test_workload_probe_timing_contract.py),
67#: which is why every probe in this repository states its timeout. For SGLang
68#: it is not even a margin question: its ``/health`` is a real check, not a
69#: ping. With ``SGLANG_ENABLE_HEALTH_ENDPOINT_GENERATION`` on by default
70#: (sglang 0.5.19, ``http_server.health_generate``) it submits a one-token
71#: generation and then polls ``await asyncio.sleep(1)`` before its first look
72#: at the result, so a healthy server answers in 1.00-1.03 s and never faster.
73#: Under the 1 s default every probe timed out: readiness flapped and liveness
74#: killed a healthy A10G container every ``initialDelay + failureThreshold *
75#: period`` seconds (live release validation, 2026-09-18 — exit 0 after
76#: SIGTERM, ``Liveness probe failed: context deadline exceeded``). Five seconds
77#: clears that floor with room for a busy CPU, stays under the 10 s readiness
78#: period so probes never queue, and still fails a hung engine, whose check
79#: runs to SGLang's own 20 s ``HEALTH_CHECK_TIMEOUT`` and returns 503. vLLM's
80#: ``/health`` answers immediately; it gets the same explicit value so the
81#: kubelet default can never quietly return.
82INFERENCE_PROBE_TIMEOUT_SECONDS = 5
85@dataclass
86class InferenceEndpointSpec:
87 """Specification for an inference endpoint deployment."""
89 image: str
90 port: int = 8000
91 replicas: int = 1
92 gpu_count: int = 1
93 gpu_type: str | None = None # e.g. "g5.xlarge" — used for nodeSelector
94 model_path: str | None = None # EFS/FSx path for model weights
95 model_source: str | None = None # S3 URI (s3://bucket/path) or HuggingFace repo ID
96 health_check_path: str = "/health"
97 env: dict[str, str] = field(default_factory=dict)
98 resources: dict[str, Any] = field(default_factory=dict)
99 command: list[str] | None = None
100 args: list[str] | None = None
101 tolerations: list[dict[str, Any]] | None = None
102 node_selector: dict[str, str] | None = None
103 autoscaling: dict[str, Any] | None = (
104 None # {enabled, min_replicas, max_replicas, metrics: [{type, target}]}
105 )
106 # Canary deployment fields
107 canary: dict[str, Any] | None = None # {image, weight, replicas}
108 # Capacity type: "on-demand", "spot", or "mixed" (base on-demand, overflow spot)
109 capacity_type: str | None = None
111 def to_dict(self) -> dict[str, Any]:
112 result: dict[str, Any] = {
113 "image": self.image,
114 "port": self.port,
115 "replicas": self.replicas,
116 "gpu_count": self.gpu_count,
117 "health_check_path": self.health_check_path,
118 }
119 if self.gpu_type:
120 result["gpu_type"] = self.gpu_type
121 if self.model_path:
122 result["model_path"] = self.model_path
123 if self.model_source:
124 result["model_source"] = self.model_source
125 if self.env:
126 result["env"] = self.env
127 if self.resources:
128 result["resources"] = self.resources
129 if self.command:
130 result["command"] = self.command
131 if self.args:
132 result["args"] = self.args
133 if self.tolerations:
134 result["tolerations"] = self.tolerations
135 if self.node_selector:
136 result["node_selector"] = self.node_selector
137 if self.autoscaling:
138 result["autoscaling"] = self.autoscaling
139 if self.canary:
140 result["canary"] = self.canary
141 if self.capacity_type:
142 result["capacity_type"] = self.capacity_type
143 return result
145 @classmethod
146 def from_dict(cls, data: dict[str, Any]) -> InferenceEndpointSpec:
147 return cls(
148 image=data["image"],
149 port=data.get("port", 8000),
150 replicas=data.get("replicas", 1),
151 gpu_count=data.get("gpu_count", 1),
152 gpu_type=data.get("gpu_type"),
153 model_path=data.get("model_path"),
154 model_source=data.get("model_source"),
155 health_check_path=data.get("health_check_path", "/health"),
156 env=data.get("env", {}),
157 resources=data.get("resources", {}),
158 command=data.get("command"),
159 args=data.get("args"),
160 tolerations=data.get("tolerations"),
161 node_selector=data.get("node_selector"),
162 autoscaling=data.get("autoscaling"),
163 canary=data.get("canary"),
164 capacity_type=data.get("capacity_type"),
165 )
168@dataclass
169class RegionStatus:
170 """Status of an endpoint in a specific region."""
172 region: str
173 state: str = RegionSyncState.PENDING.value
174 replicas_ready: int = 0
175 replicas_desired: int = 0
176 last_sync: str | None = None
177 error: str | None = None
178 endpoint_url: str | None = None
180 def to_dict(self) -> dict[str, Any]:
181 result: dict[str, Any] = {
182 "region": self.region,
183 "state": self.state,
184 "replicas_ready": self.replicas_ready,
185 "replicas_desired": self.replicas_desired,
186 }
187 if self.last_sync:
188 result["last_sync"] = self.last_sync
189 if self.error:
190 result["error"] = self.error
191 if self.endpoint_url:
192 result["endpoint_url"] = self.endpoint_url
193 return result
196@dataclass
197class InferenceEndpoint:
198 """An inference endpoint managed by GCO."""
200 endpoint_name: str
201 desired_state: str = EndpointState.DEPLOYING.value
202 target_regions: list[str] = field(default_factory=list)
203 namespace: str = "gco-inference"
204 spec: InferenceEndpointSpec | dict[str, Any] = field(
205 default_factory=lambda: InferenceEndpointSpec(image="")
206 )
207 ingress_path: str = ""
208 created_at: str | None = None
209 updated_at: str | None = None
210 created_by: str | None = None
211 region_status: dict[str, Any] = field(default_factory=dict)
212 labels: dict[str, str] = field(default_factory=dict)
214 def __post_init__(self) -> None:
215 if isinstance(self.spec, dict):
216 self.spec = InferenceEndpointSpec.from_dict(self.spec)
217 if not self.ingress_path:
218 self.ingress_path = f"/inference/{self.endpoint_name}"
220 def to_dict(self) -> dict[str, Any]:
221 spec_dict = (
222 self.spec.to_dict() if isinstance(self.spec, InferenceEndpointSpec) else self.spec
223 )
224 return {
225 "endpoint_name": self.endpoint_name,
226 "desired_state": self.desired_state,
227 "target_regions": self.target_regions,
228 "namespace": self.namespace,
229 "spec": spec_dict,
230 "ingress_path": self.ingress_path,
231 "created_at": self.created_at,
232 "updated_at": self.updated_at,
233 "created_by": self.created_by,
234 "region_status": self.region_status,
235 "labels": self.labels,
236 }
238 @classmethod
239 def from_dict(cls, data: dict[str, Any]) -> InferenceEndpoint:
240 return cls(
241 endpoint_name=data["endpoint_name"],
242 desired_state=data.get("desired_state", EndpointState.DEPLOYING.value),
243 target_regions=data.get("target_regions", []),
244 namespace=data.get("namespace", "gco-inference"),
245 spec=data.get("spec", {}),
246 ingress_path=data.get("ingress_path", ""),
247 created_at=data.get("created_at"),
248 updated_at=data.get("updated_at"),
249 created_by=data.get("created_by"),
250 region_status=data.get("region_status", {}),
251 labels=data.get("labels", {}),
252 )