orchestrata_ultimate.gateway.http_surface

Reference › orchestrata_ultimate › gateway › http_surface

http_surface builds the small operational HTTP application that runs beside the WebSocket API (SPEC-014 US-009). It serves JSON and Prometheus text only; the control API stays WebSocket JSON-RPC on port 8765, and this app listens on the separate HTTP port from config (GATEWAY_HTTP_PORT, bound to http_bind_host). server is the only importer.

build_gateway_http_app returns a FastAPI app with these routes:

• /health and its alias /api/v1/health answer a liveness alive status.
• /ready first asks the optional readiness probe whether the gateway has started and returns 503 with not_ready if not; otherwise it asks the postman.health HealthChecker for overall status and returns 503 unless that status is healthy or degraded.
• /metrics returns the default Prometheus registry; the instruments are defined in metrics.
• The detailed postman.health router is mounted as well.

A CORS middleware allows the origins named in GATEWAY_ALLOWED_ORIGINS (a single local UI origin when unset) for GET and OPTIONS. The _spec098_headers middleware echoes or mints an X-Request-ID and adds X-Content-Type-Options and X-Frame-Options headers (SPEC-098 FR-016 and FR-017). serve_http runs the app under uvicorn as an asyncio task with signal handlers disabled. See gateway.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/http_surface.py

"""Gateway HTTP surface — ``/health``, ``/api/v1/health``, ``/ready``, ``/metrics``.

The gateway is a WebSocket server; SPEC-014 FR-021 additionally requires a small
HTTP surface for liveness/readiness probes (k8s/load-balancer friendly) and
Prometheus metrics scraping. This module builds that FastAPI app and serves it
alongside the WS server. It reuses the existing ``postman.health`` HealthChecker
(real Redis/graph probes) so readiness reflects actual infrastructure health.

Spec: SPEC-014 (WebSocket Gateway) FR-021 — HTTP health/metrics endpoints.
"""

from __future__ import annotations

import contextlib
import os
import uuid
from collections.abc import Awaitable, Callable, Mapping
from typing import Any

from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from prometheus_client import CONTENT_TYPE_LATEST, generate_latest

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.postman.health import HealthChecker, create_health_router

# Returns True when the gateway itself is ready to serve (e.g. WS server up).
ReadinessProbe = Callable[[], Awaitable[bool]]


@traces(spec="014", story="US-009", reqs=["REQ-014-136", "REQ-014-140"])
def build_gateway_http_app(
    *,
    health_checker: HealthChecker | None = None,
    readiness_probe: ReadinessProbe | None = None,
    allowed_origins: Mapping[str, str] | None = None,
) -> FastAPI:
    """Build the gateway's HTTP app exposing ``/health``, ``/ready``, ``/metrics``.

    - ``/health`` — liveness: the process is up (always 200 while serving).
    - ``/ready`` — readiness: the gateway has started AND infrastructure
      (Redis/graph) is healthy/degraded; 503 otherwise.
    - ``/metrics`` — Prometheus exposition of the default registry.

    The detailed ``postman.health`` router (``/health/live``, ``/health/ready``,
    ``/health/checks``) is also mounted for richer probes.
    """
    app = FastAPI(title="Orchestrata Gateway", docs_url=None, redoc_url=None)
    checker = health_checker or HealthChecker()

    # SPEC-098 FR-022: CORS for allowed origins. The default is the single
    # origin the vizint-ui dev server runs on rather than "*", because "*"
    # plus credentials is rejected by browsers anyway and a wildcard here
    # would be a standing invitation for any page to drive the gateway.
    # ALLOWED_ORIGINS is read at call time (RUL-003) and is optional — an
    # unset value means the local UI only, not "everything".
    origins = [
        origin.strip()
        for origin in (
            (allowed_origins if allowed_origins is not None else os.environ).get(
                "GATEWAY_ALLOWED_ORIGINS", "http://localhost:3456"
            )
        ).split(",")
        if origin.strip()
    ]
    app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["GET", "OPTIONS"],
        allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
        expose_headers=["X-Request-ID"],
    )

    @app.middleware("http")
    async def _spec098_headers(
        request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        """SPEC-098 FR-016 security headers and FR-017 request tracing.

        The request id is echoed when the caller supplies one and minted when
        it does not, so a trace id always exists to correlate a gateway log
        line with the client that caused it — a caller that cannot see the id
        it was assigned cannot report it in a bug.
        """
        request_id = request.headers.get("X-Request-ID") or uuid.uuid4().hex
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        # FR-016: MIME-sniffing and framing. Nothing here is meant to be
        # embedded or interpreted as anything but what it says it is.
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        return response

    @app.get("/health")
    @traces(spec="014", story="US-009", reqs=["REQ-014-136"])
    async def health() -> dict[str, Any]:
        return {"status": "alive"}

    @app.get("/api/v1/health")
    async def versioned_health() -> dict[str, Any]:
        """SPEC-098 FR-018 — the versioned health path the spec names.

        An alias, not a second implementation: SPEC-014 FR-021 specifies
        ``/health`` and existing probes (start_test_services, the deploy
        units) use it, so both must answer. Divergence between two health
        endpoints is its own outage class.
        """
        return await health()

    @app.get("/ready")
    @traces(spec="014", story="US-009", reqs=["REQ-014-140"])
    async def ready(response: Response) -> dict[str, Any]:
        # Gateway-level readiness first — if the WS server has not started,
        # report not-ready without touching infrastructure.
        if readiness_probe is not None and not await readiness_probe():
            response.status_code = 503
            return {"status": "not_ready", "reason": "gateway not started"}
        status = await checker.get_overall_status()
        if status.get("status") not in ("healthy", "degraded"):
            response.status_code = 503
        return status

    @app.get("/metrics")
    @traces(spec="014", story="US-009", reqs=["REQ-014-141"])
    async def metrics() -> Response:
        return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

    app.include_router(create_health_router(checker))
    return app


async def serve_http(app: FastAPI, host: str, port: int) -> None:
    """Serve ``app`` with uvicorn until cancelled.

    Signal handlers are disabled so this can run as an asyncio task inside the
    gateway's event loop without clobbering the process signal handlers.
    """
    import uvicorn

    # http="h11": uvicorn's "auto" protocol loads httptools, whose C
    # extension does not declare free-threaded support — importing it
    # re-enables the GIL process-wide and breaks the janus transport
    # (SPEC-023). Same pin as scripts/ast_daemon.py.
    config = uvicorn.Config(
        app, host=host, port=port, log_level="warning", access_log=False, http="h11"
    )
    server = uvicorn.Server(config)
    # uvicorn >=0.29 removed install_signal_handlers — the old
    # `server.install_signal_handlers = lambda: None` override became a
    # silent no-op and uvicorn went back to capturing SIGINT/SIGTERM for the
    # whole serve duration. Signal capture now lives in
    # Server.capture_signals(); replace it with a no-op context manager so
    # serving as a task inside the gateway's loop never touches process
    # signal handlers.
    server.capture_signals = contextlib.nullcontext  # type: ignore[method-assign,assignment]  # noqa: E501
    await server.serve()