Count the requests the fleet gateway has open to each model

design/telemetry.md

@@ -604,6 +604,7 @@ separate those from a slow model.
 | `modelplane_frontend_request_duration_seconds` | histogram | fleet gateway |
 | `modelplane_frontend_ttft_seconds` | histogram | fleet gateway, streaming only |
 | `modelplane_frontend_tpot_seconds` | histogram | fleet gateway |
+| `modelplane_frontend_requests_active` | gauge | fleet gateway, per service |
 | `modelplane_requests_total{status}` | counter | fleet gateway |
 | `modelplane_tokens_total{direction}` | counter | fleet gateway |
 | `modelplane_requests_throttled_total` | counter | Envoy rate limits |

docs/content/platform/telemetry.md

@@ -33,6 +33,8 @@ The gateway's `modelplane_frontend_*` series have no `deployment` label. They
 carry the model instead: `gen_ai_original_model` is the model the caller sent,
 `<namespace>/<service>`, and `gen_ai_request_model` is the deployment the gateway
 sent the request to, `<namespace>/<deployment>`.
+`modelplane_frontend_requests_active` is counted per `ModelService` instead. It
+has a `service` label, and the same `namespace` label as an engine's series.
 
 Each replica publishes its own series, so combine them in your query. Which
 combination is right follows from what the metric measures:
@@ -65,6 +67,7 @@ Some examples of the available metrics:
 | `modelplane_frontend_ttft_seconds` | Time to the first token, measured at the gateway |
 | `modelplane_frontend_tpot_seconds` | Time per output token, measured at the gateway |
 | `modelplane_frontend_request_duration_seconds` | What the caller waited, end to end |
+| `modelplane_frontend_requests_active` | Requests open at the gateway, per `ModelService` |
 | `modelplane_request_queue_seconds` | How long a request waited before the engine started |
 | `modelplane_requests_waiting` | Queue depth per engine |
 | `modelplane_kv_cache_utilization_ratio` | KV-cache occupancy, averaged over replicas |
