Coverage for gco / services / api_shared.py: 100.00%
211 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Shared state, models, and helpers for the Manifest API routers.
4This module holds the global state (manifest processor, DynamoDB stores),
5Pydantic request/response models, and helper functions used across all API
6route modules. Centralizing them here avoids circular imports between
7manifest_api.py and the routers.
8"""
10from __future__ import annotations
12import logging
13from enum import StrEnum
14from typing import Any
16from fastapi import HTTPException
17from kubernetes.client.models import CoreV1Event, V1Job, V1Pod
18from pydantic import BaseModel, Field
20from gco.services.manifest_processor import ManifestProcessor
21from gco.services.metrics_publisher import ManifestProcessorMetrics
22from gco.services.request_context import current_request_id
23from gco.services.structured_logging import sanitize_log_value
24from gco.services.template_store import (
25 JobStore,
26 TemplateStore,
27 WebhookStore,
28)
30logger = logging.getLogger(__name__)
33def internal_server_error(context: str, error: Exception) -> HTTPException:
34 """Log the full exception server-side; return a generic, correlatable 500.
36 The client-facing detail carries only the constant message plus the
37 request's correlation id — never exception text (the information-exposure
38 shape CodeQL flagged across the jobs routes). The id also lands in the
39 paired log line, so an operator can grep the service logs for exactly
40 the failure a caller reported. Callers raise the returned exception with
41 ``from error`` so the traceback chain stays intact.
42 """
43 request_id = current_request_id()
44 logger.error("Error %s (request-id %s): %s", context, request_id, error)
45 return HTTPException(
46 status_code=500,
47 detail=f"Internal server error (request-id: {request_id})",
48 )
51# ---------------------------------------------------------------------------
52# Shared enums and Pydantic models
53# ---------------------------------------------------------------------------
56class SortOrder(StrEnum):
57 ASC = "asc"
58 DESC = "desc"
61class JobStatus(StrEnum):
62 PENDING = "pending"
63 RUNNING = "running"
64 COMPLETED = "completed"
65 SUCCEEDED = "succeeded"
66 FAILED = "failed"
69class WebhookEvent(StrEnum):
70 JOB_COMPLETED = "job.completed"
71 JOB_FAILED = "job.failed"
72 JOB_STARTED = "job.started"
75class ManifestSubmissionAPIRequest(BaseModel):
76 """API model for manifest submission requests."""
78 manifests: list[dict[str, Any]] = Field(
79 ..., description="List of Kubernetes manifests to apply"
80 )
81 namespace: str | None = Field(
82 None, description="Default namespace for resources without namespace specified"
83 )
84 dry_run: bool = Field(False, description="If true, validate manifests without applying them")
85 validate_manifests: bool = Field(
86 True, description="If true, perform validation checks on manifests", alias="validate"
87 )
89 model_config = {
90 "json_schema_extra": {
91 "example": {
92 "manifests": [
93 {"apiVersion": "batch/v1", "kind": "Job", "metadata": {"name": "example"}}
94 ],
95 "namespace": "gco-jobs",
96 "dry_run": False,
97 }
98 }
99 }
102class ResourceIdentifier(BaseModel):
103 api_version: str = Field(..., description="Kubernetes API version (e.g., 'apps/v1')")
104 kind: str = Field(..., description="Kubernetes resource kind (e.g., 'Deployment')")
105 name: str = Field(..., description="Resource name")
106 namespace: str = Field(..., description="Resource namespace")
109class BulkDeleteRequest(BaseModel):
110 namespace: str | None = Field(None, description="Filter by namespace")
111 status: JobStatus | None = Field(None, description="Filter by status")
112 older_than_days: int | None = Field(
113 None, description="Delete jobs older than N days", ge=1, le=365
114 )
115 label_selector: str | None = Field(
116 None,
117 description="Comma-separated exact-match label filters (key=value only)",
118 max_length=1024,
119 )
120 dry_run: bool = Field(False, description="If true, only return what would be deleted")
122 model_config = {
123 "json_schema_extra": {
124 "example": {
125 "namespace": "gco-jobs",
126 "status": "completed",
127 "older_than_days": 7,
128 "dry_run": False,
129 }
130 }
131 }
134class JobTemplateRequest(BaseModel):
135 name: str = Field(..., description="Template name", min_length=1, max_length=63)
136 description: str | None = Field(None, description="Template description")
137 manifest: dict[str, Any] = Field(..., description="Job manifest template")
138 parameters: dict[str, Any] | None = Field(None, description="Default parameter values")
140 model_config = {
141 "json_schema_extra": {
142 "example": {
143 "name": "gpu-training-template",
144 "description": "Template for GPU training jobs",
145 "manifest": {
146 "apiVersion": "batch/v1",
147 "kind": "Job",
148 "metadata": {"name": "{{name}}"},
149 },
150 "parameters": {"image": "pytorch/pytorch:latest"},
151 }
152 }
153 }
156class JobFromTemplateRequest(BaseModel):
157 name: str = Field(..., description="Job name", min_length=1, max_length=63)
158 namespace: str = Field("gco-jobs", description="Target namespace")
159 parameters: dict[str, Any] | None = Field(None, description="Parameter overrides")
161 model_config = {
162 "json_schema_extra": {
163 "example": {
164 "name": "my-training-job",
165 "namespace": "gco-jobs",
166 "parameters": {"image": "my-custom-image:v1"},
167 }
168 }
169 }
172class WebhookRequest(BaseModel):
173 url: str = Field(..., description="Webhook URL to call")
174 events: list[WebhookEvent] = Field(..., description="Events to subscribe to")
175 namespace: str | None = Field(None, description="Filter by namespace (optional)")
176 secret: str | None = Field(None, description="Secret for HMAC signature (optional)")
178 model_config = {
179 "json_schema_extra": {
180 "example": {
181 "url": "https://example.com/webhook",
182 "events": ["job.completed", "job.failed"],
183 "namespace": "gco-jobs",
184 }
185 }
186 }
189class QueuedJobRequest(BaseModel):
190 manifest: dict[str, Any] = Field(..., description="Kubernetes job manifest")
191 target_region: str = Field(..., description="Target region for job execution")
192 namespace: str = Field("gco-jobs", description="Kubernetes namespace")
193 priority: int = Field(0, description="Job priority (higher = more important)", ge=0, le=100)
194 labels: dict[str, str] | None = Field(None, description="Optional labels for filtering")
195 max_spot_price: float | None = Field(
196 None,
197 gt=0,
198 description=(
199 "Optional spot price cap in USD/hour. The job is not dispatched "
200 "until the current spot price of spot_instance_type in the target "
201 "region drops to or below this value. Requires spot_instance_type."
202 ),
203 )
204 spot_instance_type: str | None = Field(
205 None,
206 description=(
207 "EC2 instance type whose spot price gates dispatch (e.g. "
208 "g5.xlarge). Requires max_spot_price."
209 ),
210 )
212 model_config = {
213 "json_schema_extra": {
214 "example": {
215 "manifest": {
216 "apiVersion": "batch/v1",
217 "kind": "Job",
218 "metadata": {"name": "my-training-job"},
219 },
220 "target_region": "us-east-1",
221 "namespace": "gco-jobs",
222 "priority": 10,
223 "max_spot_price": 0.5,
224 "spot_instance_type": "g5.xlarge",
225 }
226 }
227 }
230class PaginatedResponse(BaseModel):
231 total: int = Field(..., description="Total number of items")
232 limit: int = Field(..., description="Items per page")
233 offset: int = Field(..., description="Current offset")
234 has_more: bool = Field(..., description="Whether more items exist")
237class ErrorResponse(BaseModel):
238 error: str = Field(..., description="Error type")
239 detail: str = Field(..., description="Error details")
240 timestamp: str = Field(..., description="Error timestamp")
243# ---------------------------------------------------------------------------
244# Global state — populated by the lifespan handler in manifest_api.py
245# ---------------------------------------------------------------------------
246manifest_processor: ManifestProcessor | None = None
247manifest_metrics: ManifestProcessorMetrics | None = None
248template_store: TemplateStore | None = None
249webhook_store: WebhookStore | None = None
250job_store: JobStore | None = None
253# ---------------------------------------------------------------------------
254# Helper functions
255# ---------------------------------------------------------------------------
258def _check_processor() -> ManifestProcessor:
259 """Check if manifest processor is initialized and return it."""
260 # Import at call-time to read the global that lifespan populates on
261 # the manifest_api module (tests also patch it there).
262 from gco.services import manifest_api as _api
264 if _api.manifest_processor is None:
265 raise HTTPException(status_code=503, detail="Manifest processor not initialized")
266 return _api.manifest_processor
269def _check_namespace(namespace: str, processor: ManifestProcessor) -> None:
270 """Check if namespace is allowed."""
271 if namespace not in processor.allowed_namespaces:
272 raise HTTPException(
273 status_code=403,
274 detail=f"Namespace '{namespace}' not allowed. Allowed: {list(processor.allowed_namespaces)}",
275 )
278def _parse_job_to_dict(job: V1Job) -> dict[str, Any]:
279 """Parse a Kubernetes Job object to a dictionary."""
280 metadata = job.metadata
281 status = job.status
282 spec = job.spec
284 conditions = status.conditions or []
285 computed_status = "pending"
286 for condition in conditions:
287 if condition.type == "Complete" and condition.status == "True":
288 computed_status = "succeeded"
289 break
290 if condition.type == "Failed" and condition.status == "True":
291 computed_status = "failed"
292 break
294 if computed_status == "pending" and (status.active or 0) > 0:
295 computed_status = "running"
297 # Pull container image refs from the pod template so callers (e.g.
298 # the orphan-image cross-reference) can identify which ECR images
299 # are still in use without a second round-trip per job.
300 template = getattr(spec, "template", None)
301 pod_spec = getattr(template, "spec", None) if template is not None else None
302 containers = getattr(pod_spec, "containers", None) or []
303 init_containers = getattr(pod_spec, "init_containers", None) or []
304 container_specs = [
305 {"name": getattr(c, "name", ""), "image": getattr(c, "image", "")} for c in containers
306 ]
307 init_container_specs = [
308 {"name": getattr(c, "name", ""), "image": getattr(c, "image", "")} for c in init_containers
309 ]
311 return {
312 "metadata": {
313 "name": metadata.name,
314 "namespace": metadata.namespace,
315 "creationTimestamp": (
316 metadata.creation_timestamp.isoformat() if metadata.creation_timestamp else None
317 ),
318 "labels": metadata.labels or {},
319 "annotations": metadata.annotations or {},
320 "uid": metadata.uid,
321 },
322 "spec": {
323 "parallelism": spec.parallelism,
324 "completions": spec.completions,
325 "backoffLimit": spec.backoff_limit,
326 "template": {
327 "spec": {
328 "containers": container_specs,
329 "initContainers": init_container_specs,
330 },
331 },
332 },
333 "status": {
334 "active": status.active or 0,
335 "succeeded": status.succeeded or 0,
336 "failed": status.failed or 0,
337 "startTime": status.start_time.isoformat() if status.start_time else None,
338 "completionTime": (
339 status.completion_time.isoformat() if status.completion_time else None
340 ),
341 "conditions": [
342 {
343 "type": c.type,
344 "status": c.status,
345 "reason": c.reason,
346 "message": c.message,
347 "lastTransitionTime": (
348 c.last_transition_time.isoformat() if c.last_transition_time else None
349 ),
350 }
351 for c in conditions
352 ],
353 },
354 "computed_status": computed_status,
355 }
358def _parse_pod_to_dict(pod: V1Pod) -> dict[str, Any]:
359 """Parse a Kubernetes Pod object to a dictionary."""
360 metadata = pod.metadata
361 status = pod.status
362 spec = pod.spec
364 container_statuses = []
365 for cs in status.container_statuses or []:
366 container_status: dict[str, Any] = {
367 "name": cs.name,
368 "ready": cs.ready,
369 "restartCount": cs.restart_count,
370 "image": cs.image,
371 }
372 if cs.state:
373 if cs.state.running:
374 container_status["state"] = "running"
375 container_status["startedAt"] = (
376 cs.state.running.started_at.isoformat() if cs.state.running.started_at else None
377 )
378 elif cs.state.waiting:
379 container_status["state"] = "waiting"
380 container_status["reason"] = cs.state.waiting.reason
381 elif cs.state.terminated:
382 container_status["state"] = "terminated"
383 container_status["exitCode"] = cs.state.terminated.exit_code
384 container_status["reason"] = cs.state.terminated.reason
385 container_statuses.append(container_status)
387 init_container_statuses = []
388 for cs in status.init_container_statuses or []:
389 init_status = {
390 "name": cs.name,
391 "ready": cs.ready,
392 "restartCount": cs.restart_count,
393 }
394 init_container_statuses.append(init_status)
396 return {
397 "metadata": {
398 "name": metadata.name,
399 "namespace": metadata.namespace,
400 "creationTimestamp": (
401 metadata.creation_timestamp.isoformat() if metadata.creation_timestamp else None
402 ),
403 "labels": metadata.labels or {},
404 "uid": metadata.uid,
405 },
406 "spec": {
407 "nodeName": spec.node_name,
408 "containers": [{"name": c.name, "image": c.image} for c in spec.containers],
409 "initContainers": [
410 {"name": c.name, "image": c.image} for c in (spec.init_containers or [])
411 ],
412 },
413 "status": {
414 "phase": status.phase,
415 "hostIP": status.host_ip,
416 "podIP": status.pod_ip,
417 "startTime": status.start_time.isoformat() if status.start_time else None,
418 "containerStatuses": container_statuses,
419 "initContainerStatuses": init_container_statuses,
420 },
421 }
424# ---------------------------------------------------------------------------
425# Node placement reporting
426# ---------------------------------------------------------------------------
428# The instance type a pod actually landed on. A workload constrained to a set
429# of interchangeable instance types (nodeAffinity ``In: [...]``) is placed by
430# Karpenter within that set, so the manifest only records what the run was
431# authorized to use — this label records what it used.
432NODE_INSTANCE_TYPE_LABEL = "node.kubernetes.io/instance-type"
434# Spot vs on-demand. Distinguishing the two matters for reconciling observed
435# cost against an estimate, and for explaining an interrupted run.
436NODE_CAPACITY_TYPE_LABEL = "karpenter.sh/capacity-type"
438# Reported alongside the two above because they cost nothing extra (they come
439# from the same Node read) and answer the immediate follow-up questions:
440# which AZ, which CPU architecture, which Karpenter NodePool provisioned it.
441_REPORTED_NODE_LABELS: tuple[str, ...] = (
442 NODE_INSTANCE_TYPE_LABEL,
443 NODE_CAPACITY_TYPE_LABEL,
444 "topology.kubernetes.io/zone",
445 "topology.kubernetes.io/region",
446 "kubernetes.io/arch",
447 "karpenter.sh/nodepool",
448)
451def _empty_scheduling_info() -> dict[str, Any]:
452 """The shape returned when nothing about placement is known yet."""
453 return {
454 "node_name": None,
455 "node_instance_type": None,
456 "node_capacity_type": None,
457 "node_labels": {},
458 "nodes": [],
459 "unscheduled_pods": 0,
460 "node_lookup_error": None,
461 }
464def _parse_node_to_dict(node: Any, name: str) -> dict[str, Any]:
465 """Reduce a Kubernetes Node to the placement facts callers ask for.
467 ``name`` is the node name the pod reported, which is also the key this
468 Node was read by — carrying it through keeps the pod/node join exact
469 regardless of what the Node object echoes back.
471 Label values are accepted only when they are genuinely strings, so a
472 malformed or partially-populated Node cannot put a non-serializable value
473 into the response and turn a successful job read into a 500.
474 """
475 metadata = getattr(node, "metadata", None)
476 raw_labels = getattr(metadata, "labels", None) if metadata is not None else None
477 labels: dict[str, str] = (
478 {k: v for k, v in raw_labels.items() if isinstance(k, str) and isinstance(v, str)}
479 if isinstance(raw_labels, dict)
480 else {}
481 )
482 return {
483 "name": name,
484 "instance_type": labels.get(NODE_INSTANCE_TYPE_LABEL),
485 "capacity_type": labels.get(NODE_CAPACITY_TYPE_LABEL),
486 "labels": {key: labels[key] for key in _REPORTED_NODE_LABELS if labels.get(key)},
487 }
490def _collect_pod_scheduling(core_v1: Any, pods: list[V1Pod]) -> dict[str, Any]:
491 """Report which nodes a workload's pods landed on, and each node's hardware.
493 One Node read per *distinct* node, so the common single-pod job costs
494 exactly one extra API call on a path already talking to the cluster.
496 ``node_name`` / ``node_instance_type`` / ``node_capacity_type`` describe
497 the earliest-created scheduled pod, which is stable for the life of the
498 workload even as retries add later pods. ``nodes`` carries every node
499 involved, with the pods on each, so a retried job that moved between
500 instance types is still fully described.
502 Never raises. A Node read that is refused (no RBAC) or 404s (node already
503 reclaimed) leaves the instance type ``None`` and records why in
504 ``node_lookup_error`` — an absent value that says so is more useful than a
505 guess that looks verified.
506 """
507 info = _empty_scheduling_info()
509 def _sort_key(pod: V1Pod) -> tuple[str, str]:
510 """Deterministic (created, name) ordering that cannot raise.
512 Both components are coerced to strings so a Node/Pod with a missing or
513 unexpected timestamp still sorts instead of aborting the whole report.
514 """
515 metadata = getattr(pod, "metadata", None)
516 created = getattr(metadata, "creation_timestamp", None) if metadata is not None else None
517 name = getattr(metadata, "name", None) if metadata is not None else None
518 stamp = ""
519 if created is not None:
520 try:
521 candidate = created.isoformat()
522 except Exception: # pragma: no cover - defensive
523 candidate = None
524 stamp = candidate if isinstance(candidate, str) else ""
525 return (stamp, name if isinstance(name, str) else "")
527 ordered = sorted(pods or [], key=_sort_key)
529 node_cache: dict[str, dict[str, Any]] = {}
530 node_order: list[str] = []
531 pods_by_node: dict[str, list[dict[str, Any]]] = {}
532 errors: list[str] = []
534 for pod in ordered:
535 metadata = getattr(pod, "metadata", None)
536 spec = getattr(pod, "spec", None)
537 status = getattr(pod, "status", None)
538 node_name = getattr(spec, "node_name", None) if spec is not None else None
539 pod_name = getattr(metadata, "name", None) if metadata is not None else None
540 phase = getattr(status, "phase", None) if status is not None else None
542 if not isinstance(node_name, str) or not node_name:
543 info["unscheduled_pods"] += 1
544 continue
546 if node_name not in node_cache:
547 node_order.append(node_name)
548 pods_by_node[node_name] = []
549 try:
550 node_cache[node_name] = _parse_node_to_dict(
551 core_v1.read_node(name=node_name), node_name
552 )
553 except Exception as exc:
554 # The node name originates outside this process and the
555 # Kubernetes error echoes it back; sanitize before logging so
556 # neither can forge log entries (CWE-117).
557 logger.warning(
558 "Could not read node %s: %s",
559 sanitize_log_value(node_name),
560 sanitize_log_value(exc),
561 )
562 errors.append(f"{node_name}: {exc}")
563 node_cache[node_name] = {
564 "name": node_name,
565 "instance_type": None,
566 "capacity_type": None,
567 "labels": {},
568 }
570 pods_by_node[node_name].append(
571 {
572 "name": pod_name if isinstance(pod_name, str) else None,
573 "phase": phase if isinstance(phase, str) else None,
574 }
575 )
577 if not node_order:
578 return info
580 info["nodes"] = [{**node_cache[name], "pods": pods_by_node[name]} for name in node_order]
581 primary = info["nodes"][0]
582 info["node_name"] = primary["name"]
583 info["node_instance_type"] = primary["instance_type"]
584 info["node_capacity_type"] = primary["capacity_type"]
585 info["node_labels"] = dict(primary["labels"])
586 if errors:
587 info["node_lookup_error"] = "; ".join(errors)
588 return info
591def _parse_event_to_dict(event: CoreV1Event) -> dict[str, Any]:
592 """Parse a Kubernetes Event object to a dictionary."""
593 return {
594 "type": event.type,
595 "reason": event.reason,
596 "message": event.message,
597 "count": event.count or 1,
598 "firstTimestamp": (event.first_timestamp.isoformat() if event.first_timestamp else None),
599 "lastTimestamp": (event.last_timestamp.isoformat() if event.last_timestamp else None),
600 "source": {
601 "component": event.source.component if event.source else None,
602 "host": event.source.host if event.source else None,
603 },
604 "involvedObject": {
605 "kind": event.involved_object.kind if event.involved_object else None,
606 "name": event.involved_object.name if event.involved_object else None,
607 "namespace": event.involved_object.namespace if event.involved_object else None,
608 },
609 }
612def _apply_template_parameters(
613 manifest: dict[str, Any], parameters: dict[str, Any]
614) -> dict[str, Any]:
615 """Apply parameter substitutions to a manifest template."""
616 import json
617 import re
619 manifest_str = json.dumps(manifest)
620 for key, value in parameters.items():
621 pattern = r"\{\{\s*" + re.escape(key) + r"\s*\}\}"
622 manifest_str = re.sub(pattern, str(value), manifest_str)
623 result: dict[str, Any] = json.loads(manifest_str)
624 return result