Coverage for gco / services / api_shared.py: 100.00%
241 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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.
454 Every key the collector can fill in is present here, so a caller sees the
455 same field set whether placement resolved, no pods exist yet, or the pod
456 listing itself failed (``node_lookup_error`` says which).
457 """
458 return {
459 "node_name": None,
460 "node_instance_type": None,
461 "node_capacity_type": None,
462 "node_labels": {},
463 "nodes": [],
464 "pod_phase": None,
465 "pod_phases": {},
466 "scheduled_pods": 0,
467 "unscheduled_pods": 0,
468 "unscheduled": [],
469 "node_lookup_error": None,
470 }
473def _pod_scheduled_condition(status: Any) -> dict[str, Any]:
474 """The ``PodScheduled`` condition of a pod, reduced to its reason and message.
476 Kubernetes already states why a pod is not placed ("0/6 nodes are
477 available: ...", a taint nobody tolerates, Karpenter's nomination churn)
478 on this condition; surfacing it here saves the second ``events`` call a
479 caller otherwise needs to learn the same thing. Only the ``False`` state
480 carries a diagnosis; ``True`` and ``Unknown`` report nothing, as does a pod
481 whose conditions have not been populated yet. Never raises: every field
482 is taken with ``getattr`` and kept only when it has the expected type.
483 """
484 empty: dict[str, Any] = {"reason": None, "message": None, "since": None}
485 conditions = getattr(status, "conditions", None) if status is not None else None
486 if not isinstance(conditions, list):
487 return empty
488 for condition in conditions:
489 if getattr(condition, "type", None) != "PodScheduled":
490 continue
491 if getattr(condition, "status", None) != "False":
492 return empty
493 reason = getattr(condition, "reason", None)
494 message = getattr(condition, "message", None)
495 transition = getattr(condition, "last_transition_time", None)
496 since = None
497 if transition is not None:
498 try:
499 since = transition.isoformat()
500 except Exception: # pragma: no cover - defensive
501 since = None
502 return {
503 "reason": reason if isinstance(reason, str) else None,
504 "message": message if isinstance(message, str) else None,
505 "since": since if isinstance(since, str) else None,
506 }
507 return empty
510def _parse_node_to_dict(node: Any, name: str) -> dict[str, Any]:
511 """Reduce a Kubernetes Node to the placement facts callers ask for.
513 ``name`` is the node name the pod reported, which is also the key this
514 Node was read by — carrying it through keeps the pod/node join exact
515 regardless of what the Node object echoes back.
517 Label values are accepted only when they are genuinely strings, so a
518 malformed or partially-populated Node cannot put a non-serializable value
519 into the response and turn a successful job read into a 500.
520 """
521 metadata = getattr(node, "metadata", None)
522 raw_labels = getattr(metadata, "labels", None) if metadata is not None else None
523 labels: dict[str, str] = (
524 {k: v for k, v in raw_labels.items() if isinstance(k, str) and isinstance(v, str)}
525 if isinstance(raw_labels, dict)
526 else {}
527 )
528 return {
529 "name": name,
530 "instance_type": labels.get(NODE_INSTANCE_TYPE_LABEL),
531 "capacity_type": labels.get(NODE_CAPACITY_TYPE_LABEL),
532 "labels": {key: labels[key] for key in _REPORTED_NODE_LABELS if labels.get(key)},
533 }
536def _collect_pod_scheduling(core_v1: Any, pods: list[V1Pod]) -> dict[str, Any]:
537 """Report which nodes a workload's pods landed on, and each node's hardware.
539 One Node read per *distinct* node, so the common single-pod job costs
540 exactly one extra API call on a path already talking to the cluster.
542 ``node_name`` / ``node_instance_type`` / ``node_capacity_type`` describe
543 the earliest-created scheduled pod, which is stable for the life of the
544 workload even as retries add later pods. ``nodes`` carries every node
545 involved, with the pods on each, so a retried job that moved between
546 instance types is still fully described.
548 The pods' own state is reported next to the placement, because the Job's
549 status alone cannot tell "the container is training" from "a Job object
550 exists and nothing is placed": a Job whose only pod cannot be scheduled
551 still counts as active. ``pod_phases`` counts every pod by phase;
552 ``pod_phase`` is the one phase they all share (``None`` as soon as they
553 disagree, so it is never a summary that hides a Pending pod behind a
554 Running one); ``scheduled_pods`` / ``unscheduled_pods`` split them by
555 whether a node has been assigned; and ``unscheduled`` names each unplaced
556 pod with the ``PodScheduled`` condition's reason and message, which is
557 where Kubernetes states why it is waiting.
559 Never raises. A Node read that is refused (no RBAC) or 404s (node already
560 reclaimed) leaves the instance type ``None`` and records why in
561 ``node_lookup_error`` — an absent value that says so is more useful than a
562 guess that looks verified.
563 """
564 info = _empty_scheduling_info()
565 phases: dict[str, int] = {}
567 def _sort_key(pod: V1Pod) -> tuple[str, str]:
568 """Deterministic (created, name) ordering that cannot raise.
570 Both components are coerced to strings so a Node/Pod with a missing or
571 unexpected timestamp still sorts instead of aborting the whole report.
572 """
573 metadata = getattr(pod, "metadata", None)
574 created = getattr(metadata, "creation_timestamp", None) if metadata is not None else None
575 name = getattr(metadata, "name", None) if metadata is not None else None
576 stamp = ""
577 if created is not None:
578 try:
579 candidate = created.isoformat()
580 except Exception: # pragma: no cover - defensive
581 candidate = None
582 stamp = candidate if isinstance(candidate, str) else ""
583 return (stamp, name if isinstance(name, str) else "")
585 ordered = sorted(pods or [], key=_sort_key)
587 node_cache: dict[str, dict[str, Any]] = {}
588 node_order: list[str] = []
589 pods_by_node: dict[str, list[dict[str, Any]]] = {}
590 errors: list[str] = []
592 for pod in ordered:
593 metadata = getattr(pod, "metadata", None)
594 spec = getattr(pod, "spec", None)
595 status = getattr(pod, "status", None)
596 node_name = getattr(spec, "node_name", None) if spec is not None else None
597 pod_name = getattr(metadata, "name", None) if metadata is not None else None
598 phase = getattr(status, "phase", None) if status is not None else None
599 if not isinstance(pod_name, str):
600 pod_name = None
601 if not isinstance(phase, str):
602 phase = None
603 phases[phase or "Unknown"] = phases.get(phase or "Unknown", 0) + 1
605 if not isinstance(node_name, str) or not node_name:
606 # No Node read for a pod that has no node: the diagnosis lives on
607 # the pod's own PodScheduled condition, which is already in hand.
608 info["unscheduled_pods"] += 1
609 info["unscheduled"].append(
610 {"name": pod_name, "phase": phase, **_pod_scheduled_condition(status)}
611 )
612 continue
614 info["scheduled_pods"] += 1
616 if node_name not in node_cache:
617 node_order.append(node_name)
618 pods_by_node[node_name] = []
619 try:
620 node_cache[node_name] = _parse_node_to_dict(
621 core_v1.read_node(name=node_name), node_name
622 )
623 except Exception as exc:
624 # The node name originates outside this process and the
625 # Kubernetes error echoes it back; sanitize before logging so
626 # neither can forge log entries (CWE-117).
627 logger.warning(
628 "Could not read node %s: %s",
629 sanitize_log_value(node_name),
630 sanitize_log_value(exc),
631 )
632 errors.append(f"{node_name}: {exc}")
633 node_cache[node_name] = {
634 "name": node_name,
635 "instance_type": None,
636 "capacity_type": None,
637 "labels": {},
638 }
640 pods_by_node[node_name].append({"name": pod_name, "phase": phase})
642 info["pod_phases"] = phases
643 if len(phases) == 1:
644 # One phase shared by every pod is the only aggregate that cannot
645 # mislead; anything mixed stays None and the caller reads the counts.
646 (info["pod_phase"],) = phases
648 if not node_order:
649 return info
651 info["nodes"] = [{**node_cache[name], "pods": pods_by_node[name]} for name in node_order]
652 primary = info["nodes"][0]
653 info["node_name"] = primary["name"]
654 info["node_instance_type"] = primary["instance_type"]
655 info["node_capacity_type"] = primary["capacity_type"]
656 info["node_labels"] = dict(primary["labels"])
657 if errors:
658 info["node_lookup_error"] = "; ".join(errors)
659 return info
662def _parse_event_to_dict(event: CoreV1Event) -> dict[str, Any]:
663 """Parse a Kubernetes Event object to a dictionary."""
664 return {
665 "type": event.type,
666 "reason": event.reason,
667 "message": event.message,
668 "count": event.count or 1,
669 "firstTimestamp": (event.first_timestamp.isoformat() if event.first_timestamp else None),
670 "lastTimestamp": (event.last_timestamp.isoformat() if event.last_timestamp else None),
671 "source": {
672 "component": event.source.component if event.source else None,
673 "host": event.source.host if event.source else None,
674 },
675 "involvedObject": {
676 "kind": event.involved_object.kind if event.involved_object else None,
677 "name": event.involved_object.name if event.involved_object else None,
678 "namespace": event.involved_object.namespace if event.involved_object else None,
679 },
680 }
683def _apply_template_parameters(
684 manifest: dict[str, Any], parameters: dict[str, Any]
685) -> dict[str, Any]:
686 """Apply parameter substitutions to a manifest template."""
687 import json
688 import re
690 manifest_str = json.dumps(manifest)
691 for key, value in parameters.items():
692 pattern = r"\{\{\s*" + re.escape(key) + r"\s*\}\}"
693 manifest_str = re.sub(pattern, str(value), manifest_str)
694 result: dict[str, Any] = json.loads(manifest_str)
695 return result