@@ -88,6 +91,20 @@ high maximum beside `modelplane_requests_preempted_total` climbing is one replic
 max by (deployment) (modelplane_kv_cache_utilization_ratio) > 0.95
 ```
 
+An engine works on a request until it finishes or is told to stop. When a caller
+disconnects and the engine isn't told, the engine's running and waiting counts don't
+drop while the gateway's count of open requests does, and new requests queue behind
+output nobody reads. To compare the two for a deployment `qwen3-8b` that a
+`ModelService` of the same name fronts, and that serves each request on one engine:
+
+```promql
+sum by (namespace) (modelplane_requests_running{deployment="qwen3-8b"} + modelplane_requests_waiting{deployment="qwen3-8b"})
+  - sum by (namespace) (modelplane_frontend_requests_active{service="qwen3-8b"})
+```
+
+If the difference is well above 0 for minutes, the engine is doing work no caller
+is waiting for.
+
 ## Send telemetry to a destination
 
 Create a `TelemetryDestination` for your OpenTelemetry-compatible endpoint:

functions/compose-serving-stack/function/collector.py

@@ -56,6 +56,21 @@
 # carries Envoy's own statistics and none of these.
 _GENAI_PORT = "aigw-admin"
 
+# Requests Envoy has sent upstream and not yet finished, per cluster it built.
+# Only the fleet's front door counts what callers are waiting on:
+# compose-inference-gateway's Gateway, on whichever cluster runs it. Each
+# inference cluster also runs a cluster-gateway the fleet gateway forwards to,
+# which counts the same requests a second time, so its copy is dropped. So are
+# the clusters that aren't a model: Envoy's own, and routes such as Grafana's.
+_IN_FLIGHT = "envoy_cluster_upstream_rq_active"
+_FLEET_GATEWAY = "inference-gateway"
+
+# The cluster Envoy builds for a ModelService: compose-model-route's
+# AIGatewayRoute, named for the service in the team's mirrored namespace, and
+# rule 0, the one rule compose_route gives it. The namespace is the one an
+# engine's series carry.
+_MODEL_ROUTE = r"httproute/(mp-[^/]+)/([^/]+)/rule/0"
+
 # What a series is attributed to, and the only resource attributes that survive
 # to the exporter. The merge across replicas groups on exactly these, so a
 # series differing in nothing else is one series.
@@ -171,6 +186,42 @@ def _annotated_port() -> list[dict[str, Any]]:
     ]
 
 
+def _in_flight_relabels() -> list[dict[str, Any]]:
+    """Keep the fleet gateway's in-flight requests to each model, as namespace and service.
+
+    Prometheus relabeling rather than a transform statement, because it runs
+    as the series are scraped, and the two drops are on labels a mapping can't
+    select on. A regex here is anchored at both ends.
+    """
+    return [
+        {
+            "source_labels": ["__name__", "envoy_cluster_name", "gateway"],
+            "regex": f"{_IN_FLIGHT};{_MODEL_ROUTE};{_FLEET_GATEWAY}",
+            "target_label": "__tmp_model",
+            "replacement": "true",
+        },
+        {"source_labels": ["__name__", "__tmp_model"], "regex": f"{_IN_FLIGHT};", "action": "drop"},
+        {
+            "source_labels": ["__name__", "envoy_cluster_name"],
+            "regex": f"{_IN_FLIGHT};{_MODEL_ROUTE}",
+            "target_label": "namespace",
+            "replacement": "$1",
+        },
+        {
+            "source_labels": ["__name__", "envoy_cluster_name"],
+            "regex": f"{_IN_FLIGHT};{_MODEL_ROUTE}",
+            "target_label": "service",
+            "replacement": "$2",
+        },
+        # An empty label is no label, so these leave the series with only the
+        # identity just set.
+        *(
+            {"source_labels": ["__name__"], "regex": _IN_FLIGHT, "target_label": label, "replacement": ""}
+            for label in ("envoy_cluster_name", "gateway", "__tmp_model")
+        ),
+    ]
+
+
 def _scrape_configs() -> list[dict[str, Any]]:
     """What to scrape on an inference cluster.
 
@@ -221,7 +272,12 @@ def _scrape_configs() -> list[dict[str, Any]]:
                 # front door 404s every interval and the modelplane_frontend_*
                 # series - the ones an SLO is written against - never arrive.
                 *_annotated_path(),
+                {
+                    "source_labels": ["__meta_kubernetes_pod_label_gateway_envoyproxy_io_owning_gateway_name"],
+                    "target_label": "gateway",
+                },
             ],
+            "metric_relabel_configs": _in_flight_relabels(),
         },
         {
             # The GenAI metrics, which are the SLO ones: what a caller waited,

functions/compose-serving-stack/function/stacks/metrics.py

@@ -40,6 +40,14 @@
     "gen_ai_server_time_per_output_token_seconds": "modelplane_frontend_tpot_seconds",
 }
 
+# The front door's own count of requests in flight to each model, which the
+# collector keeps for the fleet gateway's model routes only (see
+# collector._IN_FLIGHT). The engines' running and waiting requests well above
+# it, for minutes, is work no caller is waiting for.
+_GATEWAY_PROXY = {
+    "envoy_cluster_upstream_rq_active": "modelplane_frontend_requests_active",
+}
+
 # The engines, which explain what the gateway measured.
 _VLLM = {
     "vllm:time_to_first_token_seconds": "modelplane_request_ttft_seconds",
@@ -130,6 +138,7 @@ def mappings() -> list[v1alpha1.MetricMapping]:
     """
     return [
         _mapping("modelplane-gateway", _GATEWAY),
+        _mapping("modelplane-gateway-proxy", _GATEWAY_PROXY),
         _mapping("modelplane-vllm", _VLLM),
         _mapping("modelplane-sglang", _SGLANG),
         _mapping("modelplane-picker", _PICKER),

functions/compose-serving-stack/tests/test_collector.py

@@ -56,6 +56,19 @@ class PortRewriteCase:
     want: str
 
 
+@dataclasses.dataclass
+class InFlightCase:
+    """A test case for which gateway series of requests in flight the gateway job keeps."""
+
+    name: str
+    reason: str
+    cluster: str
+    gateway: str
+    # The namespace and service the fleet model rule captures, or None if it
+    # doesn't match, which the next rule drops.
+    want: tuple[str, str] | None
+
+
 @dataclasses.dataclass
 class StatementsCase:
     """A test case for collector.statements."""
@@ -218,6 +231,55 @@ def _config(
                                     "target_label": "__metrics_path__",
                                     "regex": "(.+)",
                                 },
+                                {
+                                    "source_labels": [
+                                        "__meta_kubernetes_pod_label_gateway_envoyproxy_io_owning_gateway_name"
+                                    ],
+                                    "target_label": "gateway",
+                                },
+                            ],
+                            "metric_relabel_configs": [
+                                {
+                                    "source_labels": ["__name__", "envoy_cluster_name", "gateway"],
+                                    "regex": "envoy_cluster_upstream_rq_active;httproute/(mp-[^/]+)/([^/]+)/rule/0;inference-gateway",
+                                    "target_label": "__tmp_model",
+                                    "replacement": "true",
+                                },
+                                {
+                                    "source_labels": ["__name__", "__tmp_model"],
+                                    "regex": "envoy_cluster_upstream_rq_active;",
+                                    "action": "drop",
+                                },
+                                {
+                                    "source_labels": ["__name__", "envoy_cluster_name"],
+                                    "regex": "envoy_cluster_upstream_rq_active;httproute/(mp-[^/]+)/([^/]+)/rule/0",
+                                    "target_label": "namespace",
+                                    "replacement": "$1",
+                                },
+                                {
+                                    "source_labels": ["__name__", "envoy_cluster_name"],
+                                    "regex": "envoy_cluster_upstream_rq_active;httproute/(mp-[^/]+)/([^/]+)/rule/0",
+                                    "target_label": "service",
+                                    "replacement": "$2",
+                                },
+                                {
+                                    "source_labels": ["__name__"],
+                                    "regex": "envoy_cluster_upstream_rq_active",
+                                    "target_label": "envoy_cluster_name",
+                                    "replacement": "",
+                                },
+                                {
+                                    "source_labels": ["__name__"],
+                                    "regex": "envoy_cluster_upstream_rq_active",
+                                    "target_label": "gateway",
+                                    "replacement": "",
+                                },
+                                {
+                                    "source_labels": ["__name__"],
+                                    "regex": "envoy_cluster_upstream_rq_active",
+                                    "target_label": "__tmp_model",
+                                    "replacement": "",
+                                },
                             ],
                         },
                         # A target of the gateway's own: its GenAI metrics are on
@@ -366,6 +428,7 @@ def _config(
                             'set(metric.name, "modelplane_frontend_request_duration_seconds") where metric.name == "gen_ai_server_request_duration_seconds"',
                             'set(metric.name, "modelplane_frontend_ttft_seconds") where metric.name == "gen_ai_server_time_to_first_token_seconds"',
                             'set(metric.name, "modelplane_frontend_tpot_seconds") where metric.name == "gen_ai_server_time_per_output_token_seconds"',
+                            'set(metric.name, "modelplane_frontend_requests_active") where metric.name == "envoy_cluster_upstream_rq_active"',
                             'set(metric.name, "modelplane_request_ttft_seconds") where metric.name == "vllm:time_to_first_token_seconds"',
                             'set(metric.name, "modelplane_request_duration_seconds") where metric.name == "vllm:e2e_request_latency_seconds"',
                             'set(metric.name, "modelplane_request_queue_seconds") where metric.name == "vllm:request_queue_time_seconds"',
@@ -687,6 +750,63 @@ def test_port_rewrite(case: PortRewriteCase) -> None:
     assert got == case.want, case.reason
 
 
+IN_FLIGHT_CASES = [
+    InFlightCase(
+        name="FleetModel",
+        reason="The fleet gateway's cluster for a ModelService's route matches, capturing its namespace and service.",
+        cluster="httproute/mp-ml-team-51733/qwen/rule/0",
+        gateway="inference-gateway",
+        want=("mp-ml-team-51733", "qwen"),
+    ),
+    InFlightCase(
+        name="ClusterGateway",
+        reason="A cluster gateway's copy of the same requests doesn't match, so a sum counts each request once.",
+        cluster="httproute/mp-ml-team-51733/qwen-34d80/rule/0",
+        gateway="cluster-gateway",
+        want=None,
+    ),
+    InFlightCase(
+        name="NotAModel",
+        reason="A route in modelplane-system, such as Grafana's, isn't a model and doesn't match.",
+        cluster="httproute/modelplane-system/grafana/rule/0",
+        gateway="inference-gateway",
+        want=None,
+    ),
+    InFlightCase(
+        name="EnvoyCluster",
+        reason="A cluster Envoy builds for itself doesn't match.",
+        cluster="xds_cluster",
+        gateway="inference-gateway",
+        want=None,
+    ),
+]
+
+
+@pytest.mark.parametrize("case", IN_FLIGHT_CASES, ids=lambda case: case.name)
+def test_in_flight(case: InFlightCase) -> None:
+    """The gateway job's fleet model rule matches the fleet gateway's requests in flight to each model, and no others."""
+    # Like test_port_rewrite, this checks what the rule does to a series, which
+    # comparing the config can't, so it picks the rule out of a rendered config.
+    config = yaml.safe_load(
+        collector.config(
+            "prod-us-east",
+            list(stacks.BUILTIN_MAPPINGS),
+            [_sink(name="primary", type_="otlphttp", secret=None)],
+            {"oauth2client/acme": {"token_url": "https://issuer.acme.example/token"}},
+        )
+    )
+    rule = next(
+        r
+        for j in config["receivers"]["prometheus"]["config"]["scrape_configs"]
+        if j["job_name"] == "modelplane-gateway"
+        for r in j["metric_relabel_configs"]
+        if r.get("target_label") == "__tmp_model" and r.get("replacement") == "true"
+    )
+    matched = re.fullmatch(rule["regex"], f"envoy_cluster_upstream_rq_active;{case.cluster};{case.gateway}")
+    got = (matched[1], matched[2]) if matched else None
+    assert got == case.want, case.reason
+
+
 STATEMENTS_CASES = [
     # The built-in mappings, read from the stacks package rather than restated,
     # because what the collector compiles them to is what this case pins.
@@ -705,6 +825,7 @@ def test_port_rewrite(case: PortRewriteCase) -> None:
                 'set(metric.name, "modelplane_frontend_request_duration_seconds") where metric.name == "gen_ai_server_request_duration_seconds"',
                 'set(metric.name, "modelplane_frontend_ttft_seconds") where metric.name == "gen_ai_server_time_to_first_token_seconds"',
                 'set(metric.name, "modelplane_frontend_tpot_seconds") where metric.name == "gen_ai_server_time_per_output_token_seconds"',
+                'set(metric.name, "modelplane_frontend_requests_active") where metric.name == "envoy_cluster_upstream_rq_active"',
                 'set(metric.name, "modelplane_request_ttft_seconds") where metric.name == "vllm:time_to_first_token_seconds"',
                 'set(metric.name, "modelplane_request_duration_seconds") where metric.name == "vllm:e2e_request_latency_seconds"',
                 'set(metric.name, "modelplane_request_queue_seconds") where metric.name == "vllm:request_queue_time_seconds"',
@@ -937,7 +1058,7 @@ def test_transform(case: TransformCase) -> None:
             (
                 "collector",
                 _deployment(
-                    config_hash="55bc4445ed790fa2",
+                    config_hash="8c64e59d01d1670e",
                     volumes=[{"name": "config", "configMap": {"name": "modelplane-collector"}}],
                     volume_mounts=[{"name": "config", "mountPath": "/conf"}],
                     env_from=None,
@@ -982,7 +1103,7 @@ def test_transform(case: TransformCase) -> None:
             (
                 "collector",
                 _deployment(
-                    config_hash="5b3619d55150aa79",
+                    config_hash="709ec5e0f4e73c4a",
                     volumes=[
                         {"name": "config", "configMap": {"name": "modelplane-collector"}},
                         {"name": "credentials-primary", "secret": {"secretName": "telemetry-credentials"}},
@@ -1049,7 +1170,7 @@ def test_transform(case: TransformCase) -> None:
             (
                 "collector",
                 _deployment(
-                    config_hash="5097b903785819b1",
+                    config_hash="9bce4f1ec98c43bd",
                     volumes=[
                         {"name": "config", "configMap": {"name": "modelplane-collector"}},
                         {"name": "credentials-vendor", "secret": {"secretName": "vendor-token"}},