[Observability][API server][Rust frontend] Count HTTP requests refused before they reach the API server

docs/serving.rst

@@ -522,7 +522,8 @@ warmup requests. Every family has a ``model_name`` label (the config's ``model``
   request failed in the engine, preprocessing or result delivery), ``abort`` (the client
   went away, or its finished result was dropped unread) or ``timeout`` (``--timeout``
   expired). Requests refused before they were submitted (a 400 for a bad form, a 409
-  for a ``request_id`` in flight) are not counted.
+  for a ``request_id`` in flight) are not counted here but in
+  ``mstar_http_refusals_total``.
 - ``mstar_requests_in_flight{model_name}``: gauge, requests the API server holds
   (submitted and not yet returned, failed or aborted).
 - ``mstar_e2e_request_latency_seconds{model_name, output_modalities}``: histogram over
@@ -532,6 +533,18 @@ warmup requests. Every family has a ``model_name`` label (the config's ``model``
   successful requests, from submission to the first result chunk of any modality
   reaching the API server. It is measured on the server, so it does not include the
   network to the client, and a non-streaming client sees nothing until the end.
+- ``mstar_http_refusals_total{model_name, route, status}``: counter, HTTP requests the
+  frontend answered with a 4xx status without submitting them to the API server: ``401``
+  (missing or wrong API key), ``413`` (body over ``MSTAR_MAX_BODY_MB``), ``422`` (body
+  that fails validation), ``400`` (a bad form field or request), ``409`` (a
+  ``request_id`` in flight), ``404`` (unknown route, or a model without the endpoint),
+  ``405`` (wrong method), plus ``503`` when the Rust frontend's
+  ``MSTAR_MAX_CONCURRENT_REQUESTS`` cap turns a request away. ``route`` is the route's
+  path as declared (``/v1/chat/completions``), ``unmatched`` when no route matches.
+  A request that was submitted and then failed counts once, as an ``error`` in
+  ``mstar_requests_total``, never here. WebSocket handshakes are not counted. With
+  ``--rust-frontend``, ``mstar-server`` keeps these counts and sends them with each
+  ``/metrics`` request, so the API server exports them under the same name.
 
 Where a successful request's time went, from the timestamps every request already
 carries (same ``model_name`` and ``output_modalities`` labels; the conductor's

mstar/api_server/entrypoint.py

@@ -40,7 +40,13 @@
 from mstar.communication.communicator import CommProtocol, make_communicator
 from mstar.model.multimodal import PromptPart
 from mstar.model.registry import model_init_kwargs
-from mstar.observability.metrics import INSTALL_HINT, MISSING_DEPENDENCY, ServingMetrics
+from mstar.observability.metrics import (
+    INSTALL_HINT,
+    MISSING_DEPENDENCY,
+    RefusalMiddleware,
+    ServingMetrics,
+    mark_submitted,
+)
 from mstar.profile.display import pretty_print_profile
 from mstar.profile.format import OutputInfo, RequestProfile, RequestTiming
 from mstar.utils import profiler
@@ -567,6 +573,8 @@ def submit_request(
                 ),
                 request_id=request_id,
             )
+        if self.metrics is not None:
+            mark_submitted()
 
         self.preprocess_worker.new_request(PreprocessInput(
             request_id=key,
@@ -1394,6 +1402,14 @@ async def _one(request: WarmupRequest, request_id: str) -> None:
 # is read.
 api_key_guard = APIKeyGuard(resolve_api_key())
 app.add_middleware(APIKeyMiddleware, guard=api_key_guard)
+# Outside the key check and the body limit, which refuse before routing, and
+# inside CORS, so a browser preflight is not counted. ``routes`` must list
+# every included router's routes (test_serving_metrics checks each route).
+app.add_middleware(
+    RefusalMiddleware,
+    metrics=lambda: getattr(api_server, "metrics", None),
+    routes=lambda: (*app.router.routes, *openai_router.routes),
+)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],

mstar/api_server/rust_frontend.py

@@ -10,15 +10,19 @@
     frontend -> bridge:  {t:"submit", rid, sub, text, file_paths,
                           input_modalities, output_modalities, model_kwargs,
                           prompt_parts, streaming} | {t:"abort", rid, sub}
-                       | {t:"ping", rid, sub} | {t:"metrics", rid, sub}
+                       | {t:"ping", rid, sub}
+                       | {t:"metrics", rid, sub, refusals:{route, status, count}}
     bridge -> frontend:  {t:"chunk", rid, sub, modality, data(bin), metadata}
                        | {t:"err", rid, sub, msg} | {t:"done", rid, sub}
                        | {t:"pong", rid, sub}
                        | {t:"metrics", rid, sub, data(bin), metadata}
 
-A ``metrics`` reply is what ``GET /metrics`` serves on the Python frontend:
-the text rendered from ``APIServer.metrics`` with its ``content_type`` in
-``metadata``, or empty ``metadata`` when metrics are off (a 404).
+A ``metrics`` request carries the frontend's cumulative refusal counts as
+three parallel lists (route template, status, count), which the API server
+exports as ``mstar_http_refusals_total``. The reply is what ``GET /metrics``
+serves on the Python frontend: the text rendered from ``APIServer.metrics``
+with its ``content_type`` in ``metadata``, or empty ``metadata`` when metrics
+are off (a 404; the counts are then ignored).
 
 ``sub`` names one submission and is never reused. A client may resubmit its
 ``rid`` as soon as it drops a stream, and that stream's ``done`` or late
@@ -207,6 +211,10 @@ def _metrics(self, msg: dict) -> None:
         reply = {"t": "metrics", "rid": msg["rid"], "sub": msg.get("sub"),
                  "data": b"", "metadata": {}}
         if metrics is not None:
+            refusals = msg.get("refusals")
+            if refusals is not None:
+                metrics.refusals.update_rust(
+                    refusals["route"], refusals["status"], refusals["count"])
             reply["data"] = metrics.render()
             reply["metadata"] = {"content_type": metrics.content_type}
         self._send(reply)

mstar/observability/metrics.py

@@ -2,8 +2,11 @@
 
 Requests are counted where they leave :class:`~mstar.api_server.entrypoint.APIServer`,
 which ``/generate``, ``/generate/ws``, the OpenAI routes, the Rust-frontend
-bridge and the Dynamo bridge all go through. Each family carries the
-deployment's ``model_name``; every other label takes values from a fixed set.
+bridge and the Dynamo bridge all go through. Requests the HTTP frontend
+refuses before that (a missing API key, a body over the limit, a bad form)
+are counted by route and status in ``mstar_http_refusals_total``. Each family
+carries the deployment's ``model_name``; every other label takes values from
+a fixed set.
 
 Only the API server process imports ``prometheus_client`` and renders
 ``/metrics``. Other processes (data worker, conductor, workers) send plain
@@ -16,8 +19,10 @@
 
 from __future__ import annotations
 
+import threading
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
+from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any, NamedTuple
 
 try:
@@ -28,7 +33,7 @@
         Histogram,
         generate_latest,
     )
-    from prometheus_client.core import GaugeMetricFamily
+    from prometheus_client.core import CounterMetricFamily, GaugeMetricFamily
 except ModuleNotFoundError as e:
     # Only a missing package means "not installed"; a broken install must
     # still raise.
@@ -109,6 +114,139 @@ class _RequestSeries(NamedTuple):
     delivery: Any
 
 
+class _Refusals:
+    """``mstar_http_refusals_total``: what the Python frontend counts here,
+    plus the latest cumulative counts the Rust frontend sends with each
+    ``/metrics`` request (``mstar-server`` keeps them; see ``rust/server/src/refusals.rs``).
+    Only one of the two frontends serves a deployment, so the sum is that
+    frontend's counts."""
+
+    def __init__(self, model_name: str):
+        self.model_name = model_name
+        self._lock = threading.Lock()
+        self._python: dict[tuple[str, str], int] = {}
+        self._rust: dict[tuple[str, str], int] = {}
+
+    def inc(self, route: str, status: int) -> None:
+        key = (route, str(status))
+        with self._lock:
+            self._python[key] = self._python.get(key, 0) + 1
+
+    def update_rust(self, routes: list[str], statuses: list[int], counts: list[int]) -> None:
+        """Take the Rust frontend's cumulative counts (three parallel lists).
+        Two scrapes can cross on their way here, so a count only moves up:
+        an older snapshot never lowers what a newer one set."""
+        with self._lock:
+            for route, status, count in zip(routes, statuses, counts, strict=True):
+                key = (str(route), str(status))
+                self._rust[key] = max(self._rust.get(key, 0), int(count))
+
+    def collect(self):
+        family = CounterMetricFamily(
+            "mstar_http_refusals",
+            "HTTP requests answered with an error before they reached the API server "
+            "(a 4xx, or the Rust frontend's admission cap 503), by route template and status.",
+            labels=["model_name", "route", "status"],
+        )
+        with self._lock:
+            counts = dict(self._python)
+            for key, n in self._rust.items():
+                counts[key] = counts.get(key, 0) + n
+        for (route, status), n in sorted(counts.items()):
+            family.add_metric([self.model_name, route, status], n)
+        yield family
+
+
+class _Request:
+    """One HTTP request seen by :class:`RefusalMiddleware`."""
+
+    __slots__ = ("submitted",)
+
+    def __init__(self):
+        self.submitted = False
+
+
+# A mutable holder rather than a bool: run_in_threadpool and new tasks run in
+# a copy of the context, which still holds the same object.
+_REQUEST: ContextVar[_Request | None] = ContextVar("mstar_http_request", default=None)
+
+
+def mark_submitted() -> None:
+    """The HTTP request being served reached ``APIServer``: it is counted in
+    ``mstar_requests_total`` when it ends, so it is never a refusal."""
+    request = _REQUEST.get()
+    if request is not None:
+        request.submitted = True
+
+
+def route_template(scope: dict, routes: Iterable) -> str:
+    """The route path the request matched (``/v1/chat/completions``), never
+    the raw URL; ``unmatched`` when no route matches. ``routes`` are the
+    app's routes, searched for a request refused before routing (the API
+    key, a declared body size) the way the router matches: a full match
+    before a method-only one."""
+    route = scope.get("route")
+    if route is None:
+        from starlette.routing import Match
+
+        partial = None
+        for candidate in routes:
+            # FastAPI 0.150+ lists an included router as one entry with no
+            # path, whose match names no route; its routes are listed too.
+            if getattr(candidate, "path", None) is None:
+                continue
+            match, _ = candidate.matches(scope)
+            if match is Match.FULL:
+                route = candidate
+                break
+            if match is Match.PARTIAL and partial is None:
+                partial = candidate
+        route = route or partial
+    return getattr(route, "path", None) or "unmatched"
+
+
+class RefusalMiddleware:
+    """Counts ``mstar_http_refusals_total`` for the Python frontend: every
+    HTTP response with a 4xx status for a request that never reached
+    ``APIServer.submit_request`` (:func:`mark_submitted`). A request that was
+    submitted and then failed is counted by ``mstar_requests_total`` instead.
+
+    Sits outside the API key and body limit middlewares (their 401 and 413 are
+    sent before routing) and inside CORS (a preflight is not a request).
+    WebSocket handshakes are not counted. With metrics off it only checks
+    ``metrics()`` for None.
+    """
+
+    def __init__(
+        self, app, metrics: Callable[[], ServingMetrics | None], routes: Callable[[], Iterable],
+    ):
+        self.app = app
+        self.metrics = metrics
+        self.routes = routes
+
+    async def __call__(self, scope, receive, send):
+        metrics = self.metrics() if scope["type"] == "http" else None
+        if metrics is None:
+            await self.app(scope, receive, send)
+            return
+        request = _Request()
+        status = 0
+
+        async def send_and_note_status(message):
+            nonlocal status
+            if message["type"] == "http.response.start":
+                status = message["status"]
+            await send(message)
+
+        token = _REQUEST.set(request)
+        try:
+            await self.app(scope, receive, send_and_note_status)
+        finally:
+            _REQUEST.reset(token)
+            if 400 <= status < 500 and not request.submitted:
+                metrics.refusals.inc(route_template(scope, self.routes()), status)
+
+
 class ServingMetrics:
     """The API server's registry and its request metrics."""
 
@@ -139,6 +277,8 @@ def __init__(self, model_name: str, in_flight: Callable[[], int]):
             registry=self.registry,
         )
         self.registry.register(_InFlight(model_name, in_flight))
