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