Coverage for gco/services/manifest_api.py: 80.25%
139 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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-07-18T01:03:40Z
54# Flowchart(s) generated from this file:
55# * ``lifespan`` -> ``diagrams/code_diagrams/gco/services/manifest_api.lifespan.html``
56# (PNG: ``diagrams/code_diagrams/gco/services/manifest_api.lifespan.png``)
57# Regenerate with ``python diagrams/code_diagrams/generate.py``.
58# <pyflowchart-code-diagram> END
61logging.basicConfig(
62 level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
63)
64logger = logging.getLogger(__name__)
67# ---------------------------------------------------------------------------
68# Global state — populated by lifespan, read by routers via this module.
69# ---------------------------------------------------------------------------
70manifest_processor: ManifestProcessor | None = None
71manifest_metrics: ManifestProcessorMetrics | None = None
72template_store: TemplateStore | None = None
73webhook_store: WebhookStore | None = None
74job_store: JobStore | None = None
77def _env_bool(name: str, default: bool = False) -> bool:
78 """Parse an explicit deployment boolean without truthy-string surprises."""
79 raw = os.getenv(name)
80 if raw is None: 80 ↛ 82line 80 didn't jump to line 82 because the condition on line 80 was always true
81 return default
82 return raw.strip().lower() in {"1", "true", "yes", "on"}
85def _env_number(name: str, default: float, minimum: float, maximum: float) -> float:
86 """Read a finite bounded worker setting from the environment."""
87 try:
88 value = float(os.getenv(name, str(default)))
89 except ValueError:
90 return default
91 return value if minimum <= value <= maximum else default
94# =============================================================================
95# Pydantic Models for API
96# =============================================================================
99# =============================================================================
100# Application Lifecycle
101# =============================================================================
104@asynccontextmanager
105async def lifespan(app: FastAPI) -> AsyncIterator[None]:
106 """Initialize API dependencies and the optional regional queue worker."""
107 global manifest_processor, manifest_metrics, template_store, webhook_store, job_store
109 queue_worker: CentralQueueWorker | None = None
110 queue_worker_task: asyncio.Task[None] | None = None
111 logger.info("Starting Manifest API Service")
112 try:
113 manifest_processor = create_manifest_processor_from_env()
115 configure_structured_logging(
116 service_name="manifest-api",
117 cluster_id=manifest_processor.cluster_id,
118 region=manifest_processor.region,
119 )
121 manifest_metrics = ManifestProcessorMetrics(
122 cluster_name=manifest_processor.cluster_id,
123 region=manifest_processor.region,
124 )
125 logger.info("Manifest processor initialized")
127 template_store = get_template_store()
128 webhook_store = get_webhook_store()
129 job_store = get_job_store()
130 logger.info("DynamoDB stores initialized")
132 if _env_bool("CENTRAL_QUEUE_WORKER_ENABLED"): 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true
133 queue_worker = CentralQueueWorker(
134 processor=manifest_processor,
135 store=job_store,
136 poll_interval_seconds=_env_number(
137 "CENTRAL_QUEUE_POLL_INTERVAL_SECONDS", 10.0, 1.0, 300.0
138 ),
139 batch_size=int(_env_number("CENTRAL_QUEUE_BATCH_SIZE", 5.0, 1.0, 20.0)),
140 reconcile_limit=int(
141 _env_number("CENTRAL_QUEUE_RECONCILE_LIMIT", 100.0, 1.0, 500.0)
142 ),
143 lease_renewal_seconds=_env_number(
144 "CENTRAL_QUEUE_LEASE_RENEWAL_SECONDS", 60.0, 1.0, 300.0
145 ),
146 )
147 queue_worker_task = asyncio.create_task(
148 queue_worker.run(),
149 name=f"central-queue-worker-{manifest_processor.region}",
150 )
151 app.state.central_queue_worker = queue_worker
152 app.state.central_queue_worker_task = queue_worker_task
153 else:
154 app.state.central_queue_worker = None
155 app.state.central_queue_worker_task = None
156 except Exception as e:
157 logger.error(f"Failed to initialize manifest processor: {e}")
158 raise
160 try:
161 yield
162 finally:
163 if queue_worker is not None and queue_worker_task is not None: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true
164 queue_worker.stop()
165 try:
166 await asyncio.wait_for(queue_worker_task, timeout=30)
167 except TimeoutError:
168 queue_worker_task.cancel()
169 with suppress(asyncio.CancelledError):
170 await queue_worker_task
171 logger.info("Shutting down Manifest API Service")
174# =============================================================================
175# Create FastAPI app and include routers
176# =============================================================================
178app = FastAPI(
179 title="GCO Manifest Processor API",
180 description="Kubernetes manifest submission and management service for GCO (Global Capacity Orchestrator on AWS)",
181 version="2.0.0",
182 lifespan=lifespan,
183)
185app.add_middleware(AuthenticationMiddleware)
187# Request size limit middleware — added after auth middleware so it executes
188# first in the request pipeline (Starlette processes middleware in LIFO order).
189_max_body_bytes = int(os.getenv("MAX_REQUEST_BODY_BYTES", str(DEFAULT_MAX_REQUEST_BODY_BYTES)))
190app.add_middleware(RequestSizeLimitMiddleware, max_body_bytes=_max_body_bytes)
192# Expose Prometheus /metrics for the in-cluster observability scrape. The auth
193# middleware exempts /metrics, so the cluster Prometheus reaches it over the
194# existing service port without credentials.
195from gco.services.service_metrics import mount_metrics # noqa: E402
197mount_metrics(app, "manifest-processor")
199# Include domain routers
200from gco.services.api_routes.cost import router as cost_router # noqa: E402
201from gco.services.api_routes.jobs import router as jobs_router # noqa: E402
202from gco.services.api_routes.manifests import router as manifests_router # noqa: E402
203from gco.services.api_routes.queue import router as queue_router # noqa: E402
204from gco.services.api_routes.templates import router as templates_router # noqa: E402
205from gco.services.api_routes.webhooks import router as webhooks_router # noqa: E402
207app.include_router(manifests_router)
208app.include_router(jobs_router)
209app.include_router(templates_router)
210app.include_router(webhooks_router)
211app.include_router(queue_router)
212app.include_router(cost_router)
215# =============================================================================
216# Root & Health Endpoints (kept here — they're thin and tightly coupled to state)
217# =============================================================================
220@app.get("/", tags=["Info"])
221async def root() -> dict[str, Any]:
222 """Root endpoint with basic service information and API overview."""
223 return {
224 "service": "GCO Manifest Processor API",
225 "version": "2.0.0",
226 "status": "running",
227 "cluster_id": (manifest_processor.cluster_id if manifest_processor else "unknown"),
228 "region": (manifest_processor.region if manifest_processor else "unknown"),
229 "endpoints": {
230 "manifests": {
231 "submit": "POST /api/v1/manifests",
232 "validate": "POST /api/v1/manifests/validate",
233 "get": "GET /api/v1/manifests/{namespace}/{name}",
234 "delete": "DELETE /api/v1/manifests/{namespace}/{name}",
235 },
236 "jobs": {
237 "list": "GET /api/v1/jobs",
238 "get": "GET /api/v1/jobs/{namespace}/{name}",
239 "logs": "GET /api/v1/jobs/{namespace}/{name}/logs",
240 "events": "GET /api/v1/jobs/{namespace}/{name}/events",
241 "pods": "GET /api/v1/jobs/{namespace}/{name}/pods",
242 "metrics": "GET /api/v1/jobs/{namespace}/{name}/metrics",
243 "delete": "DELETE /api/v1/jobs/{namespace}/{name}",
244 "bulk_delete": "DELETE /api/v1/jobs",
245 "retry": "POST /api/v1/jobs/{namespace}/{name}/retry",
246 },
247 "templates": {
248 "list": "GET /api/v1/templates",
249 "create": "POST /api/v1/templates",
250 "get": "GET /api/v1/templates/{name}",
251 "delete": "DELETE /api/v1/templates/{name}",
252 "create_job": "POST /api/v1/jobs/from-template/{name}",
253 },
254 "webhooks": {
255 "list": "GET /api/v1/webhooks",
256 "create": "POST /api/v1/webhooks",
257 "delete": "DELETE /api/v1/webhooks/{id}",
258 },
259 "cost": {
260 "status": "GET /api/v1/cost/status",
261 "reports": "GET /api/v1/cost/reports",
262 "generate_report": "POST /api/v1/cost/reports",
263 },
264 "health": "GET /api/v1/health",
265 "status": "GET /api/v1/status",
266 },
267 }
270@app.get("/healthz", tags=["Health"])
271async def kubernetes_health_check() -> dict[str, str]:
272 """Kubernetes-style liveness probe."""
273 return {"status": "ok"}
276@app.get("/readyz", tags=["Health"])
277async def kubernetes_readiness_check() -> dict[str, str]:
278 """Kubernetes readiness includes the enabled queue worker task."""
279 if manifest_processor is None:
280 raise HTTPException(status_code=503, detail="Manifest processor not ready")
281 worker_task = getattr(app.state, "central_queue_worker_task", None)
282 if worker_task is not None and worker_task.done(): 282 ↛ 283line 282 didn't jump to line 283 because the condition on line 282 was never true
283 raise HTTPException(status_code=503, detail="Central queue worker stopped unexpectedly")
284 return {"status": "ready"}
287@app.get("/api/v1/health", tags=["Health"])
288async def health_check() -> JSONResponse:
289 """Health check endpoint for load balancer health checks."""
290 try:
291 if manifest_processor is None:
292 return JSONResponse(
293 status_code=503,
294 content={
295 "status": "unhealthy",
296 "timestamp": datetime.now(UTC).isoformat(),
297 "message": "Manifest processor not initialized",
298 },
299 )
301 try:
302 manifest_processor.core_v1.list_namespace(limit=1)
303 api_healthy = True
304 except Exception as e:
305 logger.error(f"Kubernetes API health check failed: {e}")
306 api_healthy = False
308 status_code = 200 if api_healthy else 503
309 return JSONResponse(
310 status_code=status_code,
311 content={
312 "status": "healthy" if api_healthy else "unhealthy",
313 "timestamp": datetime.now(UTC).isoformat(),
314 "cluster_id": manifest_processor.cluster_id,
315 "region": manifest_processor.region,
316 "kubernetes_api": "connected" if api_healthy else "disconnected",
317 },
318 )
320 except Exception as e:
321 logger.error(f"Health check failed: {e}")
322 return JSONResponse(
323 status_code=503,
324 content={
325 "status": "unhealthy",
326 "timestamp": datetime.now(UTC).isoformat(),
327 "error": "manifest processor unavailable",
328 },
329 )
332@app.get("/api/v1/status", tags=["Health"])
333async def get_service_status() -> dict[str, Any]:
334 """Service status endpoint with detailed information."""
335 templates_count = 0
336 webhooks_count = 0
337 try:
338 if template_store: 338 ↛ 340line 338 didn't jump to line 340 because the condition on line 338 was always true
339 templates_count = len(template_store.list_templates())
340 if webhook_store:
341 webhooks_count = len(webhook_store.list_webhooks())
342 except Exception as e:
343 logger.warning(f"Failed to get store counts: {e}")
345 status_info: dict[str, Any] = {
346 "service": "GCO Manifest Processor API",
347 "version": "2.0.0",
348 "timestamp": datetime.now(UTC).isoformat(),
349 "manifest_processor_initialized": manifest_processor is not None,
350 "environment": {
351 "cluster_name": os.getenv("CLUSTER_NAME", "unknown"),
352 "region": os.getenv("REGION", "unknown"),
353 "max_cpu_per_manifest": os.getenv("MAX_CPU_PER_MANIFEST", "10"),
354 "max_memory_per_manifest": os.getenv("MAX_MEMORY_PER_MANIFEST", "32Gi"),
355 "max_gpu_per_manifest": os.getenv("MAX_GPU_PER_MANIFEST", "4"),
356 "allowed_namespaces": os.getenv("ALLOWED_NAMESPACES", "gco-jobs"),
357 "validation_enabled": os.getenv("VALIDATION_ENABLED", "true"),
358 },
359 "templates_count": templates_count,
360 "webhooks_count": webhooks_count,
361 "central_queue_worker": (
362 worker.health()
363 if (worker := getattr(app.state, "central_queue_worker", None)) is not None
364 else {"enabled": False, "running": False}
365 ),
366 }
368 if manifest_processor: 368 ↛ 383line 368 didn't jump to line 383 because the condition on line 368 was always true
369 status_info.update(
370 {
371 "cluster_id": manifest_processor.cluster_id,
372 "region": manifest_processor.region,
373 "resource_limits": {
374 "max_cpu_millicores": manifest_processor.max_cpu_per_manifest,
375 "max_memory_bytes": manifest_processor.max_memory_per_manifest,
376 "max_gpu_count": manifest_processor.max_gpu_per_manifest,
377 },
378 "allowed_namespaces": list(manifest_processor.allowed_namespaces),
379 "validation_enabled": manifest_processor.validation_enabled,
380 }
381 )
383 return status_info
386# =============================================================================
387# Error Handlers
388# =============================================================================
391@app.exception_handler(Exception)
392async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
393 """Global exception handler for unhandled errors."""
394 logger.error(f"Unhandled exception in {request.method} {request.url}: {exc}")
395 return JSONResponse(
396 status_code=500,
397 content={
398 "error": "Internal server error",
399 "detail": str(exc) if os.getenv("DEBUG") else "An unexpected error occurred",
400 "timestamp": datetime.now(UTC).isoformat(),
401 },
402 )
405# =============================================================================
406# App Factory & Entrypoint
407# =============================================================================
410def create_app() -> FastAPI:
411 """Factory function to create the FastAPI app."""
412 return app
415if __name__ == "__main__":
416 import uvicorn
418 host = os.getenv("HOST", "0.0.0.0") # nosec B104 — must bind all interfaces inside K8s pod
419 port = int(os.getenv("PORT", "8080"))
420 log_level = os.getenv("LOG_LEVEL", "info").lower()
422 logger.info(f"Starting Manifest API on {host}:{port}")
424 uvicorn.run(
425 "gco.services.manifest_api:app",
426 host=host,
427 port=port,
428 log_level=log_level,
429 reload=False,
430 )