+        self.refusals = _Refusals(model_name)
+        self.registry.register(self.refusals)
 
         # Where a successful request's time went, from the timestamps it
         # carries. The conductor's perf_counter is compared with this

rust/server/src/bridge.rs

@@ -16,15 +16,18 @@
 //!       input_modalities, output_modalities, model_kwargs, prompt_parts,
 //!       streaming}, `prompt_parts` present for a chat only
 //!                        | {t:"abort", rid, sub}
-//!                        | {t:"ping" | "metrics", rid, sub}
+//!                        | {t:"ping", rid, sub}
+//!                        | {t:"metrics", rid, sub, refusals:{route, status, count}}
 //!   conductor -> frontend: {t:"chunk", rid, sub, modality, data(bin), metadata}
 //!                        | {t:"done",  rid, sub}
 //!                        | {t:"pong",  rid, sub}
 //!                        | {t:"metrics", rid, sub, data(bin), metadata}
 //!
-//! A `metrics` reply carries the API server's rendered Prometheus text in
-//! `data` and its `content_type` in `metadata`; `metadata` has no
-//! `content_type` when the server runs without `--enable-metrics`.
+//! A `metrics` request carries this frontend's cumulative refusal counts
+//! (`refusals.rs`) as three parallel lists, for the API server to export. Its
+//! reply carries the API server's rendered Prometheus text in `data` and its
+//! `content_type` in `metadata`; `metadata` has no `content_type` when the
+//! server runs without `--enable-metrics`.
 //!
 //! `sub` names one submission and is never reused; `rid` is the client's id,
 //! which a client may reuse as soon as it drops a stream (that stream's `done`
