Coverage for gco / services / manifest_api.py: 100.00%
171 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Manifest API Service for GCO (Global Capacity Orchestrator on AWS).
4This FastAPI service provides REST endpoints for Kubernetes manifest
5submission, validation, and management. Endpoint implementations live
6in the ``api_routes`` sub-package; this module wires them together and
7owns the application lifecycle, Pydantic request/response models, and
8health probes.
10See ``api_routes/`` for the individual routers:
11 - manifests.py — manifest submit / validate / resource CRUD
12 - jobs.py — job list / get / logs / events / metrics / delete / retry
13 - templates.py — job template CRUD + create-from-template
14 - webhooks.py — webhook registration
15 - queue.py — DynamoDB-backed global job queue
16"""
18from __future__ import annotations
20import asyncio
21import logging
22import os
23from collections.abc import AsyncIterator
24from contextlib import asynccontextmanager, suppress
25from datetime import UTC, datetime
26from typing import Any
28from fastapi import FastAPI, HTTPException, Request
29from fastapi.responses import JSONResponse
31from gco.services import tracing
32from gco.services.auth_middleware import AuthenticationMiddleware
33from gco.services.central_queue_worker import CentralQueueWorker
34from gco.services.manifest_processor import (
35 ManifestProcessor,
36 create_manifest_processor_from_env,
37)
38from gco.services.metrics_publisher import ManifestProcessorMetrics
39from gco.services.request_size_middleware import (
40 DEFAULT_MAX_REQUEST_BODY_BYTES,
41 RequestSizeLimitMiddleware,
42)
43from gco.services.structured_logging import configure_structured_logging
44from gco.services.template_store import (
45 JobStore,
46 TemplateStore,
47 WebhookStore,
48 get_job_store,
49 get_template_store,
50 get_webhook_store,
51)
53# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
54# Generated at (UTC): 2026-08-30T12:00:00Z
55# Generated from Git commit: 8cb8e766461f5b7f35205e12db980988598cf6d5
56# Flowchart(s) generated from this file:
57# * ``lifespan`` -> ``diagrams/code_diagrams/gco/services/manifest_api.lifespan.html``
58# (PNG: ``diagrams/code_diagrams/gco/services/manifest_api.lifespan.png``)
59# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
60# <pyflowchart-code-diagram> END
63logging.basicConfig(
64 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
65)
66logger = logging.getLogger(__name__)
69# ---------------------------------------------------------------------------
70# Global state — populated by lifespan, read by routers via this module.
71# ---------------------------------------------------------------------------
72manifest_processor: ManifestProcessor | None = None
73manifest_metrics: ManifestProcessorMetrics | None = None
74template_store: TemplateStore | None = None
75webhook_store: WebhookStore | None = None
76job_store: JobStore | None = None
79def _env_bool(name: str, default: bool = False) -> bool:
80 """Parse an explicit deployment boolean without truthy-string surprises."""
81 raw = os.getenv(name)
82 if raw is None:
83 return default
84 return raw.strip().lower() in {"1", "true", "yes", "on"}
87def _env_number(name: str, default: float, minimum: float, maximum: float) -> float:
88 """Read a finite bounded worker setting from the environment."""
89 try:
90 value = float(os.getenv(name, str(default)))
91 except ValueError:
92 return default
93 return value if minimum <= value <= maximum else default
96# =============================================================================
97# Pydantic Models for API
98# =============================================================================
101# =============================================================================
102# Application Lifecycle
103# =============================================================================
106@asynccontextmanager
107async def lifespan(app: FastAPI) -> AsyncIterator[None]:
108 """Initialize API dependencies and the optional regional queue worker."""
109 global manifest_processor, manifest_metrics, template_store, webhook_store, job_store
111 queue_worker: CentralQueueWorker | None = None
112 queue_worker_task: asyncio.Task[None] | None = None
113 logger.info("Starting Manifest API Service")
114 try:
115 manifest_processor = create_manifest_processor_from_env()
117 configure_structured_logging(
118 service_name="manifest-api",
119 cluster_id=manifest_processor.cluster_id,
120 region=manifest_processor.region,
121 )
123 manifest_metrics = ManifestProcessorMetrics(
124 cluster_name=manifest_processor.cluster_id,
125 region=manifest_processor.region,
126 )
127 logger.info("Manifest processor initialized")
129 template_store = get_template_store()
130 webhook_store = get_webhook_store()
131 job_store = get_job_store()
132 logger.info("DynamoDB stores initialized")
134 if _env_bool("CENTRAL_QUEUE_WORKER_ENABLED"):
135 queue_worker = CentralQueueWorker(
136 processor=manifest_processor,
137 store=job_store,
138 poll_interval_seconds=_env_number(
139 "CENTRAL_QUEUE_POLL_INTERVAL_SECONDS", 10.0, 1.0, 300.0
140 ),
141 batch_size=int(_env_number("CENTRAL_QUEUE_BATCH_SIZE", 5.0, 1.0, 20.0)),
142 reconcile_limit=int(
143 _env_number("CENTRAL_QUEUE_RECONCILE_LIMIT", 100.0, 1.0, 500.0)
144 ),
145 lease_renewal_seconds=_env_number(
146 "CENTRAL_QUEUE_LEASE_RENEWAL_SECONDS", 60.0, 1.0, 300.0
147 ),
148 )
149 queue_worker_task = asyncio.create_task(
150 queue_worker.run(),
151 name=f"central-queue-worker-{manifest_processor.region}",
152 )
153 app.state.central_queue_worker = queue_worker
154 app.state.central_queue_worker_task = queue_worker_task
155 else:
156 app.state.central_queue_worker = None
157 app.state.central_queue_worker_task = None
158 except Exception as e:
159 logger.error(f"Failed to initialize manifest processor: {e}")
160 raise
162 try:
163 yield
164 finally:
165 if queue_worker is not None and queue_worker_task is not None:
166 queue_worker.stop()
167 try:
168 await asyncio.wait_for(queue_worker_task, timeout=30)
169 except TimeoutError:
170 queue_worker_task.cancel()
171 with suppress(asyncio.CancelledError):
172 await queue_worker_task
173 logger.info("Shutting down Manifest API Service")
174 tracing.shutdown_tracing()
177# =============================================================================
178# Create FastAPI app and include routers
179# =============================================================================
181app = FastAPI(
182 title="GCO Manifest Processor API",
183 description="Kubernetes manifest submission and management service for GCO (Global Capacity Orchestrator on AWS)",
184 version="2.0.0",
185 lifespan=lifespan,
186)
188# Server spans for every route except the probe and scrape paths, with the
189# cost-monitor proxy's client spans beneath them; inert unless
190# GCO_TRACING_ENABLED=true.
191tracing.configure_tracing("manifest-processor")
192tracing.instrument_fastapi_app(app)
194app.add_middleware(AuthenticationMiddleware)
196# Request size limit middleware — added after auth middleware so it executes
197# first in the request pipeline (Starlette processes middleware in LIFO order).
198_max_body_bytes = int(os.getenv("MAX_REQUEST_BODY_BYTES", str(DEFAULT_MAX_REQUEST_BODY_BYTES)))
199app.add_middleware(RequestSizeLimitMiddleware, max_body_bytes=_max_body_bytes)
201# Request correlation. Every request gets a server-generated id that is
202# echoed as the X-Request-ID response header and embedded in generic 500
203# details (see api_shared.internal_server_error), so an operator can tie a
204# client-reported failure back to the exact logged exception. The id is
205# never read from an inbound header — a client-controlled value adjacent to
206# log lines would need CWE-117 sanitization and could muddy investigations.
207from gco.services.request_context import ( # noqa: E402
208 REQUEST_ID_HEADER,
209 bind_request_id,
210 current_request_id,
211 unbind_request_id,
212)
215@app.middleware("http")
216async def request_correlation_middleware(request: Request, call_next: Any) -> Any:
217 """Bind a fresh correlation id for the request and echo it on the response."""
218 request_id, token = bind_request_id()
219 try:
220 response = await call_next(request)
221 finally:
222 unbind_request_id(token)
223 response.headers[REQUEST_ID_HEADER] = request_id
224 return response
227# Expose Prometheus /metrics for the in-cluster observability scrape. The auth
228# middleware exempts /metrics, so the cluster Prometheus reaches it over the
229# existing service port without credentials.
230from gco.services.service_metrics import mount_metrics # noqa: E402
232mount_metrics(app, "manifest-processor")
234# Include domain routers
235from gco.services.api_routes.cost import router as cost_router # noqa: E402
236from gco.services.api_routes.jobs import router as jobs_router # noqa: E402
237from gco.services.api_routes.manifests import router as manifests_router # noqa: E402
238from gco.services.api_routes.queue import router as queue_router # noqa: E402
239from gco.services.api_routes.slurm import router as slurm_router # noqa: E402
240from gco.services.api_routes.templates import router as templates_router # noqa: E402
241from gco.services.api_routes.webhooks import router as webhooks_router # noqa: E402
243app.include_router(manifests_router)
244app.include_router(jobs_router)
245app.include_router(templates_router)
246app.include_router(webhooks_router)
247app.include_router(queue_router)
248app.include_router(cost_router)
249app.include_router(slurm_router)
252# =============================================================================
253# Root & Health Endpoints (kept here — they're thin and tightly coupled to state)
254# =============================================================================
257@app.get("/", tags=["Info"])
258async def root() -> dict[str, Any]:
259 """Root endpoint with basic service information and API overview."""
260 return {
261 "service": "GCO Manifest Processor API",
262 "version": "2.0.0",
263 "status": "running",
264 "cluster_id": (manifest_processor.cluster_id if manifest_processor else "unknown"),
265 "region": (manifest_processor.region if manifest_processor else "unknown"),
266 "endpoints": {
267 "manifests": {
268 "submit": "POST /api/v1/manifests",
269 "validate": "POST /api/v1/manifests/validate",
270 "get": "GET /api/v1/manifests/{namespace}/{name}",
271 "delete": "DELETE /api/v1/manifests/{namespace}/{name}",
272 },
273 "jobs": {
274 "list": "GET /api/v1/jobs",
275 "get": "GET /api/v1/jobs/{namespace}/{name}",
276 "logs": "GET /api/v1/jobs/{namespace}/{name}/logs",
277 "events": "GET /api/v1/jobs/{namespace}/{name}/events",
278 "pods": "GET /api/v1/jobs/{namespace}/{name}/pods",
279 "metrics": "GET /api/v1/jobs/{namespace}/{name}/metrics",
280 "delete": "DELETE /api/v1/jobs/{namespace}/{name}",
281 "bulk_delete": "DELETE /api/v1/jobs",
282 "retry": "POST /api/v1/jobs/{namespace}/{name}/retry",
283 },
284 "templates": {
285 "list": "GET /api/v1/templates",
286 "create": "POST /api/v1/templates",
287 "get": "GET /api/v1/templates/{name}",
288 "delete": "DELETE /api/v1/templates/{name}",
289 "create_job": "POST /api/v1/jobs/from-template/{name}",
290 },
291 "webhooks": {
292 "list": "GET /api/v1/webhooks",
293 "create": "POST /api/v1/webhooks",
294 "delete": "DELETE /api/v1/webhooks/{id}",
295 },
296 "cost": {
297 "status": "GET /api/v1/cost/status",
298 "reports": "GET /api/v1/cost/reports",
299 "generate_report": "POST /api/v1/cost/reports",
300 },
301 "health": "GET /api/v1/health",
302 "status": "GET /api/v1/status",
303 "policy": "GET /api/v1/policy",
304 },
305 }
308@app.get("/healthz", tags=["Health"])
309async def kubernetes_health_check() -> dict[str, str]:
310 """Kubernetes-style liveness probe."""
311 return {"status": "ok"}
314@app.get("/readyz", tags=["Health"])
315async def kubernetes_readiness_check() -> dict[str, str]:
316 """Kubernetes readiness includes the enabled queue worker task."""
317 if manifest_processor is None:
318 raise HTTPException(status_code=503, detail="Manifest processor not ready")
319 worker_task = getattr(app.state, "central_queue_worker_task", None)
320 if worker_task is not None and worker_task.done():
321 raise HTTPException(status_code=503, detail="Central queue worker stopped unexpectedly")
322 return {"status": "ready"}
325@app.get("/api/v1/health", tags=["Health"])
326async def health_check() -> JSONResponse:
327 """Health check endpoint for load balancer health checks."""
328 try:
329 if manifest_processor is None:
330 return JSONResponse(
331 status_code=503,
332 content={
333 "status": "unhealthy",
334 "timestamp": datetime.now(UTC).isoformat(),
335 "message": "Manifest processor not initialized",
336 },
337 )
339 try:
340 manifest_processor.core_v1.list_namespace(limit=1)
341 api_healthy = True
342 except Exception as e:
343 logger.error(f"Kubernetes API health check failed: {e}")
344 api_healthy = False
346 status_code = 200 if api_healthy else 503
347 return JSONResponse(
348 status_code=status_code,
349 content={
350 "status": "healthy" if api_healthy else "unhealthy",
351 "timestamp": datetime.now(UTC).isoformat(),
352 "cluster_id": manifest_processor.cluster_id,
353 "region": manifest_processor.region,
354 "kubernetes_api": "connected" if api_healthy else "disconnected",
355 },
356 )
358 except Exception as e:
359 logger.error(f"Health check failed: {e}")
360 return JSONResponse(
361 status_code=503,
362 content={
363 "status": "unhealthy",
364 "timestamp": datetime.now(UTC).isoformat(),
365 "error": "manifest processor unavailable",
366 },
367 )
370@app.get("/api/v1/status", tags=["Health"])
371async def get_service_status() -> dict[str, Any]:
372 """Service status endpoint with detailed information."""
373 templates_count = 0
374 webhooks_count = 0
375 try:
376 if template_store:
377 templates_count = len(template_store.list_templates())
378 if webhook_store:
379 webhooks_count = len(webhook_store.list_webhooks())
380 except Exception as e:
381 logger.warning(f"Failed to get store counts: {e}")
383 status_info: dict[str, Any] = {
384 "service": "GCO Manifest Processor API",
385 "version": "2.0.0",
386 "timestamp": datetime.now(UTC).isoformat(),
387 "manifest_processor_initialized": manifest_processor is not None,
388 "environment": {
389 "cluster_name": os.getenv("CLUSTER_NAME", "unknown"),
390 "region": os.getenv("REGION", "unknown"),
391 "max_cpu_per_manifest": os.getenv("MAX_CPU_PER_MANIFEST", "10"),
392 "max_memory_per_manifest": os.getenv("MAX_MEMORY_PER_MANIFEST", "32Gi"),
393 "max_gpu_per_manifest": os.getenv("MAX_GPU_PER_MANIFEST", "4"),
394 "allowed_namespaces": os.getenv("ALLOWED_NAMESPACES", "gco-jobs"),
395 "validation_enabled": os.getenv("VALIDATION_ENABLED", "true"),
396 },
397 "templates_count": templates_count,
398 "webhooks_count": webhooks_count,
399 "central_queue_worker": (
400 worker.health()
401 if (worker := getattr(app.state, "central_queue_worker", None)) is not None
402 else {"enabled": False, "running": False}
403 ),
404 }
406 if manifest_processor:
407 status_info.update(
408 {
409 "cluster_id": manifest_processor.cluster_id,
410 "region": manifest_processor.region,
411 "resource_limits": {
412 "max_cpu_millicores": manifest_processor.max_cpu_per_manifest,
413 "max_memory_bytes": manifest_processor.max_memory_per_manifest,
414 "max_gpu_count": manifest_processor.max_gpu_per_manifest,
415 },
416 "allowed_namespaces": list(manifest_processor.allowed_namespaces),
417 "validation_enabled": manifest_processor.validation_enabled,
418 }
419 )
421 return status_info
424@app.get("/api/v1/policy", tags=["Health"])
425async def get_job_validation_policy() -> dict[str, Any]:
426 """The validation policy this region actually enforces, as deployed.
428 Answers "will this cluster admit the job I am about to pay to run?"
429 before submission, so a policy conflict surfaces at plan time instead of
430 after a region has been provisioned and billed.
432 Reads the live ``ManifestProcessor`` instance rather than any config file.
433 A local ``cdk.json`` is the *input* to a deploy, not the state of one:
434 the cluster may have been deployed from a different checkout, and CDK
435 augments ``trusted_registries`` with the project's own ECR hostnames at
436 synth time, so the effective allowlist is strictly larger than the
437 configured one.
439 Three layers govern admission and all three are reported:
441 1. ``policy`` — the front-door checks the manifest processor and the SQS
442 queue processor both apply (they read the same env vars, so neither
443 submission path is a bypass).
444 2. ``cluster_enforcement.limit_ranges`` — per-container ceilings.
445 3. ``cluster_enforcement.resource_quotas`` — namespace aggregate ceilings.
447 A manifest must clear all three. Layers 2 and 3 are read live from the
448 Kubernetes API and degrade to ``status="unavailable"`` rather than
449 failing the whole response.
450 """
451 if manifest_processor is None:
452 raise HTTPException(status_code=503, detail="Manifest processor not ready")
454 return {
455 "service": "GCO Manifest Processor API",
456 "timestamp": datetime.now(UTC).isoformat(),
457 "cluster_id": manifest_processor.cluster_id,
458 "region": manifest_processor.region,
459 # Names the origin of these values so a caller never mistakes this for
460 # a config-file read.
461 "source": "deployed-cluster-runtime",
462 "policy": manifest_processor.effective_job_validation_policy(),
463 "cluster_enforcement": manifest_processor.cluster_resource_governance(),
464 }
467# =============================================================================
468# Error Handlers
469# =============================================================================
472@app.exception_handler(Exception)
473async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
474 """Global exception handler for unhandled errors."""
475 request_id = current_request_id()
476 logger.error(
477 f"Unhandled exception in {request.method} {request.url} (request-id {request_id}): {exc}"
478 )
479 return JSONResponse(
480 status_code=500,
481 content={
482 "error": "Internal server error",
483 "detail": str(exc) if os.getenv("DEBUG") else "An unexpected error occurred",
484 "request_id": request_id,
485 "timestamp": datetime.now(UTC).isoformat(),
486 },
487 )
490# =============================================================================
491# App Factory & Entrypoint
492# =============================================================================
495def create_app() -> FastAPI:
496 """Factory function to create the FastAPI app."""
497 return app
500# The pod manifest gives the kubelet terminationGracePeriodSeconds > preStop +
501# this budget, so Uvicorn can finish in-flight requests before SIGKILL. The
502# same variable drives the TLS sidecar's drain (gco.services.tls_proxy).
503DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 50
506def _run_server() -> None:
507 """Run Uvicorn with the same drain budget declared by the pod manifest."""
508 import uvicorn
510 from gco.services.uvicorn_runtime import describe_uvicorn_runtime
512 host = os.getenv("HOST", "0.0.0.0") # container listener
513 port = int(os.getenv("PORT", "8080"))
514 log_level = os.getenv("LOG_LEVEL", "info").lower()
515 graceful_shutdown_seconds = int(
516 os.getenv(
517 "GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS",
518 str(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS),
519 )
520 )
522 loop_impl, http_impl = describe_uvicorn_runtime()
523 logger.info(
524 "Starting Manifest API on %s:%d (uvicorn loop=%s http=%s)",
525 host,
526 port,
527 loop_impl,
528 http_impl,
529 )
531 uvicorn.run(
532 "gco.services.manifest_api:app",
533 host=host,
534 port=port,
535 log_level=log_level,
536 reload=False,
537 timeout_graceful_shutdown=graceful_shutdown_seconds,
538 )
541if __name__ == "__main__":
542 _run_server()