fix(agentx): Dynamo frontend total, vLLM P/D roles and SGLang non-device tiers / Dynamo 前端总率、vLLM P/D 角色与 SGLang 非 GPU 层

inferencex-e2e/infx/results/agentic/backends/base.py

@@ -72,6 +72,8 @@ def empty_server_metrics(present: bool, metric_count: int) -> dict[str, Any]:
             "router_shared_cache_hit_rate": None,
             "frontend_cached_tokens": None,
             "frontend_input_tokens": None,
+            "overall_source": None,
+            "role_scope": None,
         },
         "kv_cache": {
             "gpu_usage_pct": None,

inferencex-e2e/infx/results/agentic/backends/dynamo_vllm.py

@@ -44,9 +44,11 @@ def populate(
         metrics: dict[str, dict[str, Any]],
         flat: dict[str, Any],
         nested: dict[str, Any],
+        *,
+        context: dict[str, Any] | None = None,
     ) -> None:
-        """Fill the vLLM fields, then the Dynamo frontend and router fields, from ``metrics``."""
-        super().populate(metrics, flat, nested)
+        """Fill the vLLM fields, then the Dynamo frontend and router fields."""
+        super().populate(metrics, flat, nested, context=context)
         self._populate_dynamo(metrics, flat, nested)
 
     def _populate_dynamo(
@@ -96,9 +98,13 @@ def _populate_dynamo(
             )
         )
         if flat["server_overall_cache_hit_rate"] is None:
-            overall = frontend_hit_rate if frontend_hit_rate is not None else router_shared_hit_rate
-            flat["server_overall_cache_hit_rate"] = overall
-            nested["cache"]["overall_cache_hit_rate"] = overall
+            if frontend_hit_rate is not None:
+                flat["server_overall_cache_hit_rate"] = frontend_hit_rate
+                nested["cache"]["overall_cache_hit_rate"] = frontend_hit_rate
+                nested["cache"]["overall_source"] = "frontend"
+            elif router_shared_hit_rate is not None:
+                flat["server_overall_cache_hit_rate"] = router_shared_hit_rate
+                nested["cache"]["overall_source"] = "router_estimate"
 
         nested["cache"].update(
             {

inferencex-e2e/infx/results/agentic/backends/sglang.py

@@ -61,10 +61,14 @@ def populate(
         device_hits = cached_by_source.get("device")
         host_hits = cached_by_source.get("host")
         total_cached = sum(cached_by_source.values()) if cached_by_source else None
+        non_device = [v for k, v in cached_by_source.items() if k != "device"]
 
         flat["server_gpu_cache_hit_rate"] = rate(device_hits, prompt_total)
         flat["server_cpu_cache_hit_rate"] = rate(host_hits, prompt_total)
-        flat["server_external_cache_hit_rate"] = flat["server_cpu_cache_hit_rate"]
+        # External reuse includes host and storage; CPU reuse includes host only.
+        flat["server_external_cache_hit_rate"] = (
+            rate(sum(non_device), prompt_total) if non_device else None
+        )
         flat["server_overall_cache_hit_rate"] = rate(total_cached, prompt_total)
 
         if flat["server_overall_cache_hit_rate"] is None:
@@ -136,7 +140,7 @@ def populate(
                 "generation_total": flat["total_generation_tokens"],
                 "prompt_by_source": {
                     "gpu_cache_hit": device_hits,
-                    "cpu_or_external_cache_hit": host_hits,
+                    "cpu_or_external_cache_hit": sum(non_device) if non_device else None,
                     "computed": prefill_compute,
                     "raw": cached_by_source,
                 },

inferencex-e2e/infx/results/agentic/backends/vllm.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import re
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from typing import Any
 
 from infx.results.agentic.common import (
@@ -22,6 +22,7 @@
 
 class VllmBackend(ServerMetricsBackend):
     name = "vllm"
+    accepts_context = True
     _ENGINE_TAG_RE = re.compile(r"\((?P<tag>EngineCore(?:_DP\d+)?)\s+pid=\d+\)")
     _GPU_KV_SIZE_RE = re.compile(r"GPU KV cache size:\s*(?P<tokens>[\d,]+)\s*tokens")
 
@@ -37,21 +38,28 @@ def populate(
         metrics: dict[str, dict[str, Any]],
         flat: dict[str, Any],
         nested: dict[str, Any],
+        *,
+        context: dict[str, Any] | None = None,
     ) -> None:
-        """Fill token totals, prefix and external hit rates, KV usage, offload stats and sources from ``metrics``."""
+        """Fill token totals, cache rates, KV usage, offload statistics and sources."""
+        endpoint_roles = (context or {}).get("endpoint_roles")
         prompt_total, generation_total = self.prompt_generation_totals(metrics)
         flat["total_prompt_tokens"] = counter_int(prompt_total)
         flat["total_generation_tokens"] = counter_int(generation_total)
 
+        disaggregated = nested["validity"]["deployment"]["disaggregated"]
+        role_scope, role_filter = _cache_role_filter(metrics, disaggregated, endpoint_roles)
         prefix_hits = sum_stat(
             metrics,
             "vllm:prefix_cache_hits",
             preferred_keys=("total", "sum", "max", "avg"),
+            series_filter=role_filter,
         )
         prefix_queries = sum_stat(
             metrics,
             "vllm:prefix_cache_queries",
             preferred_keys=("total", "sum", "max", "avg"),
+            series_filter=role_filter,
         )
         gpu_rate = rate(prefix_hits, prefix_queries)
         flat["server_gpu_cache_hit_rate"] = gpu_rate
@@ -60,11 +68,13 @@ def populate(
             metrics,
             "vllm:external_prefix_cache_hits",
             preferred_keys=("total", "sum", "max", "avg"),
+            series_filter=role_filter,
         )
         external_queries = sum_stat(
             metrics,
             "vllm:external_prefix_cache_queries",
             preferred_keys=("total", "sum", "max", "avg"),
+            series_filter=role_filter,
         )
         external_rate = rate(external_hits, external_queries)
         flat["server_external_cache_hit_rate"] = external_rate
@@ -75,6 +85,7 @@ def populate(
             "vllm:prompt_tokens_by_source",
             "source",
             preferred_keys=("total", "sum", "max", "avg"),
+            series_filter=role_filter,
         )
         if prompt_by_source:
             local_cache_hit = prompt_by_source.get("local_cache_hit")
@@ -145,6 +156,10 @@ def populate(
                 "prefix_cache_queries": prefix_queries,
                 "external_prefix_cache_hits": external_hits,
                 "external_prefix_cache_queries": external_queries,
+                "overall_source": "worker_counters"
+                if flat["server_overall_cache_hit_rate"] is not None
+                else None,
+                "role_scope": role_scope,
             }
         )
         nested["kv_cache"].update(
@@ -173,7 +188,7 @@ def populate(
                 "generation_total": flat["total_generation_tokens"],
             }
         )
-        nested["sources"] = _vllm_sources(metrics)
+        nested["sources"] = _vllm_sources(metrics, disaggregated, endpoint_roles)
 
     def prompt_generation_totals(
         self,
@@ -256,8 +271,12 @@ def first_counter_total(
     return None
 
 
-def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
-    """Return per-source token, hit rate and KV usage rows from vLLM ``metrics``, sorted by source id."""
+def _vllm_sources(
+    metrics: dict[str, dict[str, Any]],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> list[dict[str, Any]]:
+    """Return per-source token, hit rate and KV usage rows, sorted by source id."""
     source_ids = set()
     for metric_name in (
         "vllm:prompt_tokens",
@@ -268,16 +287,16 @@ def _vllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
         "vllm:prompt_tokens_by_source",
     ):
         for series in metric_series(metrics, metric_name):
-            source_ids.add(_source_id(series))
+            source_ids.add(_source_id(series, disaggregated, endpoint_roles))
 
     sources: list[dict[str, Any]] = []
     for source_id in sorted(source_ids):
         if not source_id:
             continue
 
         def series_filter(series: dict, source_id: str = source_id) -> bool:
-            """Return whether ``series`` belongs to ``source_id``."""
-            return _source_id(series) == source_id
+            """Return whether the series belongs to the selected source."""
+            return _source_id(series, disaggregated, endpoint_roles) == source_id
 
         prompt_tokens = sum_stat(metrics, "vllm:prompt_tokens", series_filter=series_filter)
         generation_tokens = sum_stat(metrics, "vllm:generation_tokens", series_filter=series_filter)
@@ -304,18 +323,95 @@ def series_filter(series: dict, source_id: str = source_id) -> bool:
     return sources
 
 
-def _source_id(series: dict[str, Any]) -> str:
-    """Build a ``role|endpoint|key=value`` id for ``series`` from its component, endpoint and rank labels."""
+_CACHE_METRICS = (
+    "vllm:prompt_tokens_by_source",
+    "vllm:prefix_cache_hits",
+    "vllm:prefix_cache_queries",
+    "vllm:external_prefix_cache_hits",
+    "vllm:external_prefix_cache_queries",
+)
+
+
+def _series_role(
+    series: dict[str, Any],
+    disaggregated: bool | None = None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> str:
+    role = _label_role(series, disaggregated)
+    declared = (endpoint_roles or {}).get(str(series.get("endpoint_url", "")))
+    if declared is None:
+        return role
+    declared = {
+        "prefill": "prefill",
+        "decode": "decode",
+        "unified": "combined",
+        "combined": "combined",
+    }.get(declared, "unknown")
+    labels = series.get("labels") if isinstance(series.get("labels"), dict) else {}
+    if any(key in labels for key in ("disaggregation_mode", "dynamo_component", "component")):
+        return role if role == declared else "unknown"
+    return declared
+
+
+def _label_role(series: dict[str, Any], disaggregated: bool | None = None) -> str:
     labels = series.get("labels") if isinstance(series.get("labels"), dict) else {}
+    mode = str(labels.get("disaggregation_mode") or "").lower()
+    if mode in ("prefill", "context"):
+        return "prefill"
+    if mode == "decode":
+        return "decode"
+    if mode in ("combined", "unified", "prefill_and_decode"):
+        return "combined"
+    if "disaggregation_mode" in labels:
+        return "unknown"
     component = str(labels.get("dynamo_component") or labels.get("component") or "")
     if component == "prefill":
-        role = "prefill"
-    elif component in ("backend", "decode"):
-        role = "decode"
-    elif component in ("frontend", "router"):
-        role = "router"
+        return "prefill"
+    if component == "backend" and disaggregated is False:
+        return "combined"
+    if component in ("backend", "decode"):
+        return "decode"
+    if component in ("frontend", "router"):
+        return "router"
+    return "combined"
+
+
+def _cache_role_filter(
+    metrics: dict[str, dict[str, Any]],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> tuple[str, Any]:
+    roles = {
+        _series_role(s, disaggregated, endpoint_roles)
+        for name in _CACHE_METRICS
+        for s in metric_series(metrics, name)
+    }
+    if "prefill" in roles:
+        scope = "mixed" if "combined" in roles else "prefill"
     else:
-        role = "combined"
+        scope = "unresolved" if "decode" in roles or disaggregated is True else "combined"
+    if "unknown" in roles or (disaggregated is True and "combined" in roles):
+        scope = "unresolved"
+    # Known decode counters measure KV transfer, including alongside unlabelled workers.
+    return (
+        scope,
+        lambda series: (
+            _series_role(series, disaggregated, endpoint_roles)
+            in (
+                "prefill",
+                "combined",
+            )
+        ),
+    )
+
+
+def _source_id(
+    series: dict[str, Any],
+    disaggregated: bool | None,
+    endpoint_roles: Mapping[str, str] | None = None,
+) -> str:
+    """Build a role, endpoint and rank identity for a worker series."""
+    role = _series_role(series, disaggregated, endpoint_roles)
 
     parts = [role]
     endpoint = series.get("endpoint_url")

inferencex-e2e/infx/results/agentic/common.py

@@ -200,11 +200,14 @@ def sum_by_label(
     label_keys: str | list[str],
     *,
     preferred_keys: tuple[str, ...] = ("total", "sum", "max", "avg"),
+    series_filter: Callable[[dict[str, Any]], bool] | None = None,
 ) -> dict[str, float]:
     """Sum the preferred stat of ``metric_names`` grouped by the first present label among ``label_keys``."""
     keys = [label_keys] if isinstance(label_keys, str) else label_keys
     out: dict[str, float] = {}
     for series in metric_series(metrics, metric_names):
+        if series_filter is not None and not series_filter(series):
+            continue
         labels = series.get("labels")
         if not isinstance(labels, dict):
             continue

inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py

@@ -98,6 +98,8 @@
     "router_shared_cache_hit_rate",
     "frontend_cached_tokens",
     "frontend_input_tokens",
+    "overall_source",
+    "role_scope",
 }
 SERVER_KV_CACHE_KEYS = {
     "gpu_usage_pct",
@@ -1667,13 +1669,12 @@ def test_processor_normalizes_dynamo_server_metrics(tmp_path: Path):
     assert any(source["role"] == "prefill" for source in agg["server_metrics"]["sources"])
 
 
-@pytest.mark.parametrize(("cached", "router_shared", "expected"), [
-    (0.0, 0.55, 0.0),
-    (25.0, None, 0.25),
-    (None, 0.55, 0.55),
-])
-def test_dynamo_overall_cache_hit_rate_falls_back_to_frontend_then_router(
-    cached, router_shared, expected,
+@pytest.mark.parametrize(
+    ("cached", "router_shared", "measured", "diagnostic"),
+    [(0.0, 0.55, 0.0, 0.0), (25.0, None, 0.25, 0.25), (None, 0.55, None, 0.55)],
+)
+def test_dynamo_measured_overall_uses_frontend_and_keeps_router_estimate_diagnostic(
+    cached, router_shared, measured, diagnostic,
 ):
     metrics = {
         "dynamo_frontend_input_sequence_tokens": {
@@ -1689,12 +1690,21 @@ def test_dynamo_overall_cache_hit_rate_falls_back_to_frontend_then_router(
             "type": "gauge", "series": [{"stats": {"avg": router_shared}}],
         }
 
-    _, nested, _ = compute_server_metrics(
+    flat, nested, _ = compute_server_metrics(
         {"metrics": metrics}, framework="dynamo-vllm", records=[],
     )
 
     assert nested["adapter"] == "dynamo-vllm"
-    assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(expected)
+    assert flat["server_overall_cache_hit_rate"] == pytest.approx(diagnostic)
+    if measured is None:
+        assert nested["cache"]["overall_cache_hit_rate"] is None
+        assert nested["cache"]["overall_source"] == "router_estimate"
+        assert nested["cache_validity"]["published"] is False
+        assert nested["cache_validity"]["basis"] is None
+    else:
+        assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(measured)
+        assert nested["cache"]["overall_source"] == "frontend"
+        assert nested["cache_validity"]["published"] is True