@@ -44,6 +47,7 @@ use serde_json::Value;
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
 use crate::adapters::{Part, SubmitArgs};
+use crate::refusals::Snapshot;
 
 /// One chunk of generated output, matching mstar's `ResultChunk`.
 #[derive(Debug, Clone)]
@@ -93,6 +97,16 @@ fn no_parts(parts: &&[Part]) -> bool {
     parts.is_empty()
 }
 
+/// A `ping` or `metrics` request; only `metrics` carries `refusals`.
+#[derive(Serialize)]
+struct ProbeMsg<'a> {
+    t: &'a str,
+    rid: &'a str,
+    sub: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    refusals: Option<&'a Snapshot>,
+}
+
 /// Inbound message; `t` selects chunk / err / done. `data` is msgpack binary.
 #[derive(Deserialize)]
 struct Inbound {
@@ -297,12 +311,17 @@ impl Bridge {
     /// Send `{t}` to the bridge loop and wait for its one reply (routed like
     /// a completed request). None on timeout: the bridge loop or the backend
     /// behind it is gone.
-    async fn round_trip(&self, t: &str, timeout: Duration) -> Option<StreamItem> {
+    async fn round_trip(
+        &self,
+        t: &str,
+        refusals: Option<&Snapshot>,
+        timeout: Duration,
+    ) -> Option<StreamItem> {
         let rid = format!("{t}-{}", uuid::Uuid::new_v4());
         let (tx, mut rx) = unbounded_channel();
         let sub = self.next_sub.fetch_add(1, Ordering::Relaxed);
         self.routes.lock().expect("routes lock").routes.insert(sub, Route { tx, rid: None });
-        let msg = serde_json::json!({"t": t, "rid": rid, "sub": sub});
+        let msg = ProbeMsg { t, rid: &rid, sub, refusals };
         if let Ok(payload) = rmp_serde::to_vec_named(&msg) {
             let _ = self.mbox.send("bridge", &payload);
         }
@@ -314,13 +333,14 @@ impl Bridge {
     /// Liveness round-trip for the deep health check: send `{t:"ping"}` and
     /// wait for the backend's pong. False on timeout.
     pub async fn ping(&self, timeout: Duration) -> bool {
-        self.round_trip("ping", timeout).await.is_some()
+        self.round_trip("ping", None, timeout).await.is_some()
     }
 
-    /// The API server's `/metrics` reply: the rendered text in `data` and,
-    /// when metrics are on, its `content_type` in `metadata`. None on timeout.
-    pub async fn metrics(&self, timeout: Duration) -> Option<ResultChunk> {
-        match self.round_trip("metrics", timeout).await? {
+    /// The API server's `/metrics` reply, asked with this frontend's refusal
+    /// counts: the rendered text in `data` and, when metrics are on, its
+    /// `content_type` in `metadata`. None on timeout.
+    pub async fn metrics(&self, refusals: &Snapshot, timeout: Duration) -> Option<ResultChunk> {
+        match self.round_trip("metrics", Some(refusals), timeout).await? {
             StreamItem::Chunk(chunk) => Some(chunk),
             _ => None,
         }
@@ -542,6 +562,23 @@ mod tests {
         assert!(routes.lock().unwrap().routes.is_empty(), "the reply left its route behind");
     }
 
+    #[test]
+    fn only_a_metrics_request_carries_the_refusal_counts() {
+        let encode = |msg: &ProbeMsg| -> Value {
+            rmp_serde::from_slice(&rmp_serde::to_vec_named(msg).unwrap()).unwrap()
+        };
+        let snap = Snapshot { route: vec!["/generate".into()], status: vec![401], count: vec![3] };
+        assert_eq!(
+            encode(&ProbeMsg { t: "metrics", rid: "m", sub: 4, refusals: Some(&snap) }),
+            json!({"t": "metrics", "rid": "m", "sub": 4,
+                   "refusals": {"route": ["/generate"], "status": [401], "count": [3]}})
+        );
+        assert_eq!(
+            encode(&ProbeMsg { t: "ping", rid: "p", sub: 5, refusals: None }),
+            json!({"t": "ping", "rid": "p", "sub": 5})
+        );
+    }
+
     #[test]
     fn one_caller_at_a_time_waits_out_a_live_rid_s_grace() {
         let routes: Routes = Mutex::new(Table::default());