feat: add optional prometheus metrics endpoint.

Dockerfile

@@ -15,13 +15,13 @@ COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 RUN --mount=type=cache,target=/root/.cache/uv \
   --mount=type=bind,source=uv.lock,target=uv.lock \
   --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
-  uv sync --frozen --no-install-project --no-dev
+  uv sync --frozen --no-install-project --no-dev --extra metrics
 
 # Then, add the rest of the project source code and install it
 # Installing separately from its dependencies allows optimal layer caching
 ADD . /app
 RUN --mount=type=cache,target=/root/.cache/uv \
-  uv sync --frozen --no-dev
+  uv sync --frozen --no-dev --extra metrics
 
 # Runtime stage
 FROM python:3.13-slim

docs/user-guide/configuration.md

@@ -160,9 +160,12 @@ The application is configurable via environment variables.
       "^/docs/oauth2-redirect": ["GET"],
       "^/healthz": ["GET"],
       "^/_mgmt/ping": ["GET"],
-      "^/_mgmt/health": ["GET"]
+      "^/_mgmt/health": ["GET"],
+      "^/_mgmt/metrics": ["GET"]
     }
     ```
+    > [!NOTE]
+    > `/_mgmt/metrics` is only served when the `metrics` extra is installed (`pip install stac-auth-proxy[metrics]`); otherwise it's automatically dropped from `PUBLIC_ENDPOINTS` at startup.
 
 ### `ENABLE_AUTHENTICATION_EXTENSION`
 

docs/user-guide/getting-started.md

@@ -53,6 +53,8 @@ docker run -p 8000:8000 \
    ```bash
    pip install stac-auth-proxy
    ```
+   > [!TIP]
+   > To expose a Prometheus `/_mgmt/metrics` endpoint, install the `metrics` extra instead: `pip install stac-auth-proxy[metrics]`.
 2. Set environment variables:
    ```bash
    export UPSTREAM_URL=https://your-stac-api.com/stac

pyproject.toml

