From 9740fde54d406f27ac2b9b38c369a710663c3c69 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 07:44:46 +0000 Subject: [PATCH 1/3] fix: account ATOMesh cache reuse on prefill workers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 仅汇总 prefill/unified 缓存计数，CPU reuse 使用 offload counter， overall 合并 GPU 和 CPU reuse；缺少增量或角色时保留未知值。 --- .../infx/results/agentic/__init__.py | 1 + .../infx/results/agentic/backends/atom.py | 111 ++++++++-- .../infx/results/agentic/server_metrics.py | 37 +++- .../agentic/test_atom_cache_counters.py | 195 ++++++++++++++++++ .../agentic/test_process_agentic_result.py | 49 +++++ 5 files changed, 374 insertions(+), 19 deletions(-) create mode 100644 inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py diff --git a/inferencex-e2e/infx/results/agentic/__init__.py b/inferencex-e2e/infx/results/agentic/__init__.py index 5cf7469691..92a78f8584 100644 --- a/inferencex-e2e/infx/results/agentic/__init__.py +++ b/inferencex-e2e/infx/results/agentic/__init__.py @@ -218,6 +218,7 @@ def build_result( framework=framework, records=records, server_logs=server_logs, + env=env, ) if "total_tput_tps" in request_flat and num_gpus > 0: diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 947024b111..7c6071adff 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -2,12 +2,24 @@ from __future__ import annotations +import logging +import math +from collections.abc import Mapping from typing import Any -from infx.results.agentic.common import gauge_stat, normalize_fraction, rate, sum_stat +from infx.results.agentic.common import ( + gauge_stat, + metric_series, + normalize_fraction, + rate, + series_stat, + sum_stat, +) from .base import ServerMetricsBackend, counter_int +logger = logging.getLogger(__name__) + def _atom_names(suffix: str) -> list[str]: """Accept direct worker metrics and Atomesh's colon-normalized form.""" @@ -19,6 +31,73 @@ def _atom_counter_names(stem: str) -> list[str]: return [*_atom_names(stem), *_atom_names(f"{stem}_total")] +def _counter_total(metrics: dict[str, dict[str, Any]], stem: str, endpoint: str) -> float | None: + names = _atom_counter_names(f"prefix_cache_{stem}_tokens") + series = [s for s in metric_series(metrics, names) if s.get("endpoint_url") == endpoint] + if not series or any( + (value := series_stat(s, preferred_keys=("total",))) is None + or not math.isfinite(value) + or value < 0 + for s in series + ): + logger.warning("ATOM %s cache counter missing or invalid at %s", stem, endpoint) + return None + total = sum_stat( + metrics, + names, + preferred_keys=("total",), + series_filter=lambda s: s.get("endpoint_url") == endpoint, + ) + return total if total is not None and math.isfinite(total) else None + + +def _cache_counts( + metrics: dict[str, dict[str, Any]], + endpoint_roles: Mapping[str, str] | None, + disaggregated: bool | None, +) -> tuple[float | None, float | None, float | None]: + names = [ + name + for stem in ("full", "cached", "offload") + for name in _atom_counter_names(f"prefix_cache_{stem}_tokens") + ] + endpoints = [s.get("endpoint_url") for s in metric_series(metrics, names)] + if any(not isinstance(endpoint, str) or not endpoint for endpoint in endpoints): + return None, None, None + observed = set(endpoints) + if endpoint_roles is None: + accounted = observed if disaggregated is False and len(observed) == 1 else set() + else: + # Decode counts the prompt again without serving its prefill cache reuse. + accounted = {url for url, role in endpoint_roles.items() if role in {"prefill", "unified"}} + if observed - endpoint_roles.keys(): + accounted = set() + if not accounted: + logger.warning("ATOM cache endpoint roles missing or ambiguous") + return None, None, None + full_total = cached_total = offload_total = 0.0 + cpu_complete = True + for endpoint in accounted: + full, cached, offload = ( + _counter_total(metrics, stem, endpoint) for stem in ("full", "cached", "offload") + ) + if full is None or cached is None or cached > full: + return None, None, None + full_total += full + cached_total += cached + if offload is None or cached + offload > full: + cpu_complete = False + else: + offload_total += offload + if full_total <= 0 or not all(math.isfinite(v) for v in (full_total, cached_total)): + return None, None, None + return ( + full_total, + cached_total, + offload_total if cpu_complete and math.isfinite(offload_total) else None, + ) + + class AtomBackend(ServerMetricsBackend): name = "atom" @@ -33,6 +112,9 @@ def populate( metrics: dict[str, dict[str, Any]], flat: dict[str, Any], nested: dict[str, Any], + *, + endpoint_roles: Mapping[str, str] | None = None, + disaggregated: bool | None = None, ) -> None: prompt_total = sum_stat( metrics, @@ -47,24 +129,17 @@ def populate( flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) - cached_tokens = sum_stat( - metrics, - _atom_counter_names("prefix_cache_cached_tokens"), - preferred_keys=("total", "sum", "max", "avg"), - ) - full_tokens = sum_stat( - metrics, - _atom_counter_names("prefix_cache_full_tokens"), - preferred_keys=("total", "sum", "max", "avg"), + full_tokens, cached_tokens, external_tokens = _cache_counts( + metrics, endpoint_roles, disaggregated ) cache_hit_rate = rate(cached_tokens, full_tokens) - external_tokens = sum_stat( - metrics, - _atom_counter_names("lmcache_loaded_tokens"), - preferred_keys=("total", "sum", "max", "avg"), - ) + # Offload is CPU reuse beyond the GPU prefix; loaded tokens measure transfer volume. external_hit_rate = rate(external_tokens, full_tokens) - overall_hit_rate = cache_hit_rate + overall_hit_rate = ( + rate(cached_tokens + external_tokens, full_tokens) + if cached_tokens is not None and external_tokens is not None + else None + ) flat["server_gpu_cache_hit_rate"] = cache_hit_rate flat["server_cpu_cache_hit_rate"] = external_hit_rate @@ -88,7 +163,9 @@ def populate( "prefix_cache_hits": cached_tokens, "prefix_cache_queries": full_tokens, "external_prefix_cache_hits": external_tokens, - "external_prefix_cache_queries": full_tokens, + "external_prefix_cache_queries": ( + full_tokens if external_tokens is not None else None + ), } ) nested["kv_cache"]["gpu_usage_pct"] = flat["gpu_kv_cache_usage_pct"] diff --git a/inferencex-e2e/infx/results/agentic/server_metrics.py b/inferencex-e2e/infx/results/agentic/server_metrics.py index d14962a61f..e100011642 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -2,10 +2,11 @@ from __future__ import annotations -from collections.abc import Iterable +from collections.abc import Iterable, Mapping from typing import Any from .backends import detect_backend +from .backends.atom import AtomBackend from .backends.base import ( SERVER_CACHE_FLAT_FIELDS, apply_profile_totals, @@ -14,12 +15,34 @@ from .common import index_server_metrics +def endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: + if env.get("AIPERF_SERVER_METRICS_URLS") or env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": + return {} + groups = ( + [("unified", env["SRT_AGG_ENDPOINTS"])] + if env.get("SRT_AGG_ENDPOINTS") + else [ + ("prefill", env.get("SRT_PREFILL_ENDPOINTS", "")), + ("decode", env.get("SRT_DECODE_ENDPOINTS", "")), + ] + ) + roles: dict[str, str] = {} + for role, endpoints in groups: + for endpoint in endpoints.split(","): + if endpoint: + url = f"http://{endpoint}/metrics" + if roles.setdefault(url, role) != role: + return {} + return roles or None + + def compute_server_metrics( server_metrics: dict[str, Any], *, framework: str, records: list[dict[str, Any]], server_logs: Iterable[str | None] = (), + env: Mapping[str, str] | None = None, ) -> tuple[dict[str, Any], dict[str, Any], list[str]]: metrics = index_server_metrics(server_metrics) flat = SERVER_CACHE_FLAT_FIELDS.copy() @@ -46,7 +69,17 @@ def compute_server_metrics( f"framework={framework.lower()!r}, metric_names=[{preview}]" ) nested["adapter"] = adapter.name - adapter.populate(metrics, flat, nested) + if isinstance(adapter, AtomBackend): + source_env = env or {} + adapter.populate( + metrics, + flat, + nested, + endpoint_roles=endpoint_roles_from_env(source_env), + disaggregated={"true": True, "false": False}.get(source_env.get("DISAGG", "").lower()), + ) + else: + adapter.populate(metrics, flat, nested) if framework.lower() == "dynamo-sglang" and adapter.name == "sglang": nested["kv_cache"]["gpu_total_tokens"] = None diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py new file mode 100644 index 0000000000..2d0e6f433d --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +import math + +import pytest + +from infx.results.agentic.server_metrics import compute_server_metrics + + +def _export(full: dict, cached: dict, offload: dict) -> dict: + return { + "metrics": { + f"atom:prefix_cache_{stem}_tokens": { + "series": [ + {"endpoint_url": f"http://{endpoint}/metrics", "stats": {"total": value}} + for endpoint, value in values.items() + ] + } + for stem, values in (("full", full), ("cached", cached), ("offload", offload)) + } + } + + +def _compute(export: dict, env: dict) -> tuple[dict, dict]: + flat, nested, _ = compute_server_metrics(export, framework="mooncake-atom", records=[], env=env) + return flat, nested + + +def _assert_rates( + flat: dict, nested: dict, gpu: float | None, cpu: float | None, overall: float | None +): + for name, expected in (("gpu", gpu), ("cpu", cpu), ("external", cpu), ("overall", overall)): + actual = flat[f"server_{name}_cache_hit_rate"] + assert actual == (pytest.approx(expected) if expected is not None else None) + assert nested["cache"][f"{name}_cache_hit_rate"] == actual + + +@pytest.mark.parametrize("decode_cached", [800, math.nan]) +def test_prefill_counters_exclude_decode_and_transfer_volume(decode_cached: float) -> None: + export = _export({"p": 1000, "d": 1000}, {"p": 600, "d": decode_cached}, {"p": 100, "d": -1}) + export["metrics"]["atom:lmcache_loaded_tokens"] = { + "series": [{"endpoint_url": "http://p/metrics", "stats": {"total": 900}}] + } + flat, nested = _compute( + export, {"SRT_PREFILL_ENDPOINTS": "p", "SRT_DECODE_ENDPOINTS": "d", "DISAGG": "true"} + ) + _assert_rates(flat, nested, 0.6, 0.1, 0.7) + assert nested["cache"]["prefix_cache_queries"] == 1000 + assert nested["cache"]["external_prefix_cache_hits"] == 100 + assert nested["cache"]["external_prefix_cache_queries"] == 1000 + + +@pytest.mark.parametrize( + ("env", "offload", "expected"), + [ + ({"SRT_AGG_ENDPOINTS": "worker", "SRT_PREFILL_ENDPOINTS": "unused"}, 50, 0.5), + ({"DISAGG": "false"}, 0, 0), + ], +) +def test_unified_zero_gpu_and_offload_counter_alias( + env: dict, offload: int, expected: float +) -> None: + export = _export({"worker": 100}, {"worker": 0}, {"worker": offload}) + export["metrics"] = { + name.replace("atom:", "atom_") + "_total": value + for name, value in export["metrics"].items() + } + flat, nested = _compute(export, env) + _assert_rates(flat, nested, 0, expected, expected) + + +def test_multiple_prefills_and_labelled_series_are_token_weighted() -> None: + export = _export( + {"p": 100, "q": 900, "idle": 0}, + {"p": 80, "q": 20, "idle": 0}, + {"p": 10, "q": 180, "idle": 0}, + ) + for stem, values in (("full", (400, 500)), ("cached", (5, 15)), ("offload", (80, 100))): + series = export["metrics"][f"atom:prefix_cache_{stem}_tokens"]["series"] + series[:] = [s for s in series if s["endpoint_url"] != "http://q/metrics"] + series.extend( + { + "endpoint_url": "http://q/metrics", + "labels": {"engine": str(i)}, + "stats": {"total": v}, + } + for i, v in enumerate(values) + ) + flat, nested = _compute(export, {"SRT_PREFILL_ENDPOINTS": "p,q,idle"}) + _assert_rates(flat, nested, 0.1, 0.19, 0.29) + + +@pytest.mark.parametrize( + "env", + [ + {}, + {"DISAGG": "true"}, + {"SRT_PREFILL_ENDPOINTS": "p", "SRT_DECODE_ENDPOINTS": "p,d"}, + {"SRT_PREFILL_ENDPOINTS": "p,missing", "SRT_DECODE_ENDPOINTS": "d"}, + {"SRT_PREFILL_ENDPOINTS": "p"}, + { + "SRT_PREFILL_ENDPOINTS": "p", + "SRT_DECODE_ENDPOINTS": "d", + "AIPERF_SERVER_METRICS_URLS": "custom", + }, + { + "SRT_PREFILL_ENDPOINTS": "p", + "SRT_DECODE_ENDPOINTS": "d", + "SRTCTL_FRONTEND_TYPE": "dynamo", + }, + ], +) +def test_unknown_conflicting_or_incomplete_roles_do_not_mix_endpoints(env: dict) -> None: + flat, nested = _compute( + _export({"p": 100, "d": 100}, {"p": 60, "d": 0}, {"p": 10, "d": 0}), env + ) + _assert_rates(flat, nested, None, None, None) + + +def test_single_endpoint_without_explicit_unified_topology_stays_unknown() -> None: + flat, nested = _compute(_export({"worker": 100}, {"worker": 60}, {"worker": 10}), {}) + _assert_rates(flat, nested, None, None, None) + + +@pytest.mark.parametrize( + ("offload", "cpu", "overall"), + [({"p": 0, "q": 0}, 0, 0.2), ({}, None, None), ({"p": 10}, None, None)], +) +def test_missing_or_partial_offload_is_distinct_from_zero( + offload: dict, cpu: float | None, overall: float | None +) -> None: + flat, nested = _compute( + _export({"p": 100, "q": 100}, {"p": 20, "q": 20}, offload), + {"SRT_PREFILL_ENDPOINTS": "p,q"}, + ) + _assert_rates(flat, nested, 0.2, cpu, overall) + assert nested["cache"]["external_prefix_cache_queries"] == (200 if cpu is not None else None) + + +@pytest.mark.parametrize( + ("stem", "value"), + [ + ("full", 0), + ("cached", -1), + ("cached", 101), + ("cached", math.nan), + ("cached", math.inf), + ("cached", None), + ], +) +def test_invalid_counted_counter_does_not_produce_rates(stem: str, value: object) -> None: + export = _export({"worker": 100}, {"worker": 20}, {"worker": 10}) + export["metrics"][f"atom:prefix_cache_{stem}_tokens"]["series"][0]["stats"]["total"] = value + flat, nested = _compute(export, {"DISAGG": "false"}) + _assert_rates(flat, nested, None, None, None) + + +def test_unreadable_series_does_not_get_silently_skipped() -> None: + export = _export({"worker": 100}, {"worker": 20}, {"worker": 10}) + export["metrics"]["atom:prefix_cache_cached_tokens"]["series"].append( + { + "endpoint_url": "http://worker/metrics", + "labels": {"engine": "other"}, + "stats": {"max": 5}, + } + ) + flat, nested = _compute(export, {"DISAGG": "false"}) + _assert_rates(flat, nested, None, None, None) + + +@pytest.mark.parametrize("offload", [30, math.nan]) +def test_invalid_worker_cpu_count_cannot_hide_in_another_workers_denominator( + offload: float, +) -> None: + flat, nested = _compute( + _export({"p": 100, "q": 900}, {"p": 80, "q": 20}, {"p": offload, "q": 0}), + {"SRT_PREFILL_ENDPOINTS": "p,q"}, + ) + _assert_rates(flat, nested, 0.1, None, None) + + +def test_unknown_cache_roles_preserve_other_atom_telemetry() -> None: + export = _export({"p": 100, "d": 100}, {"p": 60, "d": 0}, {"p": 10, "d": 0}) + export["metrics"].update( + { + "atom:prompt_tokens": {"series": [{"stats": {"total": 200}}]}, + "atom:generation_tokens": {"series": [{"stats": {"total": 30}}]}, + "atom:kv_cache_usage_ratio": {"series": [{"stats": {"max": 0.75}}]}, + } + ) + flat, nested = _compute(export, {}) + _assert_rates(flat, nested, None, None, None) + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 200 + assert flat["total_generation_tokens"] == nested["tokens"]["generation_total"] == 30 + assert flat["gpu_kv_cache_usage_pct"] == nested["kv_cache"]["gpu_usage_pct"] == 0.75 diff --git a/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py b/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py index e162f49685..29aeeeef94 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_process_agentic_result.py @@ -365,6 +365,55 @@ def test_processor_records_run_attempt(tmp_path: Path): assert "run_attempt" not in local +def test_processor_uses_launcher_roles_for_atom_cache_counts(tmp_path: Path) -> None: + result_dir = _write_fixture(tmp_path) + metrics = { + f"atom:prefix_cache_{stem}_tokens": { + "series": [ + {"endpoint_url": f"http://{endpoint}/metrics", "stats": {"total": value}} + for endpoint, value in values.items() + ] + } + for stem, values in ( + ("full", {"prefill:8000": 1000, "decode:8000": 1000}), + ("cached", {"prefill:8000": 600, "decode:8000": 0}), + ("offload", {"prefill:8000": 100, "decode:8000": 0}), + ) + } + (result_dir / "aiperf_artifacts" / "server_metrics_export.json").write_text( + json.dumps({"metrics": metrics}) + ) + agg = _run_processor( + result_dir, + tmp_path / "out", + { + "FRAMEWORK": "mooncake-atom", + "DISAGG": "true", + "IS_MULTINODE": "true", + "PREFILL_NUM_WORKERS": "1", + "PREFILL_TP": "4", + "PREFILL_EP": "1", + "DECODE_NUM_WORKERS": "1", + "DECODE_TP": "4", + "DECODE_EP": "1", + "SRT_PREFILL_ENDPOINTS": "prefill:8000", + "SRT_DECODE_ENDPOINTS": "decode:8000", + "SRT_AGG_ENDPOINTS": "", + "AIPERF_SERVER_METRICS_URLS": "", + "SRTCTL_FRONTEND_TYPE": "", + "KV_OFFLOADING": "dram", + "KV_OFFLOAD_BACKEND": "lmcache", + "KV_OFFLOAD_BACKEND_METADATA": '{"name":"lmcache","version":"0.4.5"}', + }, + ) + cache = agg["server_metrics"]["cache"] + assert cache["gpu_cache_hit_rate"] == 0.6 + assert cache["cpu_cache_hit_rate"] == cache["external_cache_hit_rate"] == 0.1 + assert cache["overall_cache_hit_rate"] == 0.7 + assert cache["prefix_cache_queries"] == cache["external_prefix_cache_queries"] == 1000 + assert agg["num_gpus"] == 8 + + def test_processor_preserves_dataset_provenance(tmp_path: Path): result_dir = _write_fixture(tmp_path) output_dir = tmp_path / "out" From 984280caef3ac62f0a498e8f539da9904ea0a939 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Thu, 22 Oct 2026 20:28:52 +0000 Subject: [PATCH 2/3] fix(results): preserve ATOM unified counters and prefill totals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 仅在确认关闭 CPU offload 且无计数器时补零，保留单端点自定义 metrics 的角色回退，并从输入总量中排除 decode 重复计数。补充相应行为回归。 --- .../infx/results/agentic/backends/atom.py | 30 ++++++++++++++-- .../infx/results/agentic/server_metrics.py | 9 ++++- .../agentic/test_atom_cache_counters.py | 34 +++++++++++++++++-- 3 files changed, 67 insertions(+), 6 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 7c6071adff..17929bb0c8 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -55,6 +55,7 @@ def _cache_counts( metrics: dict[str, dict[str, Any]], endpoint_roles: Mapping[str, str] | None, disaggregated: bool | None, + cpu_offload_disabled: bool = False, ) -> tuple[float | None, float | None, float | None]: names = [ name @@ -77,10 +78,16 @@ def _cache_counts( return None, None, None full_total = cached_total = offload_total = 0.0 cpu_complete = True + offload_absent = not any( + s.get("endpoint_url") in accounted + for s in metric_series(metrics, _atom_counter_names("prefix_cache_offload_tokens")) + ) for endpoint in accounted: full, cached, offload = ( _counter_total(metrics, stem, endpoint) for stem in ("full", "cached", "offload") ) + if cpu_offload_disabled and offload_absent: + offload = 0.0 if full is None or cached is None or cached > full: return None, None, None full_total += full @@ -115,11 +122,28 @@ def populate( *, endpoint_roles: Mapping[str, str] | None = None, disaggregated: bool | None = None, + cpu_offload_disabled: bool = False, ) -> None: + prompt_names = _atom_counter_names("prompt_tokens") + prompt_endpoints = {s.get("endpoint_url") for s in metric_series(metrics, prompt_names)} + if endpoint_roles is None: + accounted = ( + prompt_endpoints if disaggregated is False and len(prompt_endpoints) == 1 else set() + ) + else: + accounted = { + url for url, role in endpoint_roles.items() if role in {"prefill", "unified"} + } + if prompt_endpoints - endpoint_roles.keys(): + accounted = set() + accounted = {url for url in accounted if isinstance(url, str) and url} + if accounted - prompt_endpoints: + accounted = set() prompt_total = sum_stat( metrics, - _atom_counter_names("prompt_tokens"), - preferred_keys=("total", "sum", "max", "avg"), + prompt_names, + preferred_keys=("total",), + series_filter=lambda s: s.get("endpoint_url") in accounted, ) generation_total = sum_stat( metrics, @@ -130,7 +154,7 @@ def populate( flat["total_generation_tokens"] = counter_int(generation_total) full_tokens, cached_tokens, external_tokens = _cache_counts( - metrics, endpoint_roles, disaggregated + metrics, endpoint_roles, disaggregated, cpu_offload_disabled ) cache_hit_rate = rate(cached_tokens, full_tokens) # Offload is CPU reuse beyond the GPU prefix; loaded tokens measure transfer volume. diff --git a/inferencex-e2e/infx/results/agentic/server_metrics.py b/inferencex-e2e/infx/results/agentic/server_metrics.py index e100011642..2ef47da115 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -16,8 +16,10 @@ def endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: - if env.get("AIPERF_SERVER_METRICS_URLS") or env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": + if env.get("SRTCTL_FRONTEND_TYPE") == "dynamo": return {} + if env.get("AIPERF_SERVER_METRICS_URLS"): + return None groups = ( [("unified", env["SRT_AGG_ENDPOINTS"])] if env.get("SRT_AGG_ENDPOINTS") @@ -77,6 +79,11 @@ def compute_server_metrics( nested, endpoint_roles=endpoint_roles_from_env(source_env), disaggregated={"true": True, "false": False}.get(source_env.get("DISAGG", "").lower()), + cpu_offload_disabled=(source_env.get("KV_OFFLOADING", "").strip().lower() == "none") + and not ( + source_env.get("KV_OFFLOAD_BACKEND") + or source_env.get("KV_OFFLOAD_BACKEND_METADATA") + ), ) else: adapter.populate(metrics, flat, nested) diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py index 2d0e6f433d..40ea3d9b89 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py @@ -41,9 +41,16 @@ def test_prefill_counters_exclude_decode_and_transfer_volume(decode_cached: floa export["metrics"]["atom:lmcache_loaded_tokens"] = { "series": [{"endpoint_url": "http://p/metrics", "stats": {"total": 900}}] } + export["metrics"]["atom:prompt_tokens"] = { + "series": [ + {"endpoint_url": f"http://{endpoint}/metrics", "stats": {"total": 1000}} + for endpoint in ("p", "d") + ] + } flat, nested = _compute( export, {"SRT_PREFILL_ENDPOINTS": "p", "SRT_DECODE_ENDPOINTS": "d", "DISAGG": "true"} ) + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 1000 _assert_rates(flat, nested, 0.6, 0.1, 0.7) assert nested["cache"]["prefix_cache_queries"] == 1000 assert nested["cache"]["external_prefix_cache_hits"] == 100 @@ -55,6 +62,7 @@ def test_prefill_counters_exclude_decode_and_transfer_volume(decode_cached: floa [ ({"SRT_AGG_ENDPOINTS": "worker", "SRT_PREFILL_ENDPOINTS": "unused"}, 50, 0.5), ({"DISAGG": "false"}, 0, 0), + ({"DISAGG": "false", "AIPERF_SERVER_METRICS_URLS": "http://worker/metrics"}, 50, 0.5), ], ) def test_unified_zero_gpu_and_offload_counter_alias( @@ -188,8 +196,30 @@ def test_unknown_cache_roles_preserve_other_atom_telemetry() -> None: "atom:kv_cache_usage_ratio": {"series": [{"stats": {"max": 0.75}}]}, } ) - flat, nested = _compute(export, {}) + flat, nested, _ = compute_server_metrics( + export, + framework="mooncake-atom", + records=[{"metrics": {"input_sequence_length": {"value": 123}}}], + env={}, + ) _assert_rates(flat, nested, None, None, None) - assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 200 + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 123 assert flat["total_generation_tokens"] == nested["tokens"]["generation_total"] == 30 assert flat["gpu_kv_cache_usage_pct"] == nested["kv_cache"]["gpu_usage_pct"] == 0.75 + + +@pytest.mark.parametrize( + ("env", "cpu", "overall"), + [ + ({"KV_OFFLOADING": " NONE "}, 0, 0.6), + ({"KV_OFFLOADING": "dram"}, None, None), + ({"KV_OFFLOADING": "none", "KV_OFFLOAD_BACKEND": "mooncake"}, None, None), + ], +) +def test_missing_offload_is_zero_only_with_confirmed_disabled_tier( + env: dict, cpu: float | None, overall: float | None +) -> None: + flat, nested = _compute( + _export({"worker": 100}, {"worker": 60}, {}), {"DISAGG": "false", **env} + ) + _assert_rates(flat, nested, 0.6, cpu, overall) From c14c0fde87eae59e1bf6194098255bf7078c9953 Mon Sep 17 00:00:00 2001 From: Chun Fang Date: Sun, 25 Oct 2026 17:25:34 +0000 Subject: [PATCH 3/3] fix(results): validate ATOM cache counters per label domain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 中文：按端点与完整标签域校验缓存分层计数；仅在 CPU offload 明确关闭且现有计数均为零时补齐缺失零值，拒绝无效或矛盾的 offload 数据。 --- .../infx/results/agentic/backends/atom.py | 76 ++++++++++++------- .../agentic/test_atom_cache_counters.py | 38 ++++++++-- 2 files changed, 78 insertions(+), 36 deletions(-) diff --git a/inferencex-e2e/infx/results/agentic/backends/atom.py b/inferencex-e2e/infx/results/agentic/backends/atom.py index 183ca99832..32543a005b 100644 --- a/inferencex-e2e/infx/results/agentic/backends/atom.py +++ b/inferencex-e2e/infx/results/agentic/backends/atom.py @@ -31,24 +31,31 @@ def _atom_counter_names(stem: str) -> list[str]: return [*_atom_names(stem), *_atom_names(f"{stem}_total")] -def _counter_total(metrics: dict[str, dict[str, Any]], stem: str, endpoint: str) -> float | None: +def _counter_values( + metrics: dict[str, dict[str, Any]], stem: str, endpoint: str +) -> dict[tuple, float] | None: names = _atom_counter_names(f"prefix_cache_{stem}_tokens") - series = [s for s in metric_series(metrics, names) if s.get("endpoint_url") == endpoint] - if not series or any( - (value := series_stat(s, preferred_keys=("total",))) is None - or not math.isfinite(value) - or value < 0 - for s in series - ): - logger.warning("ATOM %s cache counter missing or invalid at %s", stem, endpoint) - return None - total = sum_stat( - metrics, - names, - preferred_keys=("total",), - series_filter=lambda s: s.get("endpoint_url") == endpoint, - ) - return total if total is not None and math.isfinite(total) else None + values = {} + for series in metric_series(metrics, names): + if series.get("endpoint_url") != endpoint: + continue + labels = series.get("labels") + labels = {} if labels is None else labels + value = series_stat(series, preferred_keys=("total",)) + if ( + not isinstance(labels, dict) + or any(not isinstance(k, str) or not isinstance(v, str) for k, v in labels.items()) + or value is None + or not math.isfinite(value) + or value < 0 + ): + return None + key = tuple(sorted(labels.items())) + if key in values: + logger.warning("ATOM %s cache counter duplicated at %s", stem, endpoint) + return None + values[key] = value + return values def _cache_counts( @@ -78,24 +85,35 @@ def _cache_counts( return None, None, None full_total = cached_total = offload_total = 0.0 cpu_complete = True - offload_absent = not any( - s.get("endpoint_url") in accounted - for s in metric_series(metrics, _atom_counter_names("prefix_cache_offload_tokens")) - ) for endpoint in accounted: full, cached, offload = ( - _counter_total(metrics, stem, endpoint) for stem in ("full", "cached", "offload") + _counter_values(metrics, stem, endpoint) for stem in ("full", "cached", "offload") ) - if cpu_offload_disabled and offload_absent: - offload = 0.0 - if full is None or cached is None or cached > full: + if ( + not full + or cached is None + or full.keys() != cached.keys() + or any(cached[key] > full[key] for key in full) + ): return None, None, None - full_total += full - cached_total += cached - if offload is None or cached + offload > full: + if cpu_offload_disabled: + offload = ( + dict.fromkeys(full, 0.0) + if offload is not None + and offload.keys() <= full.keys() + and not any(offload.values()) + else None + ) + full_total += sum(full.values()) + cached_total += sum(cached.values()) + if ( + offload is None + or offload.keys() != full.keys() + or any(cached[key] + offload[key] > full[key] for key in full) + ): cpu_complete = False else: - offload_total += offload + offload_total += sum(offload.values()) if full_total <= 0 or not all(math.isfinite(v) for v in (full_total, cached_total)): return None, None, None return ( diff --git a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py index 40ea3d9b89..8356e5bd39 100644 --- a/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py +++ b/inferencex-e2e/infx/tests/results/agentic/test_atom_cache_counters.py @@ -177,13 +177,17 @@ def test_unreadable_series_does_not_get_silently_skipped() -> None: @pytest.mark.parametrize("offload", [30, math.nan]) +@pytest.mark.parametrize("labelled", [False, True]) def test_invalid_worker_cpu_count_cannot_hide_in_another_workers_denominator( - offload: float, + offload: float, labelled: bool ) -> None: - flat, nested = _compute( - _export({"p": 100, "q": 900}, {"p": 80, "q": 20}, {"p": offload, "q": 0}), - {"SRT_PREFILL_ENDPOINTS": "p,q"}, - ) + export = _export({"p": 100, "q": 900}, {"p": 80, "q": 20}, {"p": offload, "q": 0}) + if labelled: + for metric in export["metrics"].values(): + for series in metric["series"]: + series["labels"] = {"engine": series["endpoint_url"]} + series["endpoint_url"] = "http://worker/metrics" + flat, nested = _compute(export, {"SRT_PREFILL_ENDPOINTS": "worker" if labelled else "p,q"}) _assert_rates(flat, nested, 0.1, None, None) @@ -216,10 +220,30 @@ def test_unknown_cache_roles_preserve_other_atom_telemetry() -> None: ({"KV_OFFLOADING": "none", "KV_OFFLOAD_BACKEND": "mooncake"}, None, None), ], ) +@pytest.mark.parametrize("offload", [{}, {"p": 0}]) def test_missing_offload_is_zero_only_with_confirmed_disabled_tier( - env: dict, cpu: float | None, overall: float | None + env: dict, cpu: float | None, overall: float | None, offload: dict ) -> None: flat, nested = _compute( - _export({"worker": 100}, {"worker": 60}, {}), {"DISAGG": "false", **env} + _export({"p": 100, "q": 100}, {"p": 60, "q": 60}, offload), + {"SRT_PREFILL_ENDPOINTS": "p,q", **env}, ) _assert_rates(flat, nested, 0.6, cpu, overall) + + +@pytest.mark.parametrize("offload", [10, math.nan, None]) +def test_disabled_cpu_tier_does_not_hide_invalid_present_offload(offload: object) -> None: + flat, nested = _compute( + _export({"p": 100, "q": 100}, {"p": 60, "q": 60}, {"p": offload}), + {"SRT_PREFILL_ENDPOINTS": "p,q", "KV_OFFLOADING": "none"}, + ) + _assert_rates(flat, nested, 0.6, None, None) + + +def test_disabled_cpu_tier_does_not_fill_an_unmatched_label_domain() -> None: + export = _export({"worker": 100}, {"worker": 60}, {"worker": 0}) + export["metrics"]["atom:prefix_cache_offload_tokens"]["series"][0]["labels"] = { + "engine": "other" + } + flat, nested = _compute(export, {"DISAGG": "false", "KV_OFFLOADING": "none"}) + _assert_rates(flat, nested, 0.6, None, None)