diff --git a/inferencex-e2e/infx/results/agentic/backends/base.py b/inferencex-e2e/infx/results/agentic/backends/base.py index c17639b192..d0bb6994f4 100644 --- a/inferencex-e2e/infx/results/agentic/backends/base.py +++ b/inferencex-e2e/infx/results/agentic/backends/base.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math from collections.abc import Iterable from typing import Any @@ -99,6 +100,15 @@ def empty_server_metrics(present: bool, metric_count: int) -> dict[str, Any]: }, }, "sources": [], + "validity": { + "deployment": {"disaggregated": None}, + "cache_rate": { + "status": "NOT_EVALUATED", + "published": False, + "eligible_exact_actual": False, + "reasons": ["adapter_does_not_assess_cache_rate"], + }, + }, } @@ -114,8 +124,8 @@ def apply_profile_totals(flat: dict[str, Any], records: list[dict[str, Any]]) -> def counter_int(value: float | None) -> int | None: - """Round a counter ``value`` to an int, keeping None.""" - if value is None: + """Round a finite counter to an int, keeping unavailable values None.""" + if value is None or not math.isfinite(value): return None return round(value) @@ -132,6 +142,6 @@ def _record_token_sum(records: list[dict[str, Any]], metric_name: str) -> int | try: total += int(value) found = True - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): continue return total if found else None diff --git a/inferencex-e2e/infx/results/agentic/backends/trtllm.py b/inferencex-e2e/infx/results/agentic/backends/trtllm.py index cb85924d28..3ca290b171 100644 --- a/inferencex-e2e/infx/results/agentic/backends/trtllm.py +++ b/inferencex-e2e/infx/results/agentic/backends/trtllm.py @@ -2,12 +2,14 @@ from __future__ import annotations -from collections.abc import Iterable +import math +from collections.abc import Callable, Iterable from typing import Any +from infx.results.agentic.cache_validity import modern_profile_error from infx.results.agentic.common import ( gauge_stat, - label_value, + metric_series, normalize_fraction, rate, sum_stat, @@ -15,9 +17,17 @@ from .base import ServerMetricsBackend, counter_int +PROMPT_TOKENS = "trtllm_prompt_tokens" +CACHED_TOKENS = "trtllm_prompt_cached_tokens" +GENERATION_TOKENS = "trtllm_generation_tokens" +# Interval counters do not identify the billed cohort or prove expected worker coverage. +UNBOUND_REASONS = ("request_scope_not_bound", "expected_instances_not_verified") +ROLE_LABELS = {"disaggregation_mode", "dynamo_component"} + class TrtllmBackend(ServerMetricsBackend): name = "trtllm" + accepts_context = True def matches(self, metrics: dict[str, dict[str, Any]], framework: str) -> bool: """Return True for trtllm_ metric names, or for an empty scrape from a TRT-LLM ``framework``.""" @@ -31,35 +41,20 @@ 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, KV hit rate, usage and offload bytes and per-endpoint sources from ``metrics``.""" - prompt_total = _first_counter_total( - metrics, - ["dynamo_frontend_input_sequence_tokens", "trtllm_prompt_tokens_total"], - ) - generation_total = _first_counter_total( - metrics, - ["dynamo_frontend_output_tokens", "trtllm_generation_tokens_total"], - ) - cached_tokens = sum_stat( - metrics, - "trtllm_prompt_cached_tokens_total", - preferred_keys=("total", "sum", "max", "avg"), + """Fill token totals, interval cache reuse and per-endpoint diagnostics.""" + context = context or {} + prompt_total = sum_stat( + metrics, "dynamo_frontend_input_sequence_tokens", preferred_keys=("total", "sum") ) + generation_total = _first_total(metrics, "dynamo_frontend_output_tokens", GENERATION_TOKENS) flat["total_prompt_tokens"] = counter_int(prompt_total) flat["total_generation_tokens"] = counter_int(generation_total) - flat["server_gpu_cache_hit_rate"] = normalize_fraction( - gauge_stat( - metrics, - "trtllm_kv_cache_hit_rate", - preferred_keys=("avg", "max", "total"), - combine="avg", - ) - ) - if flat["server_gpu_cache_hit_rate"] is None: - flat["server_gpu_cache_hit_rate"] = rate(cached_tokens, prompt_total) - flat["server_overall_cache_hit_rate"] = flat["server_gpu_cache_hit_rate"] + flat["server_gpu_cache_hit_rate"] = None + flat["server_overall_cache_hit_rate"] = None flat["gpu_kv_cache_usage_pct"] = normalize_fraction( gauge_stat( @@ -77,28 +72,14 @@ def populate( combine="max", ) ) - flat["kv_offload_bytes_gpu_to_cpu"] = sum_stat( - metrics, - "trtllm_kv_cache_offload_bytes_total", - preferred_keys=("total", "sum", "max", "avg"), - ) - flat["kv_offload_bytes_cpu_to_gpu"] = sum_stat( - metrics, - "trtllm_kv_cache_onboard_bytes_total", - preferred_keys=("total", "sum", "max", "avg"), - ) - - computed_tokens = None - if prompt_total is not None and cached_tokens is not None: - computed_tokens = max(prompt_total - cached_tokens, 0.0) + flat["kv_offload_bytes_gpu_to_cpu"] = _alias_total(metrics, "trtllm_kv_cache_offload_bytes") + flat["kv_offload_bytes_cpu_to_gpu"] = _alias_total(metrics, "trtllm_kv_cache_onboard_bytes") nested["cache"].update( { - "gpu_cache_hit_rate": flat["server_gpu_cache_hit_rate"], - "overall_cache_hit_rate": flat["server_overall_cache_hit_rate"], - "cached_tokens_by_source": {"device": cached_tokens} - if cached_tokens is not None - else {}, + "gpu_cache_hit_rate": None, + "overall_cache_hit_rate": None, + "cached_tokens_by_source": {}, } ) nested["kv_cache"].update( @@ -118,14 +99,36 @@ def populate( "prompt_total": flat["total_prompt_tokens"], "generation_total": flat["total_generation_tokens"], "prompt_by_source": { - "gpu_cache_hit": cached_tokens, + "gpu_cache_hit": None, "cpu_or_external_cache_hit": None, - "computed": computed_tokens, - "raw": {"device": cached_tokens} if cached_tokens is not None else {}, + "computed": None, + "raw": {}, }, } ) - nested["sources"] = _trtllm_sources(metrics) + reuse, paired = _reuse_rate(metrics, context) + nested["validity"]["cache_rate"] = reuse + nested["sources"] = _trtllm_sources(metrics, paired, context.get("disagg")) + if reuse["published"]: + prompt = reuse["prompt_tokens"] + cached = sum(item[1] for item in paired.values()) + if flat["total_prompt_tokens"] is None: + flat["total_prompt_tokens"] = nested["tokens"]["prompt_total"] = prompt + flat["server_overall_cache_hit_rate"] = cached / prompt + nested["cache"].update( + { + "overall_cache_hit_rate": cached / prompt, + "cached_tokens_by_source": {"combined": cached}, + } + ) + nested["tokens"]["prompt_by_source"].update( + { + "computed": prompt - cached + if nested["tokens"]["prompt_total"] == prompt + else None, + "raw": {"combined": cached}, + } + ) def gpu_kv_capacity_tokens( self, @@ -150,26 +153,241 @@ def gpu_kv_capacity_tokens( return counter_int(max_blocks * tokens_per_block) -def _first_counter_total( +def _reuse_rate( + metrics: dict[str, dict[str, Any]], context: dict[str, Any] +) -> tuple[dict[str, Any], dict[str, tuple[int, int]]]: + result = { + "status": "UNKNOWN", + "published": False, + "eligible_exact_actual": False, + "reasons": [], + "role_scope": None, + "prompt_tokens": None, + } + disaggregated = context.get("disagg") + roles = _endpoint_role_sets(metrics, disaggregated, context.get("endpoint_roles")) + observed = { + _series_role(series, roles, disaggregated) + for name in (PROMPT_TOKENS, CACHED_TOKENS) + for series in metric_series(metrics, [name, f"{name}_total"]) + } + scope = "prefill" if disaggregated is True else "combined" if disaggregated is False else None + if scope is None: + candidates = observed - {"decode"} + scope = next(iter(candidates)) if len(candidates) == 1 else None + if ( + scope not in ("prefill", "combined") + or observed - {scope, "decode"} + or (scope == "combined" and "decode" in observed) + ): + result["reasons"] = ["role_or_deployment_unresolved"] + return result, {} + if any( + len(roles.get(str(series.get("endpoint_url", "")), set())) > 1 + for name in (PROMPT_TOKENS, CACHED_TOKENS) + for series in metric_series(metrics, [name, f"{name}_total"]) + if _series_role(series, roles, disaggregated) == scope + ): + result["reasons"] = ["role_conflict"] + return result, {} + result["role_scope"] = scope + try: + prompt = _counter_domains(metrics, PROMPT_TOKENS, roles, scope, disaggregated) + cached = _counter_domains(metrics, CACHED_TOKENS, roles, scope, disaggregated) + except ValueError as error: + result.update(status="REJECTED", reasons=[str(error)]) + return result, {} + if not prompt or prompt.keys() != cached.keys(): + result.update(status="PARTIAL_COVERAGE", reasons=["missing_counter_pair"]) + return result, {} + error = modern_profile_error(context.get("collector_export") or {}, {key[0] for key in prompt}) + if error: + result["reasons"] = [error] + return result, {} + if any(cached[key] > value for key, value in prompt.items()): + result.update(status="REJECTED", reasons=["cached_exceeds_prompt"]) + return result, {} + total = sum(prompt.values()) + if total <= 0: + result["reasons"] = ["no_prompt_tokens"] + return result, {} + paired: dict[str, tuple[int, int]] = {} + for key, value in prompt.items(): + full, hit = paired.get(key[0], (0, 0)) + paired[key[0]] = (full + value, hit + cached[key]) + result.update( + status="CONDITIONAL_INTERVAL", + published=True, + reasons=list(UNBOUND_REASONS), + prompt_tokens=total, + ) + return result, paired + + +def _counter_domains( metrics: dict[str, dict[str, Any]], - metric_names: list[str], + name: str, + roles: dict[str, set[str]], + scope: str, + disaggregated: bool | None, +) -> dict[tuple, int]: + totals: dict[tuple, int] = {} + for alias in (name, f"{name}_total"): + entry = metrics.get(alias) + if entry is None: + continue + if ( + not isinstance(entry, dict) + or entry.get("type") != "counter" + or not isinstance(entry.get("series"), list) + ): + raise ValueError("invalid_counter_series") + seen = set() + for series in entry["series"]: + if not isinstance(series, dict): + raise ValueError("invalid_counter_series") + if _series_role(series, roles, disaggregated) != scope: + continue + endpoint = series.get("endpoint_url") + labels = series.get("labels") + labels = {} if labels is None else labels + if ( + not isinstance(endpoint, str) + or not endpoint + or not isinstance(labels, dict) + or any( + not isinstance(key, str) or not isinstance(value, str) + for key, value in labels.items() + ) + ): + raise ValueError("invalid_counter_domain") + key = endpoint, tuple(sorted((k, v) for k, v in labels.items() if k not in ROLE_LABELS)) + stats = series.get("stats") + value = stats.get("total") if isinstance(stats, dict) else None + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or (isinstance(value, float) and not math.isfinite(value)) + or value < 0 + or int(value) != value + ): + raise ValueError("invalid_counter_total") + if key in seen or (key in totals and totals[key] != value): + raise ValueError("counter_alias_conflict") + seen.add(key) + totals[key] = int(value) + return totals + + +def _first_total( + metrics: dict[str, dict[str, Any]], frontend_metric: str, trtllm_metric: str ) -> float | None: - """Return the first non-None counter total among ``metric_names`` in ``metrics``.""" - for metric_name in metric_names: - value = sum_stat( - metrics, - metric_name, - preferred_keys=("total", "sum", "max", "avg"), - ) - if value is not None: - return value - return None + frontend = sum_stat(metrics, frontend_metric, preferred_keys=("total", "sum", "max", "avg")) + if frontend is not None: + return frontend + return _alias_total(metrics, trtllm_metric) -def _trtllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: - """Return per-endpoint token, hit rate and KV usage rows from TRT-LLM ``metrics``, sorted by endpoint.""" +def _alias_total( + metrics: dict[str, dict[str, Any]], + base: str, + series_filter: Callable[[dict[str, Any]], bool] | None = None, +) -> float | None: + totals: dict[tuple, float] = {} + for name in (base, f"{base}_total"): + entry = metrics.get(name) + if entry is None: + continue + if ( + not isinstance(entry, dict) + or entry.get("type") != "counter" + or not isinstance(entry.get("series"), list) + ): + return None + seen = set() + for series in metric_series(metrics, name): + if series_filter is not None and not series_filter(series): + continue + labels = series.get("labels") + labels = {} if labels is None else labels + if not isinstance(labels, dict) or any( + not isinstance(key, str) or not isinstance(value, str) + for key, value in labels.items() + ): + return None + key = str(series.get("endpoint_url", "")), tuple(sorted(labels.items())) + stats = series.get("stats") + value = stats.get("total") if isinstance(stats, dict) else None + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value < 0 + or key in seen + or (key in totals and totals[key] != value) + ): + return None + seen.add(key) + totals[key] = value + return sum(totals.values()) if totals else None + + +def _endpoint_role_sets( + metrics: dict[str, dict[str, Any]], + disaggregated: bool | None = None, + declared: dict[str, str] | None = None, +) -> dict[str, set[str]]: + explicit: dict[str, set[str]] = {} + inferred: dict[str, set[str]] = {} + for entry in metrics.values(): + if not isinstance(entry, dict): + continue + series_list = entry.get("series") + for series in series_list if isinstance(series_list, list) else []: + if not isinstance(series, dict): + continue + endpoint = str(series.get("endpoint_url", "")) + labels = series.get("labels") + if isinstance(labels, dict) and "disaggregation_mode" in labels: + mode = labels["disaggregation_mode"] + role = _normalize_role(mode) if isinstance(mode, str) and mode else "unknown" + explicit.setdefault(endpoint, set()).add(role) + elif isinstance(labels, dict) and "dynamo_component" in labels: + component = labels["dynamo_component"] + role = ( + "unknown" + if not isinstance(component, str) or not component + else "combined" + if component == "backend" and disaggregated is False + else _normalize_role(component) + ) + inferred.setdefault(endpoint, set()).add(role) + roles = {**inferred, **explicit} + for endpoint, role in (declared or {}).items(): + roles.setdefault(endpoint, set()).add(_normalize_role(role)) + return roles + + +def _series_role( + series: dict[str, Any], roles: dict[str, set[str]], disaggregated: bool | None = None +) -> str: + labels = series.get("labels") + if isinstance(labels, dict) and "disaggregation_mode" in labels: + mode = labels["disaggregation_mode"] + return _normalize_role(mode) if isinstance(mode, str) and mode else "unknown" + found = roles.get(str(series.get("endpoint_url", "")), set()) + if len(found) == 1: + return next(iter(found)) + return "combined" if not found and disaggregated is False else "unknown" + + +def _trtllm_sources( + metrics: dict[str, dict[str, Any]], + paired: dict[str, tuple[int, int]], + disaggregated: bool | None, +) -> list[dict[str, Any]]: + """Return per-endpoint token, hit rate and KV usage rows, sorted by endpoint.""" endpoints: set[str] = set() - roles: dict[str, str] = {} for entry in metrics.values(): if not isinstance(entry, dict): continue @@ -177,15 +395,12 @@ def _trtllm_sources(metrics: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: if not isinstance(series_list, list): continue for series in series_list: - if not isinstance(series, dict) or not series.get("endpoint_url"): - continue - endpoint = str(series["endpoint_url"]) - endpoints.add(endpoint) - mode = label_value(series, "disaggregation_mode") or label_value( - series, "dynamo_component" - ) - if mode: - roles[endpoint] = _normalize_role(mode) + if isinstance(series, dict) and series.get("endpoint_url"): + endpoints.add(str(series["endpoint_url"])) + roles = { + endpoint: next(iter(found)) if len(found) == 1 else "unknown" + for endpoint, found in _endpoint_role_sets(metrics, disaggregated).items() + } sources: list[dict[str, Any]] = [] for endpoint in sorted(endpoints): @@ -194,21 +409,8 @@ def series_filter(series: dict, endpoint: str = endpoint) -> bool: """Return whether ``series`` was scraped from ``endpoint``.""" return str(series.get("endpoint_url", "")) == endpoint - prompt_tokens = sum_stat( - metrics, - "trtllm_prompt_tokens_total", - series_filter=series_filter, - ) - generation_tokens = sum_stat( - metrics, - "trtllm_generation_tokens_total", - series_filter=series_filter, - ) - cached_tokens = sum_stat( - metrics, - "trtllm_prompt_cached_tokens_total", - series_filter=series_filter, - ) + prompt_tokens = _alias_total(metrics, PROMPT_TOKENS, series_filter) + generation_tokens = _alias_total(metrics, GENERATION_TOKENS, series_filter) hit_rate = normalize_fraction( gauge_stat( metrics, @@ -217,8 +419,9 @@ def series_filter(series: dict, endpoint: str = endpoint) -> bool: series_filter=series_filter, ) ) - if hit_rate is None: - hit_rate = rate(cached_tokens, prompt_tokens) + if hit_rate is None and endpoint in paired: + prompt, cached = paired[endpoint] + hit_rate = rate(cached, prompt) sources.append( { "id": f"{roles.get(endpoint, 'worker')}|{endpoint}", @@ -245,6 +448,6 @@ def _normalize_role(mode: str) -> str: return "prefill" if lowered in ("decode", "generation", "gen", "backend"): return "decode" - if lowered in ("aggregated", "prefill_and_decode", "agg"): + if lowered in ("aggregated", "prefill_and_decode", "agg", "combined", "unified"): return "combined" return lowered diff --git a/inferencex-e2e/infx/results/agentic/server_metrics.py b/inferencex-e2e/infx/results/agentic/server_metrics.py index 05dbe94bc2..a4b860ed5a 100644 --- a/inferencex-e2e/infx/results/agentic/server_metrics.py +++ b/inferencex-e2e/infx/results/agentic/server_metrics.py @@ -79,6 +79,63 @@ def endpoint_roles_from_env(env: Mapping[str, str]) -> dict[str, str] | None: return roles or None +def _integer_total(values: list[Any]) -> int | None: + if not values or any( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or (isinstance(value, float) and not math.isfinite(value)) + or value < 0 + or int(value) != value + for value in values + ): + return None + return sum(int(value) for value in values) + + +def _disaggregated_prompt_total( + metrics: dict[str, dict[str, Any]], records: list[dict[str, Any]], nested: dict[str, Any] +) -> int | None: + values = [] + for record in records: + metric = record.get("metrics", {}).get("input_sequence_length") + values.append(metric.get("value") if isinstance(metric, dict) else metric) + prompt = _integer_total(values) + if prompt is not None: + return prompt + frontend = metrics.get("dynamo_frontend_input_sequence_tokens") + if isinstance(frontend, dict): + series = frontend.get("series") + key = {"counter": "total", "histogram": "sum"}.get(frontend.get("type")) + if isinstance(series, list) and key is not None: + prompt = _integer_total( + [ + item.get("stats", {}).get(key) + if isinstance(item, dict) and isinstance(item.get("stats"), dict) + else None + for item in series + ] + ) + if prompt is not None: + return prompt + reuse = nested["validity"]["cache_rate"] + if reuse.get("published") and reuse.get("role_scope") == "prefill": + return _integer_total([reuse.get("prompt_tokens")]) + return None + + +def _apply_disaggregated_prompt_total( + metrics: dict[str, dict[str, Any]], + records: list[dict[str, Any]], + flat: dict[str, Any], + nested: dict[str, Any], +) -> None: + prompt = _disaggregated_prompt_total(metrics, records, nested) + if prompt != nested["tokens"]["prompt_total"]: + for field in ("gpu_cache_hit", "cpu_or_external_cache_hit", "computed"): + nested["tokens"]["prompt_by_source"][field] = None + flat["total_prompt_tokens"] = nested["tokens"]["prompt_total"] = prompt + + def compute_server_metrics( server_metrics: dict[str, Any], *, @@ -89,6 +146,7 @@ def compute_server_metrics( endpoint_roles: Mapping[str, str] | None = None, request_scope: Mapping[str, Any] | None = None, deployment: Mapping[str, Any] | None = None, + disaggregated: bool | None = None, ) -> tuple[dict[str, Any], dict[str, Any], list[str]]: """Return flat and nested server metrics plus warnings from aiperf's ``server_metrics`` export. @@ -98,6 +156,11 @@ def compute_server_metrics( flat = SERVER_CACHE_FLAT_FIELDS.copy() warnings: list[str] = [] nested = empty_server_metrics(bool(metrics), len(metrics)) + source_env = env or {} + deployment = {**deployment_from_env(source_env), **dict(deployment or {})} + if disaggregated is None: + disaggregated = deployment.get("disagg") + nested["validity"]["deployment"] = {"disaggregated": disaggregated} adapter = detect_backend(metrics, framework) if not metrics: @@ -110,6 +173,8 @@ def compute_server_metrics( nested["tokens"]["prompt_total"] = flat["total_prompt_tokens"] nested["tokens"]["generation_total"] = flat["total_generation_tokens"] nested["tokens"]["requests_completed"] = flat["total_requests_completed"] + if disaggregated is True and (adapter is None or adapter.name != "atom"): + _apply_disaggregated_prompt_total(metrics, records, flat, nested) nested["cache_validity"] = summarize_cache_validity(nested) return flat, nested, warnings @@ -121,7 +186,6 @@ def compute_server_metrics( ) nested["adapter"] = adapter.name if getattr(adapter, "accepts_context", False): - source_env = env or {} adapter.populate( metrics, flat, @@ -132,8 +196,8 @@ def compute_server_metrics( if endpoint_roles is not None else endpoint_roles_from_env(source_env), "request_scope": request_scope, - **deployment_from_env(source_env), - **dict(deployment or {}), + **deployment, + "disagg": disaggregated, "collector_export": server_metrics, "warnings": warnings, }, @@ -158,5 +222,8 @@ def compute_server_metrics( nested["tokens"]["generation_total"] = flat["total_generation_tokens"] nested["tokens"]["requests_completed"] = flat["total_requests_completed"] + if disaggregated is True and adapter.name != "atom": + _apply_disaggregated_prompt_total(metrics, records, flat, nested) + nested["cache_validity"] = summarize_cache_validity(nested) return flat, nested, warnings diff --git a/inferencex-e2e/infx/tests/results/agentic/test_disaggregated_prompt_totals.py b/inferencex-e2e/infx/tests/results/agentic/test_disaggregated_prompt_totals.py new file mode 100644 index 0000000000..970cf0cb5b --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_disaggregated_prompt_totals.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import pytest + +from infx.results.agentic.server_metrics import compute_server_metrics + + +def _counter(value: float) -> dict: + return { + "type": "counter", + "series": [ + { + "endpoint_url": "http://10.0.0.5:8000/metrics", + "labels": {}, + "stats": {"total": value}, + } + ], + } + + +class TestDisaggregatedPromptTotals: + @pytest.mark.parametrize( + "framework, metric", + [ + ("vllm", "vllm:prompt_tokens"), + ("sglang", "sglang:prompt_tokens"), + ("trtllm", "trtllm_prompt_tokens"), + ], + ) + def test_complete_profile_records_replace_duplicate_worker_prompts(self, framework, metric): + export = {"metrics": {metric: _counter(2000.0)}} + records = [{"metrics": {"input_sequence_length": {"value": n}}} for n in (800, 200)] + flat, nested, _ = compute_server_metrics( + export, framework=framework, records=records, disaggregated=True + ) + + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 1000 + assert nested["cache_validity"]["pricing_eligible"] is False + + @pytest.mark.parametrize("value", [None, True, -1, 1.5, float("inf")]) + def test_partial_or_invalid_profile_records_do_not_supply_a_total(self, value): + records = [{"metrics": {"input_sequence_length": n}} for n in (800, value)] + flat, nested, _ = compute_server_metrics( + {"metrics": {"vllm:prompt_tokens": _counter(2000.0)}}, + framework="vllm", + records=records, + disaggregated=True, + ) + + assert flat["total_prompt_tokens"] is None + assert nested["tokens"]["prompt_total"] is None + + def test_missing_metrics_does_not_publish_a_partial_client_total(self): + records = [{"metrics": {"input_sequence_length": 800}}, {"metrics": {}}] + flat, nested, _ = compute_server_metrics( + {}, framework="vllm", records=records, disaggregated=True + ) + + assert flat["total_prompt_tokens"] is None + assert nested["tokens"]["prompt_total"] is None + assert nested["cache_validity"]["status_class"] == "MISSING" + + @pytest.mark.parametrize("totals", [None, [1234, None], [-10, 1244], [True], [float("inf")]]) + def test_incomplete_or_invalid_frontend_cannot_supply_a_prompt_total(self, totals): + counter = _counter(0) + if totals is None: + counter = [] + else: + counter["series"] = [{"stats": {"total": value}} for value in totals] + flat, nested, _ = compute_server_metrics( + {"metrics": {"dynamo_frontend_input_sequence_tokens": counter}}, + framework="dynamo-trt", + records=[], + disaggregated=True, + ) + + assert flat["total_prompt_tokens"] is None + assert nested["tokens"]["prompt_total"] is None + + @pytest.mark.parametrize("metric_type, key", [("counter", "total"), ("histogram", "sum")]) + def test_frontend_total_remains_available_without_client_usage(self, metric_type, key): + export = { + "metrics": { + name: {"type": metric_type, "series": [{"stats": {key: value, "count": 10}}]} + for name, value in ( + ("dynamo_frontend_input_sequence_tokens", 1000.0), + ("dynamo_frontend_cached_tokens", 800.0), + ) + } + } + flat, nested, _ = compute_server_metrics( + export, framework="dynamo-vllm", records=[], disaggregated=True + ) + + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 1000 + assert flat["server_overall_cache_hit_rate"] == pytest.approx(0.8) + assert nested["cache"]["frontend_cache_hit_rate"] == pytest.approx(0.8) + + def test_replacing_worker_scope_clears_normalized_tiers_and_keeps_raw_diagnostics(self): + export = { + "metrics": { + "sglang:prompt_tokens": { + "type": "counter", + "series": [ + {"endpoint_url": endpoint, "stats": {"total": 1000}} + for endpoint in ("http://prefill/metrics", "http://decode/metrics") + ], + }, + "sglang:cached_tokens": { + "type": "counter", + "series": [ + { + "endpoint_url": endpoint, + "labels": {"cache_source": source}, + "stats": {"total": total}, + } + for source, total in (("device", 900), ("host", 100)) + for endpoint in ("http://prefill/metrics", "http://decode/metrics") + ], + }, + } + } + flat, nested, _ = compute_server_metrics( + export, + framework="sglang", + records=[{"metrics": {"input_sequence_length": 1000}}], + disaggregated=True, + ) + + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 1000 + assert nested["tokens"]["prompt_by_source"] == { + "gpu_cache_hit": None, + "cpu_or_external_cache_hit": None, + "computed": None, + "raw": {"device": 1800, "host": 200}, + } + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(1.0) + assert nested["cache_validity"]["pricing_eligible"] is False 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 a32fab31e0..6ced7b401c 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 @@ -76,6 +76,7 @@ "adapter", "metric_count", "cache_validity", + "validity", "cache", "kv_cache", "kv_offload", @@ -1562,8 +1563,15 @@ def test_processor_normalizes_trtllm_server_metrics(tmp_path: Path): _assert_stable_server_metrics_schema(agg) assert agg["server_metrics"]["tokens"]["prompt_total"] == 1000 assert agg["server_metrics"]["tokens"]["generation_total"] == 200 - assert agg["server_metrics"]["cache"]["gpu_cache_hit_rate"] == pytest.approx(0.5) - assert agg["server_metrics"]["cache"]["overall_cache_hit_rate"] == pytest.approx(0.5) + assert agg["server_metrics"]["cache"]["gpu_cache_hit_rate"] is None + assert agg["server_metrics"]["cache"]["overall_cache_hit_rate"] is None + validity = agg["server_metrics"]["validity"] + assert validity["cache_rate"]["status"] == "UNKNOWN" + assert agg["server_metrics"]["cache_validity"]["pricing_eligible"] is False + assert {source["prefix_cache_hit_rate"] for source in agg["server_metrics"]["sources"]} == { + 0.4, + 0.6, + } assert agg["server_metrics"]["kv_cache"]["gpu_usage_pct"] == pytest.approx(0.8) assert agg["server_metrics"]["kv_cache"]["cpu_usage_pct"] == pytest.approx(0.25) assert agg["server_metrics"]["kv_cache"]["gpu_total_tokens"] == 19_200 diff --git a/inferencex-e2e/infx/tests/results/agentic/test_trtllm_cache_counters.py b/inferencex-e2e/infx/tests/results/agentic/test_trtllm_cache_counters.py new file mode 100644 index 0000000000..18797d7cc8 --- /dev/null +++ b/inferencex-e2e/infx/tests/results/agentic/test_trtllm_cache_counters.py @@ -0,0 +1,260 @@ +from __future__ import annotations + +import copy + +import pytest + +from infx.results.agentic.server_metrics import compute_server_metrics + +PREFILL = "http://prefill/metrics" +DECODE = "http://decode/metrics" +PROMPT = "trtllm_prompt_tokens" +CACHED = "trtllm_prompt_cached_tokens" + + +def _series(endpoint: str, value: float, mode: str = "prefill") -> dict: + return { + "endpoint_url": endpoint, + "labels": {"dynamo_component": "backend", "disaggregation_mode": mode, "model_name": "m"}, + "stats": {"total": value}, + } + + +def _pair(cached: float = 600, suffix: str = "", mode: str = "prefill") -> dict: + return { + "schema_version": "1.1", + "aiperf_version": "1.0.12", + "metrics_phase": "profiling", + "summary": { + "phase_time_ranges": {"profiling": {"start_ns": 1, "end_ns": 100}}, + "endpoints_configured": [PREFILL, DECODE], + "endpoints_successful": [PREFILL, DECODE], + }, + "metrics": { + name + suffix: {"type": "counter", "series": [_series(PREFILL, value, mode)]} + for name, value in ((PROMPT, 1000), (CACHED, cached)) + }, + } + + +def _run(export: dict, disaggregated: bool | None = True, **kwargs) -> dict: + return compute_server_metrics( + export, framework="dynamo-trt", records=[], disaggregated=disaggregated, **kwargs + )[1] + + +@pytest.mark.parametrize("suffix", ["", "_total", "both"]) +def test_prefill_counter_aliases_exclude_decode_and_publish_only_combined_interval(suffix): + export = _pair(suffix="_total" if suffix == "_total" else "") + for name, entry in list(export["metrics"].items()): + entry["series"].append(_series(DECODE, 1000, "decode")) + if suffix == "both": + export["metrics"][name + "_total"] = copy.deepcopy(entry) + nested = _run(export) + + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + assert nested["cache"]["gpu_cache_hit_rate"] is None + assert nested["cache"]["cpu_cache_hit_rate"] is None + assert nested["cache"]["cached_tokens_by_source"] == {"combined": 600} + assert nested["tokens"]["prompt_total"] == 1000 + assert nested["tokens"]["prompt_by_source"]["computed"] == 400 + core = nested["cache_validity"] + assert (core["status_class"], core["basis"], core["published"], core["pricing_eligible"]) == ( + "VALID_INTERVAL", + "combined", + True, + False, + ) + source = next(source for source in nested["sources"] if source["role"] == "prefill") + assert source["prefix_cache_hit_rate"] == pytest.approx(0.6) + + +@pytest.mark.parametrize("cached", [0, 1000]) +def test_zero_and_full_reuse_remain_publishable(cached): + nested = _run(_pair(cached)) + assert nested["cache"]["overall_cache_hit_rate"] == (0 if cached == 0 else 1) + assert nested["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize("labelled", [False, True]) +def test_aggregated_deployment_accepts_unlabelled_or_explicit_unified_counter(labelled): + export = _pair(mode="aggregated") + if not labelled: + for entry in export["metrics"].values(): + entry["series"][0]["labels"] = None + nested = _run(export, False) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + + +@pytest.mark.parametrize("bad", [None, True, -1, 0.5, float("nan"), float("inf"), "600"]) +def test_invalid_counter_total_blocks_headline_and_source_fallback(bad): + nested = _run(_pair(bad)) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache_validity"]["pricing_eligible"] is False + assert nested["sources"][0]["prefix_cache_hit_rate"] is None + + +@pytest.mark.parametrize( + "fault", ["unpaired", "rank", "duplicate", "alias", "bounds", "labels", "type", "empty"] +) +def test_counter_domains_must_pair_without_conflicts_or_compensating_totals(fault): + export = _pair() + prompt = export["metrics"][PROMPT]["series"] + cached = export["metrics"][CACHED]["series"] + if fault == "unpaired": + cached.append(_series(DECODE, 1)) + elif fault == "rank": + cached[0]["labels"]["rank"] = "1" + elif fault == "duplicate": + cached.append(copy.deepcopy(cached[0])) + elif fault == "alias": + export["metrics"][CACHED + "_total"] = { + "type": "counter", + "series": [_series(PREFILL, 601)], + } + elif fault == "bounds": + prompt[0]["stats"]["total"] = 100 + prompt.append(_series(DECODE, 1000)) + cached.append(_series(DECODE, 0)) + elif fault == "labels": + cached[0]["labels"] = [] + elif fault == "type": + export["metrics"][CACHED]["type"] = "gauge" + else: + prompt[0]["stats"]["total"] = cached[0]["stats"]["total"] = 0 + nested = _run(export) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert all(source["prefix_cache_hit_rate"] is None for source in nested["sources"]) + + +@pytest.mark.parametrize("fault", ["unknown", "decode_only", "conflict", "deployment", "declared"]) +def test_unproven_or_contradictory_roles_do_not_publish(fault): + export = _pair() + kwargs = {} + disaggregated = None + if fault == "unknown": + for entry in export["metrics"].values(): + entry["series"].append({**_series(DECODE, 1), "labels": {}}) + elif fault == "decode_only": + for entry in export["metrics"].values(): + entry["series"][0]["labels"]["disaggregation_mode"] = "decode" + elif fault == "conflict": + export["metrics"]["trtllm_requests"] = {"series": [_series(PREFILL, 1, "decode")]} + elif fault == "deployment": + disaggregated = False + else: + kwargs["endpoint_roles"] = {PREFILL: "decode"} + nested = _run(export, disaggregated, **kwargs) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache_validity"]["published"] is False + + +@pytest.mark.parametrize("fault", ["old", "phase", "window", "unscraped", "unconfigured"]) +def test_only_modern_profiling_totals_with_accounted_endpoint_coverage_publish(fault): + export = _pair() + if fault == "old": + export["aiperf_version"] = "1.0.11" + elif fault == "phase": + export["metrics_phase"] = "warmup" + elif fault == "window": + export["summary"]["phase_time_ranges"]["profiling"]["end_ns"] = 0 + else: + key = "endpoints_successful" if fault == "unscraped" else "endpoints_configured" + export["summary"][key].remove(PREFILL) + nested = _run(export) + assert nested["cache"]["overall_cache_hit_rate"] is None + assert nested["cache_validity"]["pricing_eligible"] is False + + +def test_gauge_remains_source_diagnostic_and_never_overrides_counter_rate(): + export = _pair() + export["metrics"]["trtllm_kv_cache_hit_rate"] = { + "type": "gauge", + "series": [{"endpoint_url": PREFILL, "stats": {"avg": 0.9}}], + } + nested = _run(export) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + assert nested["sources"][0]["prefix_cache_hit_rate"] == pytest.approx(0.9) + + +def test_idle_domain_and_series_order_do_not_change_weighted_rate(): + export = _pair() + for name, value in ((PROMPT, 3000), (CACHED, 900)): + series = export["metrics"][name]["series"] + series.append( + {**_series(PREFILL, value), "labels": {"rank": "1", "disaggregation_mode": "prefill"}} + ) + series.append(_series(DECODE, 0)) + export["metrics"][CACHED]["series"].reverse() + nested = _run(export) + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.375) + assert nested["tokens"]["prompt_total"] == 4000 + + +@pytest.mark.parametrize("label", ["disaggregation_mode", "dynamo_component"]) +@pytest.mark.parametrize("mode", [None, "", "router", "unexpected"]) +def test_explicit_invalid_role_cannot_fall_back_to_aggregated_deployment(label, mode): + export = _pair() + for entry in export["metrics"].values(): + entry["series"][0]["labels"] = {label: mode} + assert _run(export, False)["cache"]["overall_cache_hit_rate"] is None + + +def test_conflicts_on_excluded_decode_do_not_block_prefill_measurement(): + export = _pair() + for entry in export["metrics"].values(): + entry["series"].append(_series(DECODE, 1000, "decode")) + export["metrics"]["trtllm_other_metric"] = {"series": [_series(DECODE, 1, "aggregated")]} + assert _run(export)["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + + +@pytest.mark.parametrize("fault", ["shape", "missing", "partial", "type"]) +def test_incomplete_frontend_does_not_hide_the_validated_prefill_total(fault): + export = _pair() + frontend = {"type": "counter", "series": [{"stats": {"total": None}}]} + if fault == "shape": + frontend = [] + elif fault == "partial": + frontend["series"].append({"stats": {"total": 1234}}) + elif fault == "type": + frontend["type"] = "gauge" + export["metrics"]["dynamo_frontend_input_sequence_tokens"] = frontend + flat, nested, _ = compute_server_metrics( + export, framework="dynamo-trt", records=[], disaggregated=True + ) + assert flat["total_prompt_tokens"] == nested["tokens"]["prompt_total"] == 1000 + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.6) + assert nested["cache_validity"]["pricing_eligible"] is False + + +@pytest.mark.parametrize("overlap", ["none", "identical", "conflicting"]) +def test_alias_totals_combine_disjoint_domains_without_counting_aliases_twice(overlap): + export = _pair() + other = "http://prefill-2/metrics" + for key in ("endpoints_configured", "endpoints_successful"): + export["summary"][key].append(other) + for name, first, second in ( + (PROMPT, 1000, 3000), + (CACHED, 600, 900), + ("trtllm_generation_tokens", 200, 600), + ("trtllm_kv_cache_offload_bytes", 100, 300), + ): + export["metrics"][name] = {"type": "counter", "series": [_series(PREFILL, first)]} + export["metrics"][name + "_total"] = { + "type": "counter", + "series": [_series(other, second)], + } + if overlap != "none": + export["metrics"][name + "_total"]["series"].append(_series(PREFILL, first)) + if overlap == "conflicting": + export["metrics"]["trtllm_generation_tokens_total"]["series"][-1]["stats"]["total"] = 201 + flat, nested, _ = compute_server_metrics( + export, framework="dynamo-trt", records=[], disaggregated=True + ) + assert nested["tokens"]["prompt_total"] == 4000 + assert nested["cache"]["overall_cache_hit_rate"] == pytest.approx(0.375) + assert flat["total_generation_tokens"] == (None if overlap == "conflicting" else 800) + assert flat["kv_offload_bytes_gpu_to_cpu"] == 400 + source = next(row for row in nested["sources"] if row["id"].endswith(other)) + assert source["prompt_tokens"] == 3000 + assert source["generation_tokens"] == 600