@@ -43,6 +43,9 @@ docs = [
 lambda = [
     "mangum>=0.19.0",
 ]
+metrics = [
+    "prometheus-fastapi-instrumentator>=8.0.2",
+]
 
 [tool.coverage.run]
 branch = true
@@ -101,6 +104,7 @@ dev = [
   "pytest>=8.3.3",
   "ruff>=0.0.238",
   "starlette-cramjam>=0.4.0",
+  "prometheus-fastapi-instrumentator>=8.0.2",
   "types-simplejson",
   "types-attrs",
 ]

src/stac_auth_proxy/app.py

@@ -6,6 +6,7 @@
 """
 
 import logging
+import re
 from typing import Any, Optional
 
 import httpx
@@ -16,6 +17,7 @@
 from .config import Settings
 from .handlers import HealthzHandler, ReverseProxyHandler, SwaggerUI
 from .lifespan import build_lifespan
+from .metrics import METRICS_AVAILABLE, instrument_app
 from .middleware import (
     AddProcessTimeHeaderMiddleware,
     AuthenticationExtensionMiddleware,
@@ -85,6 +87,14 @@ def configure_app(
             prefix=settings.healthz_prefix,
         )
 
+    if METRICS_AVAILABLE and r"^/_mgmt/metrics" in settings.public_endpoints:
+        excluded_handlers = [r"/_mgmt/"]
+        if settings.healthz_prefix:
+            excluded_handlers.append(re.escape(settings.healthz_prefix))
+        instrument_app(app, excluded_handlers=excluded_handlers)
+    else:
+        settings.public_endpoints.pop(r"^/_mgmt/metrics", None)
+
     #
     # Middleware (order is important, last added = first to run)
     #

src/stac_auth_proxy/config.py

@@ -111,6 +111,7 @@ class Settings(BaseSettings):
         r"^/healthz": ["GET"],
         r"^/_mgmt/ping": ["GET"],
         r"^/_mgmt/health": ["GET"],
+        r"^/_mgmt/metrics": ["GET"],
     }
     private_endpoints: EndpointMethodsWithScope = {
         # https://github.com/stac-api-extensions/collection-transaction/blob/v1.0.0-beta.1/README.md#methods

src/stac_auth_proxy/metrics.py

@@ -0,0 +1,94 @@
+"""Optional Prometheus metrics for STAC operations."""
+
+import re
+from typing import Any, Optional, Sequence
+
+METRICS_AVAILABLE = False
+
+OPERATIONS = [  # ordered; first match wins
+    (r"^/$", {"GET": "landing"}),
+    (r"^/conformance$", {"GET": "conformance"}),
+    (r"^/search$", {"GET": "search", "POST": "search"}),
+    (r"^/collections$", {"GET": "list_collections", "POST": "create_collection"}),
+    (
+        r"^/collections/[^/]+$",
+        {
+            "GET": "get_collection",
+            "PUT": "edit_collection",
+            "PATCH": "edit_collection",
+            "DELETE": "delete_collection",
+        },
+    ),
+    (r"^/collections/[^/]+/items$", {"GET": "list_items", "POST": "create_item"}),
+    (
+        r"^/collections/[^/]+/items/[^/]+$",
+        {
+            "GET": "get_item",
+            "PUT": "edit_item",
+            "PATCH": "edit_item",
+            "DELETE": "delete_item",
+        },
+    ),
+    (r"^/collections/[^/]+/bulk_items$", {"POST": "bulk"}),
+]
+_COMPILED = [(re.compile(p), m) for p, m in OPERATIONS]
+
+
+def classify_operation(method: str, path: str) -> str:
+    """Map a request to a low-cardinality STAC operation name."""
+    for pattern, methods in _COMPILED:
+        if pattern.match(path):
+            return methods.get(method.upper(), "unknown")
+    return "unknown"
+
+
+def instrument_app(
+    app: Any,
+    excluded_handlers: Optional[Sequence[str]] = None,
+) -> None:
+    """Instrument a FastAPI app and expose Prometheus metrics when available."""
+    if not METRICS_AVAILABLE:
+        return
+    _instrument_app(app, list(excluded_handlers or []))
+
+
+try:
+    from prometheus_client import Counter, Histogram
+    from prometheus_fastapi_instrumentator import Instrumentator
+    from prometheus_fastapi_instrumentator.metrics import Info
+
+    REQUESTS = Counter(
+        "http_requests_total",
+        "Total HTTP requests by STAC operation.",
+        labelnames=("operation", "method", "status"),
+    )
+    LATENCY = Histogram(
+        "http_request_duration_seconds",
+        "HTTP request latency by STAC operation.",
+        labelnames=("operation", "method"),
+        buckets=(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, float("inf")),
+    )
+
+    def record_stac_metrics(info: Info) -> None:
+        """Record request count and latency using STAC operation labels."""
+        operation = classify_operation(info.request.method, info.request.url.path)
+        REQUESTS.labels(operation, info.method, info.modified_status).inc()
+        LATENCY.labels(operation, info.method).observe(info.modified_duration)
+
+    def _instrument_app(app: Any, excluded_handlers: list[str]) -> None:
+        (
+            Instrumentator(
+                should_group_status_codes=True,
+                excluded_handlers=excluded_handlers,
+            )
+            .add(record_stac_metrics)
+            .instrument(app)
+            .expose(app, endpoint="/_mgmt/metrics", include_in_schema=False)
+        )
+
+    METRICS_AVAILABLE = True
+
+except ImportError:
+
+    def _instrument_app(app: Any, excluded_handlers: list[str]) -> None:
+        return

tests/test_configure_app.py

@@ -1,8 +1,12 @@
 """Tests for configuring an external FastAPI application."""
 
+import pytest
 from fastapi import APIRouter, FastAPI
+from fastapi.testclient import TestClient
 
+import stac_auth_proxy.app as app_module
 from stac_auth_proxy import Settings, configure_app
+from stac_auth_proxy.metrics import classify_operation
 
 
 def get_flattened_routes(app: FastAPI | APIRouter, prefix=""):
@@ -65,3 +69,76 @@ def test_configure_app_excludes_proxy_route():
     routes = get_flattened_routes(app)
     assert settings.healthz_prefix in routes
     assert "/{path:path}" not in routes
+
+
+def test_metrics_endpoint_returns_prometheus_output():
+    """Metrics returns Prometheus exposition format when enabled in PUBLIC_ENDPOINTS."""
+    app = FastAPI()
+    settings = Settings(
+        upstream_url="https://example.com",
+        oidc_discovery_url="https://example.com/.well-known/openid-configuration",
+        wait_for_upstream=False,
+        check_conformance=False,
+        default_public=True,
+    )
+
+    configure_app(app, settings)
+    app.add_api_route("/collections", lambda: {"collections": []}, methods=["GET"])
+    client = TestClient(app)
+    assert client.get("/collections").status_code == 200
+    response = client.get("/_mgmt/metrics")
+
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/plain")
+    assert "# HELP" in response.text
+    assert (
+        'http_requests_total{method="GET",operation="list_collections",status="2xx"}'
+        in response.text
+    )
+    assert (
+        'http_request_duration_seconds_count{method="GET",operation="list_collections"}'
+        in response.text
+    )
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "expected"),
+    [
+        ("GET", "/", "landing"),
+        ("GET", "/conformance", "conformance"),
+        ("GET", "/search", "search"),
+        ("POST", "/search", "search"),
+        ("GET", "/collections", "list_collections"),
+        ("POST", "/collections", "create_collection"),
+        ("GET", "/collections/sentinel-2", "get_collection"),
+        ("PUT", "/collections/sentinel-2", "edit_collection"),
+        ("DELETE", "/collections/sentinel-2", "delete_collection"),
+        ("GET", "/collections/sentinel-2/items", "list_items"),
+        ("POST", "/collections/sentinel-2/items", "create_item"),
+        ("GET", "/collections/sentinel-2/items/abc", "get_item"),
+        ("DELETE", "/collections/sentinel-2/items/abc", "delete_item"),
+        ("POST", "/collections/sentinel-2/bulk_items", "bulk"),
+        ("GET", "/unknown", "unknown"),
+        ("POST", "/conformance", "unknown"),
+    ],
+)
+def test_classify_operation(method, path, expected):
+    """STAC paths map to low-cardinality operation names."""
+    assert classify_operation(method, path) == expected
+
+
+def test_metrics_endpoint_skipped_without_instrumentator(monkeypatch):
+    """Metrics route and public endpoint are omitted when the extra is absent."""
+    monkeypatch.setattr(app_module, "METRICS_AVAILABLE", False)
+    app = FastAPI()
+    settings = Settings(
+        upstream_url="https://example.com",
+        oidc_discovery_url="https://example.com/.well-known/openid-configuration",
+        wait_for_upstream=False,
+        check_conformance=False,
+    )
+
+    configure_app(app, settings)
+
+    assert "/_mgmt/metrics" not in get_flattened_routes(app)
+    assert r"^/_mgmt/metrics" not in settings.public_endpoints