diff --git a/ruff-legacy-baseline.json b/ruff-legacy-baseline.json index 9cef922462e5..571197795369 100644 --- a/ruff-legacy-baseline.json +++ b/ruff-legacy-baseline.json @@ -1,8 +1,8 @@ { "_meta": { "generated_by": "scripts/legacy_utils.py lint-update-violations", - "total_violations": 5588, - "total_files": 500 + "total_violations": 5341, + "total_files": 493 }, ".github/scripts/label_community_user.py": { "D212": 1 @@ -485,7 +485,7 @@ "tensorrt_llm/__init__.py": { "D202": 1, "D212": 1, - "E402": 28 + "E402": 27 }, "tensorrt_llm/_torch/attention_backend/interface.py": { "F821": 2 @@ -933,7 +933,7 @@ "D205": 11, "D208": 2, "D210": 7, - "D212": 36, + "D212": 37, "D300": 8, "D415": 2 }, @@ -992,12 +992,6 @@ "D300": 1, "D415": 4 }, - "tensorrt_llm/metrics/__init__.py": { - "F405": 3 - }, - "tensorrt_llm/metrics/collector.py": { - "D212": 3 - }, "tensorrt_llm/models/baichuan/config.py": { "F821": 2 }, @@ -1581,7 +1575,8 @@ }, "tensorrt_llm/serve/openai_server.py": { "D205": 1, - "D212": 2 + "D212": 2, + "F821": 2 }, "tensorrt_llm/serve/responses_utils.py": { "D200": 2, @@ -1686,10 +1681,6 @@ "D205": 1, "D212": 1 }, - "tests/integration/defs/accuracy/test_cli_flow.py": { - "D300": 1, - "D415": 1 - }, "tests/integration/defs/accuracy/test_disaggregated_serving.py": { "D212": 1, "F601": 1 @@ -1705,10 +1696,9 @@ "F811": 2 }, "tests/integration/defs/accuracy/test_llm_api_pytorch.py": { - "D205": 3, - "D212": 4, + "D212": 1, "D300": 3, - "D415": 5, + "D415": 3, "E402": 7 }, "tests/integration/defs/common.py": { @@ -1770,34 +1760,26 @@ "D415": 1 }, "tests/integration/defs/examples/test_bindings.py": { - "D300": 2, - "D415": 2 + "D300": 1, + "D415": 1 }, "tests/integration/defs/examples/test_chatglm.py": { "D300": 1 }, "tests/integration/defs/examples/test_commandr.py": { - "D300": 2 - }, - "tests/integration/defs/examples/test_draft_target_model.py": { - "D300": 2, - "D415": 2 + "D300": 1 }, "tests/integration/defs/examples/test_gemma.py": { "D202": 1, - "D300": 7, - "D403": 6, - "D415": 6 + "D300": 6, + "D403": 5, + "D415": 5 }, "tests/integration/defs/examples/test_gpt.py": { - "D202": 5, - "D205": 2, - "D209": 1, - "D210": 1, - "D212": 1, - "D300": 22, + "D202": 3, + "D300": 5, "D403": 1, - "D415": 23 + "D415": 4 }, "tests/integration/defs/examples/test_granite.py": { "D202": 1, @@ -1808,16 +1790,10 @@ "D415": 1 }, "tests/integration/defs/examples/test_llama.py": { - "D200": 1, - "D202": 6, - "D205": 6, - "D208": 10, - "D209": 1, - "D210": 1, - "D212": 6, - "D300": 30, - "D403": 12, - "D415": 26, + "D202": 2, + "D300": 6, + "D403": 4, + "D415": 4, "E712": 1 }, "tests/integration/defs/examples/test_mamba.py": { @@ -1825,47 +1801,39 @@ "D415": 2 }, "tests/integration/defs/examples/test_mistral.py": { - "D202": 2, - "D300": 2, - "D403": 1, - "D415": 1 + "D202": 1 }, "tests/integration/defs/examples/test_mixtral.py": { - "D300": 3, - "D403": 2, - "D415": 1 + "D300": 1, + "D403": 1 }, "tests/integration/defs/examples/test_multimodal.py": { "D202": 1, "D300": 2, "D415": 2 }, - "tests/integration/defs/examples/test_nemotron_nas.py": { - "D202": 3 - }, "tests/integration/defs/examples/test_openai.py": { "D300": 3, "D403": 1, "D415": 3 }, "tests/integration/defs/examples/test_phi.py": { - "D202": 3, - "D300": 6, + "D202": 1, + "D300": 3, "D403": 1, "D415": 3 }, "tests/integration/defs/examples/test_qwen.py": { "D202": 1, - "D300": 14, - "D403": 3, - "D415": 7 + "D300": 2, + "D403": 1 }, "tests/integration/defs/examples/test_qwen2audio.py": { "D300": 2, "D415": 1 }, "tests/integration/defs/examples/test_qwenvl.py": { - "D300": 2, + "D300": 1, "D415": 1 }, "tests/integration/defs/examples/test_recurrentgemma.py": { @@ -1909,12 +1877,9 @@ "D212": 3 }, "tests/integration/defs/perf/open_search_db_utils.py": { - "D200": 7, - "D205": 3, - "D212": 10, - "D415": 5, - "E402": 1, - "E712": 1 + "D200": 2, + "D212": 4, + "E402": 1 }, "tests/integration/defs/perf/pytorch_model_config.py": { "D200": 1, @@ -1978,14 +1943,14 @@ "F821": 2 }, "tests/integration/defs/test_e2e.py": { - "D200": 6, - "D202": 4, + "D200": 5, + "D202": 3, "D205": 3, "D210": 1, - "D212": 9, - "D300": 10, - "D403": 5, - "D415": 13, + "D212": 8, + "D300": 9, + "D403": 4, + "D415": 10, "F811": 2 }, "tests/integration/defs/test_list_parser.py": { @@ -2003,11 +1968,9 @@ }, "tests/integration/defs/test_mlpf_results.py": { "D200": 1, - "D202": 1, "D205": 1, "D208": 4, "D212": 2, - "D300": 1, "D415": 2 }, "tests/integration/defs/triton_server/common.py": { @@ -2105,19 +2068,6 @@ "tests/scripts/allreduce_perf/allreduce_perf_viz.py": { "D214": 3 }, - "tests/scripts/perf-sanity/parse_benchmark_results.py": { - "D200": 2, - "D205": 1, - "D212": 3, - "D415": 5 - }, - "tests/scripts/perf-sanity/run_benchmark_serve.py": { - "D200": 2, - "D202": 1, - "D212": 2, - "D415": 8, - "F601": 1 - }, "tests/unittest/_torch/misc/test_autotuner.py": { "E731": 1 }, @@ -2199,7 +2149,7 @@ "F811": 1 }, "tests/unittest/llmapi/test_executor.py": { - "D205": 3, + "D205": 5, "D209": 3 }, "tests/unittest/llmapi/test_llm.py": { @@ -2215,9 +2165,7 @@ "tests/unittest/llmapi/test_llm_args.py": { "D200": 1, "D202": 2, - "D205": 3, - "D209": 1, - "D212": 3, + "D212": 2, "E712": 20, "F811": 1 }, diff --git a/tensorrt_llm/executor/result.py b/tensorrt_llm/executor/result.py index 91cb0b77b6aa..e798c2c6b567 100644 --- a/tensorrt_llm/executor/result.py +++ b/tensorrt_llm/executor/result.py @@ -25,6 +25,8 @@ from ..llmapi.tracer import global_tracer from ..llmapi.utils import AsyncQueue, print_traceback_on_error from ..metrics import MetricNames, MetricsCollector, RequestEventTiming +from ..metrics.perf_utils import \ + process_req_perf_metrics as _process_req_perf_metrics from ..sampling_params import LogprobParams, SamplingParams from .utils import ErrorResponse, has_event_loop, is_llm_response @@ -179,6 +181,7 @@ def __init__(self, self._done = False self._aborted = False self.metrics_dict = {} + self.candidate_metrics: list[dict] = [] self.trace_headers: Optional[dict[str, str]] = None # torch backend will use trtllm sampler in beam search mode, but it does not support return logprobs incrementally self.use_trtllm_sampler = sampling_params.use_beam_search and sampling_params.best_of > 1 @@ -393,9 +396,14 @@ def _handle_sequence(self, raise ValueError( f"Unknown finish reason: {finish_reasons[src_idx]}") - # Only record stats and do tracing when the entire request is done + # Record per-candidate metrics as each sequence finishes so that + # GENERATION_TOKENS and TPOT are captured for every candidate when + # sampling_params.n > 1. + if sequence_is_finished: + self.record_stats(output, req_perf_metrics_dict, seq_idx) + + # Tracing is recorded once when the entire request is done. if self._done: - self.record_stats(output, req_perf_metrics_dict) self.do_tracing(output, req_perf_metrics_dict) @print_traceback_on_error @@ -547,12 +555,19 @@ def _handle_response(self, def record_stats(self, output: CompletionOutput, - stats: Optional[dict[str, float]] = None) -> None: + stats: Optional[dict[str, float]] = None, + sequence_index: int = 0) -> None: """Record the stats of the generation result. + Called once per candidate when it finishes. When ``n > 1`` each + candidate has its own timestamps so TPOT and GENERATION_TOKENS are + computed independently per candidate. PROMPT_TOKENS are only recorded + for ``sequence_index == 0`` to avoid double-counting the shared prompt. + Args: output (CompletionOutput): The output of the generation result. stats (Optional[dict[str, float]]): The stats of the generation result. Defaults to None. + sequence_index (int): Index of this candidate (0 for the first / only sequence). Defaults to 0. """ if not stats: return @@ -563,9 +578,16 @@ def record_stats(self, output.finish_reason }) processed_metrics_stat = _process_req_perf_metrics( - stats, len(output.token_ids), self.sampling_params.n > 1) + stats, len(output.token_ids)) if processed_metrics_stat: metrics_stats.update(processed_metrics_stat) + # Record prompt tokens only for the first candidate to avoid + # double-counting the shared prompt across n candidates. + if output.finish_reason and sequence_index == 0: + prompt_token_ids = getattr(self, "prompt_token_ids", None) + if prompt_token_ids is not None and len(prompt_token_ids) > 0: + metrics_stats[MetricNames.PROMPT_TOKENS] = len(prompt_token_ids) + self.candidate_metrics.append(metrics_stats) self.metrics_dict.update(metrics_stats) def do_tracing( @@ -1042,30 +1064,3 @@ def _topk_logprobs(logits: torch.Tensor, top_k: int, return LogProbsResult(prompt=prompt_logprobs, generation=generation_logprobs) - - -def _process_req_perf_metrics( - req_perf_metrics_dict: Optional[dict[str, float]], - output_length: int, - is_multiple_response: bool = False) -> dict[MetricNames, float]: - stat = {} - if not req_perf_metrics_dict: - return stat - ttft = req_perf_metrics_dict.get(RequestEventTiming.FIRST_TOKEN_TIME, 0) - \ - req_perf_metrics_dict.get(RequestEventTiming.ARRIVAL_TIME, 0) - e2e = req_perf_metrics_dict.get(RequestEventTiming.LAST_TOKEN_TIME, 0) - \ - req_perf_metrics_dict.get(RequestEventTiming.ARRIVAL_TIME, 0) - request_queue_time = req_perf_metrics_dict.get(RequestEventTiming.FIRST_SCHEDULED_TIME, 0) - \ - req_perf_metrics_dict.get(RequestEventTiming.ARRIVAL_TIME, 0) - stat = { - MetricNames.TTFT: ttft, - MetricNames.E2E: e2e, - MetricNames.REQUEST_QUEUE_TIME: request_queue_time - } - if output_length > 1 and not is_multiple_response: - tpot = (req_perf_metrics_dict.get( - RequestEventTiming.LAST_TOKEN_TIME, 0) - req_perf_metrics_dict.get( - RequestEventTiming.FIRST_TOKEN_TIME, 0)) / (output_length - 1) - stat.update({MetricNames.TPOT: tpot}) - stat = dict(filter(lambda item: item[1] > 0, stat.items())) - return stat diff --git a/tensorrt_llm/llmapi/__init__.py b/tensorrt_llm/llmapi/__init__.py index 9576d2ba8060..36dd69bc0267 100644 --- a/tensorrt_llm/llmapi/__init__.py +++ b/tensorrt_llm/llmapi/__init__.py @@ -14,11 +14,12 @@ ExtendedRuntimePerfKnobConfig, KvCacheConfig, LlmArgs, LookaheadDecodingConfig, MedusaDecodingConfig, MoeConfig, MTPDecodingConfig, NGramDecodingConfig, - PARDDecodingConfig, RocketSparseAttentionConfig, - SADecodingConfig, SAEnhancerConfig, - SaveHiddenStatesDecodingConfig, SchedulerConfig, - SkipSoftmaxAttentionConfig, TorchCompileConfig, - TorchLlmArgs, TrtLlmArgs, UserProvidedDecodingConfig) + PARDDecodingConfig, PrometheusMetricsConfig, + RocketSparseAttentionConfig, SADecodingConfig, + SAEnhancerConfig, SaveHiddenStatesDecodingConfig, + SchedulerConfig, SkipSoftmaxAttentionConfig, + TorchCompileConfig, TorchLlmArgs, TrtLlmArgs, + UserProvidedDecodingConfig) from .llm_utils import (BuildConfig, KvCacheRetentionConfig, QuantAlgo, QuantConfig) from .mm_encoder import MultimodalEncoder @@ -74,4 +75,5 @@ 'RocketSparseAttentionConfig', 'DeepSeekSparseAttentionConfig', 'SkipSoftmaxAttentionConfig', + 'PrometheusMetricsConfig', ] diff --git a/tensorrt_llm/llmapi/llm_args.py b/tensorrt_llm/llmapi/llm_args.py index cac912bb3e41..9647df388e00 100644 --- a/tensorrt_llm/llmapi/llm_args.py +++ b/tensorrt_llm/llmapi/llm_args.py @@ -1603,6 +1603,87 @@ def supports_backend(self, backend: str) -> bool: return backend == "pytorch" +class PrometheusMetricsConfig(StrictBaseModel): + """ + Configuration for Prometheus metrics collection. + + Groups all Prometheus-related parameters including custom histogram bucket + boundaries for latency metrics. + """ + + e2e_request_latency_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_e2e_request_latency_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + time_to_first_token_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_time_to_first_token_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + time_per_output_token_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_time_per_output_token_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + request_queue_time_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_request_queue_time_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + request_prefill_time_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_request_prefill_time_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + request_decode_time_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_request_decode_time_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + request_inference_time_buckets: Optional[List[float]] = Field( + default=None, + description= + "Custom histogram bucket boundaries (in seconds) for trtllm_request_inference_time_seconds. " + "Defaults to built-in values when unset.", + status="prototype") + + @field_validator( + "e2e_request_latency_buckets", + "time_to_first_token_buckets", + "time_per_output_token_buckets", + "request_queue_time_buckets", + "request_prefill_time_buckets", + "request_decode_time_buckets", + "request_inference_time_buckets", + ) + @classmethod + def validate_histogram_buckets(cls, v: Optional[List[float]], + info) -> Optional[List[float]]: + """Validate that histogram bucket lists are non-empty and strictly increasing.""" + if v is None: + return v + if len(v) == 0: + raise ValueError( + f"{info.field_name} must not be empty when provided.") + if any(a >= b for a, b in zip(v, v[1:])): + raise ValueError( + f"{info.field_name} must be strictly increasing, got {v}.") + return v + + class RayPlacementConfig(StrictBaseModel): """ Configuration for Ray GPU workers placement. @@ -2778,6 +2859,12 @@ class BaseLlmArgs(StrictBaseModel): "The maximum number of requests for perf metrics. Must also set return_perf_metrics to true to get perf metrics.", status="prototype") + prometheus_metrics_config: Optional[PrometheusMetricsConfig] = Field( + default=None, + description="Configuration for Prometheus metrics collection, including " + "custom histogram bucket boundaries.", + status="prototype") + enable_energy_metrics: bool = Field( default=False, description= diff --git a/tensorrt_llm/metrics/__init__.py b/tensorrt_llm/metrics/__init__.py index f68d9f698ace..5df5dde1f439 100644 --- a/tensorrt_llm/metrics/__init__.py +++ b/tensorrt_llm/metrics/__init__.py @@ -1,4 +1,8 @@ -from .collector import * -from .enums import * +from .collector import MetricsCollector +from .enums import MetricNames, RequestEventTiming +from .perf_utils import process_req_perf_metrics -__all__ = ["MetricsCollector", "MetricNames", "RequestEventTiming"] +__all__ = [ + "MetricsCollector", "MetricNames", "RequestEventTiming", + "process_req_perf_metrics" +] diff --git a/tensorrt_llm/metrics/collector.py b/tensorrt_llm/metrics/collector.py index 2215fc734541..f10c622693bd 100644 --- a/tensorrt_llm/metrics/collector.py +++ b/tensorrt_llm/metrics/collector.py @@ -15,15 +15,14 @@ """Utilities for Prometheus Metrics Collection.""" import time -from typing import Dict, Union +from typing import Dict, List, Optional, Union from .enums import MetricNames # Adapted from https://github.com/vllm-project/vllm/blob/v0.10.0rc1/vllm/engine/metrics.py#L30 class MetricsCollector: - """ - Collects and logs metrics from TensorRT-LLM engine stats and request performance metrics to Prometheus. + """Collects and logs metrics from TensorRT-LLM engine stats to Prometheus. Used by OpenAIServer in tensorrt_llm/serve/openai_server.py. @@ -33,30 +32,96 @@ class MetricsCollector: {"model_name": "nemotron-nano-3", "engine_type": "trtllm"} Created Prometheus metrics: - trtllm_request_success_total - trtllm_e2e_request_latency_seconds - trtllm_time_to_first_token_seconds - trtllm_time_per_output_token_seconds - trtllm_request_queue_time_seconds - trtllm_kv_cache_hit_rate - trtllm_kv_cache_reused_blocks_total - trtllm_kv_cache_missed_blocks_total - trtllm_kv_cache_utilization - trtllm_kv_cache_host_utilization - trtllm_kv_cache_iter_reuse_rate - trtllm_kv_cache_iter_reused_blocks_total - trtllm_kv_cache_iter_full_reused_blocks_total - trtllm_kv_cache_iter_partial_reused_blocks_total - trtllm_kv_cache_iter_missed_blocks_total - trtllm_kv_cache_gen_alloc_blocks_total - trtllm_kv_cache_onboard_bytes_total - trtllm_kv_cache_offload_bytes_total - trtllm_kv_cache_intra_device_copy_bytes_total + Per-request metrics: + trtllm_request_success_total + trtllm_e2e_request_latency_seconds + trtllm_time_to_first_token_seconds + trtllm_time_per_output_token_seconds + trtllm_request_queue_time_seconds + trtllm_request_prefill_time_seconds + trtllm_request_decode_time_seconds + trtllm_request_inference_time_seconds + trtllm_prompt_tokens_total + trtllm_generation_tokens_total + + Iteration-level metrics: + trtllm_kv_cache_hit_rate + trtllm_kv_cache_utilization + trtllm_kv_cache_host_utilization + trtllm_kv_cache_iter_reuse_rate + trtllm_kv_cache_reused_blocks_total + trtllm_kv_cache_missed_blocks_total + trtllm_kv_cache_iter_reused_blocks_total + trtllm_kv_cache_iter_full_reused_blocks_total + trtllm_kv_cache_iter_partial_reused_blocks_total + trtllm_kv_cache_iter_missed_blocks_total + trtllm_kv_cache_gen_alloc_blocks_total + trtllm_kv_cache_onboard_bytes_total + trtllm_kv_cache_offload_bytes_total + trtllm_kv_cache_intra_device_copy_bytes_total + trtllm_num_requests_running + trtllm_num_requests_waiting + trtllm_num_requests_completed_total + trtllm_max_num_active_requests + trtllm_iteration_latency_seconds + trtllm_gpu_memory_usage_bytes + trtllm_cpu_memory_usage_bytes + trtllm_pinned_memory_usage_bytes + trtllm_max_batch_size_static + trtllm_max_batch_size_runtime + trtllm_max_num_tokens_runtime + trtllm_kv_cache_max_blocks + trtllm_kv_cache_free_blocks + trtllm_kv_cache_used_blocks + trtllm_kv_cache_tokens_per_block + trtllm_num_context_requests + trtllm_num_generation_requests + trtllm_num_paused_requests + trtllm_num_scheduled_requests + trtllm_total_context_tokens + trtllm_avg_decoded_tokens_per_iter + trtllm_spec_decode_num_draft_tokens_total + trtllm_spec_decode_num_accepted_tokens_total + trtllm_spec_decode_acceptance_length + trtllm_spec_decode_draft_overhead + + Config info metrics (logged once at startup via log_config_info): + trtllm_model_config_info + trtllm_parallel_config_info + trtllm_speculative_config_info + trtllm_kv_cache_config_info """ labelname_finish_reason = "finished_reason" - def __init__(self, labels: Dict[str, str]) -> None: + def __init__( + self, + labels: Dict[str, str], + e2e_request_latency_buckets: Optional[List[float]] = None, + time_to_first_token_buckets: Optional[List[float]] = None, + time_per_output_token_buckets: Optional[List[float]] = None, + request_queue_time_buckets: Optional[List[float]] = None, + request_prefill_time_buckets: Optional[List[float]] = None, + request_decode_time_buckets: Optional[List[float]] = None, + request_inference_time_buckets: Optional[List[float]] = None, + ) -> None: from prometheus_client import Counter, Gauge, Histogram + _bucket_params = { + "e2e_request_latency_buckets": e2e_request_latency_buckets, + "time_to_first_token_buckets": time_to_first_token_buckets, + "time_per_output_token_buckets": time_per_output_token_buckets, + "request_queue_time_buckets": request_queue_time_buckets, + "request_prefill_time_buckets": request_prefill_time_buckets, + "request_decode_time_buckets": request_decode_time_buckets, + "request_inference_time_buckets": request_inference_time_buckets, + } + for name, buckets in _bucket_params.items(): + if buckets is None: + continue + if len(buckets) == 0: + raise ValueError(f"{name} must not be empty when provided.") + if any(a >= b for a, b in zip(buckets, buckets[1:])): + raise ValueError( + f"{name} must be strictly increasing, got {buckets}.") self.last_log_time = time.time() self.labels = labels self.metric_prefix = "trtllm_" @@ -77,7 +142,7 @@ def __init__(self, labels: Dict[str, str]) -> None: self.histogram_e2e_time_request = Histogram( name=self.metric_prefix + "e2e_request_latency_seconds", documentation="Histogram of end to end request latency in seconds.", - buckets=[ + buckets=e2e_request_latency_buckets or [ 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 50.0, 60.0, 120.0, 240.0, 480.0, 960.0, 1920.0, 7680.0 ], @@ -86,7 +151,7 @@ def __init__(self, labels: Dict[str, str]) -> None: self.histogram_time_to_first_token = Histogram( name=self.metric_prefix + "time_to_first_token_seconds", documentation="Histogram of time to first token in seconds.", - buckets=[ + buckets=time_to_first_token_buckets or [ 0.001, 0.005, 0.01, 0.02, 0.04, 0.06, 0.08, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0, 20.0, 40.0, 80.0, 160.0, 640.0, 2560.0 @@ -96,7 +161,7 @@ def __init__(self, labels: Dict[str, str]) -> None: self.histogram_time_per_output_token = Histogram( name=self.metric_prefix + "time_per_output_token_seconds", documentation="Histogram of time per output token in seconds.", - buckets=[ + buckets=time_per_output_token_buckets or [ 0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.2, 0.3, 0.4, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0, 20.0, 40.0, 80.0 ], @@ -106,12 +171,56 @@ def __init__(self, labels: Dict[str, str]) -> None: name=self.metric_prefix + "request_queue_time_seconds", documentation= "Histogram of time spent in WAITING phase for request.", - buckets=[ + buckets=request_queue_time_buckets or [ + 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, + 40.0, 50.0, 60.0, 120.0, 240.0, 480.0, 960.0, 1920.0, 7680.0 + ], + labelnames=self.labels.keys()) + + self.histogram_prefill_time_request = Histogram( + name=self.metric_prefix + "request_prefill_time_seconds", + documentation= + "Histogram of prefill (context) phase duration in seconds " + "(first_token_time - first_scheduled_time).", + buckets=request_prefill_time_buckets or [ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, + 10.0, 20.0, 40.0, 80.0, 160.0, 640.0, 2560.0 + ], + labelnames=self.labels.keys()) + + self.histogram_decode_time_request = Histogram( + name=self.metric_prefix + "request_decode_time_seconds", + documentation= + "Histogram of decode (generation) phase duration in seconds " + "(last_token_time - first_token_time).", + buckets=request_decode_time_buckets or [ + 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, + 40.0, 50.0, 60.0, 120.0, 240.0, 480.0, 960.0, 1920.0, 7680.0 + ], + labelnames=self.labels.keys()) + + self.histogram_inference_time_request = Histogram( + name=self.metric_prefix + "request_inference_time_seconds", + documentation="Histogram of total inference duration in seconds " + "(last_token_time - first_scheduled_time).", + buckets=request_inference_time_buckets or [ 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 50.0, 60.0, 120.0, 240.0, 480.0, 960.0, 1920.0, 7680.0 ], labelnames=self.labels.keys()) + self.counter_prompt_tokens = Counter( + name=self.metric_prefix + "prompt_tokens_total", + documentation= + "Cumulative number of prompt (input) tokens processed.", + labelnames=self.labels.keys()) + + self.counter_generation_tokens = Counter( + name=self.metric_prefix + "generation_tokens_total", + documentation= + "Cumulative number of generation (output) tokens produced.", + labelnames=self.labels.keys()) + self.kv_cache_hit_rate = Gauge(name=self.metric_prefix + "kv_cache_hit_rate", documentation="KV cache hit rate", @@ -178,6 +287,169 @@ def __init__(self, labels: Dict[str, str]) -> None: "Total bytes copied within GPU (intra-device block copies)", labelnames=self.labels.keys()) + # Queue & load metrics + self.num_requests_running = Gauge( + name=self.metric_prefix + "num_requests_running", + documentation="Number of active requests", + labelnames=self.labels.keys()) + self.num_requests_waiting = Gauge( + name=self.metric_prefix + "num_requests_waiting", + documentation="Number of queued requests", + labelnames=self.labels.keys()) + self.counter_num_requests_completed = Counter( + name=self.metric_prefix + "num_requests_completed_total", + documentation="Total number of completed requests across iterations", + labelnames=self.labels.keys()) + self.max_num_active_requests = Gauge( + name=self.metric_prefix + "max_num_active_requests", + documentation="Maximum number of active requests", + labelnames=self.labels.keys()) + + # Iteration latency + self.iteration_latency_seconds = Gauge( + name=self.metric_prefix + "iteration_latency_seconds", + documentation="Iteration latency in seconds", + labelnames=self.labels.keys()) + + # Memory usage + self.gpu_memory_usage_bytes = Gauge( + name=self.metric_prefix + "gpu_memory_usage_bytes", + documentation="GPU memory usage in bytes", + labelnames=self.labels.keys()) + self.cpu_memory_usage_bytes = Gauge( + name=self.metric_prefix + "cpu_memory_usage_bytes", + documentation="CPU memory usage in bytes", + labelnames=self.labels.keys()) + self.pinned_memory_usage_bytes = Gauge( + name=self.metric_prefix + "pinned_memory_usage_bytes", + documentation="Pinned memory usage in bytes", + labelnames=self.labels.keys()) + + # Batch size + self.max_batch_size_static = Gauge( + name=self.metric_prefix + "max_batch_size_static", + documentation="Static maximum batch size", + labelnames=self.labels.keys()) + self.max_batch_size_runtime = Gauge( + name=self.metric_prefix + "max_batch_size_runtime", + documentation="Runtime maximum batch size", + labelnames=self.labels.keys()) + self.max_num_tokens_runtime = Gauge( + name=self.metric_prefix + "max_num_tokens_runtime", + documentation="Runtime maximum number of tokens", + labelnames=self.labels.keys()) + + # KV cache block metrics + self.kv_cache_max_blocks = Gauge( + name=self.metric_prefix + "kv_cache_max_blocks", + documentation="Maximum number of KV cache blocks", + labelnames=self.labels.keys()) + self.kv_cache_free_blocks = Gauge( + name=self.metric_prefix + "kv_cache_free_blocks", + documentation="Number of free KV cache blocks", + labelnames=self.labels.keys()) + self.kv_cache_used_blocks = Gauge( + name=self.metric_prefix + "kv_cache_used_blocks", + documentation="Number of used KV cache blocks", + labelnames=self.labels.keys()) + self.kv_cache_tokens_per_block = Gauge( + name=self.metric_prefix + "kv_cache_tokens_per_block", + documentation="Number of tokens per KV cache block", + labelnames=self.labels.keys()) + + # Inflight batching metrics + self.num_context_requests = Gauge( + name=self.metric_prefix + "num_context_requests", + documentation="Number of context (prefill) requests", + labelnames=self.labels.keys()) + self.num_generation_requests = Gauge( + name=self.metric_prefix + "num_generation_requests", + documentation="Number of generation (decode) requests", + labelnames=self.labels.keys()) + self.num_paused_requests = Gauge( + name=self.metric_prefix + "num_paused_requests", + documentation="Number of paused requests", + labelnames=self.labels.keys()) + self.num_scheduled_requests = Gauge( + name=self.metric_prefix + "num_scheduled_requests", + documentation="Number of scheduled requests", + labelnames=self.labels.keys()) + self.total_context_tokens = Gauge( + name=self.metric_prefix + "total_context_tokens", + documentation="Total number of context tokens", + labelnames=self.labels.keys()) + self.avg_decoded_tokens_per_iter = Gauge( + name=self.metric_prefix + "avg_decoded_tokens_per_iter", + documentation="Average number of decoded tokens per iteration", + labelnames=self.labels.keys()) + + # Speculative decoding metrics + self.counter_spec_decode_num_draft_tokens = Counter( + name=self.metric_prefix + "spec_decode_num_draft_tokens_total", + documentation="Total number of draft tokens in speculative decoding", + labelnames=self.labels.keys()) + self.counter_spec_decode_num_accepted_tokens = Counter( + name=self.metric_prefix + "spec_decode_num_accepted_tokens_total", + documentation= + "Total number of accepted tokens in speculative decoding", + labelnames=self.labels.keys()) + self.spec_decode_acceptance_length = Gauge( + name=self.metric_prefix + "spec_decode_acceptance_length", + documentation="Acceptance length in speculative decoding", + labelnames=self.labels.keys()) + self.spec_decode_draft_overhead = Gauge( + name=self.metric_prefix + "spec_decode_draft_overhead", + documentation="Draft overhead in speculative decoding", + labelnames=self.labels.keys()) + + def log_config_info( + self, + model_config: Optional[Dict[str, str]] = None, + parallel_config: Optional[Dict[str, str]] = None, + speculative_config: Optional[Dict[str, str]] = None, + kv_cache_config: Optional[Dict[str, str]] = None) -> None: + """Log static configuration as Prometheus info-style gauges (set to 1 with config labels). + + Should be called once at startup. Each config dict's keys become Prometheus labels. + Follows the same pattern as vLLM/SGLang config info metrics. + + Args: + model_config: Model configuration labels (model, dtype, quantization, gpu_type, etc.) + parallel_config: Parallelism configuration labels (tp_size, pp_size, etc.) + speculative_config: Speculative decoding configuration labels (method, draft_model, etc.) + kv_cache_config: KV cache configuration labels (page_size, enable_block_reuse, etc.) + """ + from prometheus_client import Gauge + + if model_config: + info_labels = {**self.labels, **model_config} + gauge = Gauge(name=self.metric_prefix + "model_config_info", + documentation="Model configuration info", + labelnames=info_labels.keys()) + gauge.labels(**info_labels).set(1) + + if parallel_config: + info_labels = {**self.labels, **parallel_config} + gauge = Gauge(name=self.metric_prefix + "parallel_config_info", + documentation="Parallelism configuration info", + labelnames=info_labels.keys()) + gauge.labels(**info_labels).set(1) + + if speculative_config: + info_labels = {**self.labels, **speculative_config} + gauge = Gauge( + name=self.metric_prefix + "speculative_config_info", + documentation="Speculative decoding configuration info", + labelnames=info_labels.keys()) + gauge.labels(**info_labels).set(1) + + if kv_cache_config: + info_labels = {**self.labels, **kv_cache_config} + gauge = Gauge(name=self.metric_prefix + "kv_cache_config_info", + documentation="KV cache configuration info", + labelnames=info_labels.keys()) + gauge.labels(**info_labels).set(1) + def _label_merge(self, labels: Dict[str, str]) -> Dict[str, str]: if labels is None or len(labels) == 0: return self.labels @@ -197,8 +469,7 @@ def _log_gauge(self, gauge, data: Union[int, float]) -> None: gauge.labels(**self.labels).set(data) def log_request_metrics_dict(self, metrics_dict: dict[str, float]) -> None: - """ - Log per-request metrics from TRTLLM engine responses. + """Log per-request metrics from TRTLLM engine responses. This method updates Prometheus metrics including: - counter_request_success @@ -206,6 +477,11 @@ def log_request_metrics_dict(self, metrics_dict: dict[str, float]) -> None: - histogram_time_to_first_token - histogram_time_per_output_token - histogram_queue_time_request + - histogram_prefill_time_request + - histogram_decode_time_request + - histogram_inference_time_request + - counter_prompt_tokens + - counter_generation_tokens Args: metrics_dict: A dictionary containing request metrics with the following expected keys: @@ -215,6 +491,11 @@ def log_request_metrics_dict(self, metrics_dict: dict[str, float]) -> None: - `MetricNames.TTFT` (float): Time to first token in seconds. - `MetricNames.TPOT` (float): Time per output token in seconds. - `MetricNames.REQUEST_QUEUE_TIME` (float): Request queue time in seconds. + - `MetricNames.PREFILL_TIME` (float): Prefill phase duration in seconds. + - `MetricNames.DECODE_TIME` (float): Decode phase duration in seconds. + - `MetricNames.INFERENCE_TIME` (float): Total inference duration in seconds. + - `MetricNames.PROMPT_TOKENS` (int): Number of input tokens. + - `MetricNames.GENERATION_TOKENS` (int): Number of output tokens. Returns: None: Metrics are logged to Prometheus; nothing is returned. @@ -238,42 +519,89 @@ def log_request_metrics_dict(self, metrics_dict: dict[str, float]) -> None: self._log_histogram(self.histogram_time_to_first_token, ttft) if tpot := metrics_dict.get(MetricNames.TPOT, 0): self._log_histogram(self.histogram_time_per_output_token, tpot) - if request_queue_time := metrics_dict.get( - MetricNames.REQUEST_QUEUE_TIME, 0): + if (request_queue_time := metrics_dict.get( + MetricNames.REQUEST_QUEUE_TIME)) is not None: self._log_histogram(self.histogram_queue_time_request, request_queue_time) + if prefill_time := metrics_dict.get(MetricNames.PREFILL_TIME, 0): + self._log_histogram(self.histogram_prefill_time_request, + prefill_time) + if decode_time := metrics_dict.get(MetricNames.DECODE_TIME, 0): + self._log_histogram(self.histogram_decode_time_request, + decode_time) + if inference_time := metrics_dict.get(MetricNames.INFERENCE_TIME, + 0): + self._log_histogram(self.histogram_inference_time_request, + inference_time) + if prompt_tokens := metrics_dict.get(MetricNames.PROMPT_TOKENS, 0): + self._log_counter(self.counter_prompt_tokens, {}, prompt_tokens) + if generation_tokens := metrics_dict.get( + MetricNames.GENERATION_TOKENS, 0): + self._log_counter(self.counter_generation_tokens, {}, + generation_tokens) self.last_log_time = time.time() def log_iteration_stats(self, iteration_stats: dict) -> None: - """ - Log iteration-level statistics from TRTLLM engine. + """Log iteration-level statistics from TRTLLM engine. - This method updates Prometheus metrics including: - - kv_cache_hit_rate - - kv_cache_reused_blocks - - kv_cache_missed_blocks - - kv_cache_utilization + Updates Prometheus gauges/counters for queue load, memory usage, batch sizes, + KV cache blocks, inflight batching, and speculative decoding stats. Args: - iteration_stats: A JSON dict returned from `BaseLLM.get_stats()` containing iteration-level statistics - with the following expected structure: - - "kvCacheStats" (dict): KV cache statistics containing: - - "cacheHitRate" (float): Cache hit rate (0.0 to 1.0). If present (including zero), - the kv_cache_hit_rate gauge is updated. - - "reusedBlocks" (int): Number of KV cache blocks reused (cache hits). - - "missedBlocks" (int): Number of KV cache blocks missed (cache misses). - - "usedNumBlocks" (int): Number of KV cache blocks currently in use. - - "maxNumBlocks" (int): Maximum number of KV cache blocks available. Should always be - non-zero. - - Returns: - None: Metrics are logged to Prometheus; nothing is returned. + iteration_stats: A JSON dict returned from `BaseLLM.get_stats()` containing iteration-level statistics. + Top-level fields: numActiveRequests, numQueuedRequests, numCompletedRequests, + maxNumActiveRequests, iterLatencyMS, gpuMemUsage, cpuMemUsage, pinnedMemUsage, + maxBatchSizeStatic, maxBatchSizeRuntime, maxNumTokensRuntime. + Nested dicts: kvCacheStats, inflightBatchingStats, specDecodingStats. Note: - - Needs to include `enable_iter_perf_stats: true` in LLM args to collect iteration-level stats. - - KV cache utilization is only calculated and logged when both "usedNumBlocks" and - "maxNumBlocks" are present in kvCacheStats and "maxNumBlocks" is non-zero. + - Needs `enable_iter_perf_stats: true` in LLM args to collect iteration-level stats. + - inflightBatchingStats and specDecodingStats are only present when applicable. """ + # Top-level queue & load metrics + if "numActiveRequests" in iteration_stats: + self._log_gauge(self.num_requests_running, + iteration_stats["numActiveRequests"]) + if "numQueuedRequests" in iteration_stats: + self._log_gauge(self.num_requests_waiting, + iteration_stats["numQueuedRequests"]) + if "numCompletedRequests" in iteration_stats: + completed = iteration_stats["numCompletedRequests"] + if completed > 0: + self._log_counter(self.counter_num_requests_completed, {}, + completed) + if "maxNumActiveRequests" in iteration_stats: + self._log_gauge(self.max_num_active_requests, + iteration_stats["maxNumActiveRequests"]) + + # Iteration latency (convert ms to seconds) + if "iterLatencyMS" in iteration_stats: + self._log_gauge(self.iteration_latency_seconds, + iteration_stats["iterLatencyMS"] / 1000.0) + + # Memory usage + if "gpuMemUsage" in iteration_stats: + self._log_gauge(self.gpu_memory_usage_bytes, + iteration_stats["gpuMemUsage"]) + if "cpuMemUsage" in iteration_stats: + self._log_gauge(self.cpu_memory_usage_bytes, + iteration_stats["cpuMemUsage"]) + if "pinnedMemUsage" in iteration_stats: + self._log_gauge(self.pinned_memory_usage_bytes, + iteration_stats["pinnedMemUsage"]) + + # Batch size + if "maxBatchSizeStatic" in iteration_stats: + self._log_gauge(self.max_batch_size_static, + iteration_stats["maxBatchSizeStatic"]) + if "maxBatchSizeRuntime" in iteration_stats: + self._log_gauge(self.max_batch_size_runtime, + iteration_stats["maxBatchSizeRuntime"]) + if "maxNumTokensRuntime" in iteration_stats: + self._log_gauge(self.max_num_tokens_runtime, + iteration_stats["maxNumTokensRuntime"]) + + # KV cache stats if kv_stats := iteration_stats.get("kvCacheStats"): cache_hit_rate = kv_stats.get("cacheHitRate") if cache_hit_rate is not None: @@ -295,6 +623,59 @@ def log_iteration_stats(self, iteration_stats: dict) -> None: if max_num_blocks: utilization = kv_stats["usedNumBlocks"] / max_num_blocks self._log_gauge(self.kv_cache_utilization, utilization) + if "maxNumBlocks" in kv_stats: + self._log_gauge(self.kv_cache_max_blocks, + kv_stats["maxNumBlocks"]) + if "freeNumBlocks" in kv_stats: + self._log_gauge(self.kv_cache_free_blocks, + kv_stats["freeNumBlocks"]) + if "usedNumBlocks" in kv_stats: + self._log_gauge(self.kv_cache_used_blocks, + kv_stats["usedNumBlocks"]) + if "tokensPerBlock" in kv_stats: + self._log_gauge(self.kv_cache_tokens_per_block, + kv_stats["tokensPerBlock"]) + + # Inflight batching stats + if ifb_stats := iteration_stats.get("inflightBatchingStats"): + if "numContextRequests" in ifb_stats: + self._log_gauge(self.num_context_requests, + ifb_stats["numContextRequests"]) + if "numGenRequests" in ifb_stats: + self._log_gauge(self.num_generation_requests, + ifb_stats["numGenRequests"]) + if "numPausedRequests" in ifb_stats: + self._log_gauge(self.num_paused_requests, + ifb_stats["numPausedRequests"]) + if "numScheduledRequests" in ifb_stats: + self._log_gauge(self.num_scheduled_requests, + ifb_stats["numScheduledRequests"]) + if "numCtxTokens" in ifb_stats: + self._log_gauge(self.total_context_tokens, + ifb_stats["numCtxTokens"]) + if "avgNumDecodedTokensPerIter" in ifb_stats: + self._log_gauge(self.avg_decoded_tokens_per_iter, + ifb_stats["avgNumDecodedTokensPerIter"]) + + # Speculative decoding stats + if spec_stats := iteration_stats.get("specDecodingStats"): + if "numDraftTokens" in spec_stats: + draft_tokens = spec_stats["numDraftTokens"] + if draft_tokens > 0: + self._log_counter(self.counter_spec_decode_num_draft_tokens, + {}, draft_tokens) + if "numAcceptedTokens" in spec_stats: + accepted_tokens = spec_stats["numAcceptedTokens"] + if accepted_tokens > 0: + self._log_counter( + self.counter_spec_decode_num_accepted_tokens, {}, + accepted_tokens) + if "acceptanceLength" in spec_stats: + self._log_gauge(self.spec_decode_acceptance_length, + spec_stats["acceptanceLength"]) + if "draftOverhead" in spec_stats: + self._log_gauge(self.spec_decode_draft_overhead, + spec_stats["draftOverhead"]) # Per-iteration KV cache stats (aggregated across window sizes) if kv_iter := iteration_stats.get("kvCacheIterationStats"): diff --git a/tensorrt_llm/metrics/enums.py b/tensorrt_llm/metrics/enums.py index e5beaeea7f93..c26ccc3b23ba 100644 --- a/tensorrt_llm/metrics/enums.py +++ b/tensorrt_llm/metrics/enums.py @@ -1,3 +1,17 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. from enum import Enum @@ -7,6 +21,11 @@ class MetricNames(Enum): E2E = "e2e" REQUEST_QUEUE_TIME = "request_queue_time" ARRIVAL_TIMESTAMP = 'arrival_timestamp' + PREFILL_TIME = "prefill_time" + DECODE_TIME = "decode_time" + INFERENCE_TIME = "inference_time" + PROMPT_TOKENS = "prompt_tokens" + GENERATION_TOKENS = "generation_tokens" class RequestEventTiming(Enum): diff --git a/tensorrt_llm/metrics/perf_utils.py b/tensorrt_llm/metrics/perf_utils.py new file mode 100644 index 000000000000..3bbf7f7a8130 --- /dev/null +++ b/tensorrt_llm/metrics/perf_utils.py @@ -0,0 +1,94 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Pure-Python utilities for computing per-request performance metrics. + +Kept in tensorrt_llm/metrics/ so that this logic has no GPU / heavy +dependencies and can be unit-tested without a full TensorRT-LLM install. +""" + +from typing import Optional + +from .enums import MetricNames, RequestEventTiming + + +def process_req_perf_metrics( + req_perf_metrics_dict: Optional[dict], output_length: int +) -> dict[MetricNames, float | int]: + """Compute derived per-request latency and token-count metrics. + + Args: + req_perf_metrics_dict: Raw timing dict from the executor, keyed by + ``RequestEventTiming`` enum members. May be ``None`` or empty. + output_length: Number of output tokens generated for this request. + + Returns: + Dict mapping ``MetricNames`` enum members to numeric values. + Keys with value <= 0 are filtered out, except ``REQUEST_QUEUE_TIME`` + which may be 0 (meaning the request was scheduled immediately). + """ + if not req_perf_metrics_dict: + return {} + + arrival = req_perf_metrics_dict.get(RequestEventTiming.ARRIVAL_TIME, 0) + first_scheduled = req_perf_metrics_dict.get(RequestEventTiming.FIRST_SCHEDULED_TIME, 0) + first_token = req_perf_metrics_dict.get(RequestEventTiming.FIRST_TOKEN_TIME, 0) + last_token = req_perf_metrics_dict.get(RequestEventTiming.LAST_TOKEN_TIME, 0) + + stat: dict[MetricNames, float | int] = {} + + # Base latency metrics — only compute when all required timestamps are + # present (> 0). Absent timestamps default to 0, so a difference that + # would be negative or zero indicates the timestamp was missing. + if first_token > 0 and arrival > 0: + stat[MetricNames.TTFT] = first_token - arrival + if last_token > 0 and arrival > 0: + stat[MetricNames.E2E] = last_token - arrival + # REQUEST_QUEUE_TIME is >= 0 for normally scheduled requests; zero is a + # valid value (immediate scheduling) so we include it when both timestamps + # are present. + if first_scheduled > 0 and arrival > 0: + stat[MetricNames.REQUEST_QUEUE_TIME] = first_scheduled - arrival + + # Phase latency metrics — require all three anchor timestamps to be valid. + # PREFILL_TIME = time from first scheduling to first generated token. + if first_token > 0 and first_scheduled > 0: + stat[MetricNames.PREFILL_TIME] = first_token - first_scheduled + # DECODE_TIME = time from first token to last token (generation phase). + if last_token > 0 and first_token > 0: + stat[MetricNames.DECODE_TIME] = last_token - first_token + # INFERENCE_TIME = first_scheduled → last_token (total execution time). + if last_token > 0 and first_scheduled > 0: + stat[MetricNames.INFERENCE_TIME] = last_token - first_scheduled + + # Token counts — recorded per candidate. When n>1 each candidate has + # its own timestamps and token stream so metrics are valid per candidate. + if output_length > 0: + stat[MetricNames.GENERATION_TOKENS] = output_length + + # TPOT = decode duration per output token. Requires at least 2 tokens + # (denominator would be 0 for a single-token output) and both timestamps + # present (first_token=0 default would produce bogus values). + if output_length > 1 and first_token > 0 and last_token > 0: + stat[MetricNames.TPOT] = (last_token - first_token) / (output_length - 1) + + # Filter out non-positive values: negatives indicate clock-skew anomalies + # and should not be reported; absent timestamps produce 0 which is filtered + # here except for REQUEST_QUEUE_TIME (which is re-added below if valid). + result = {k: v for k, v in stat.items() if v > 0} + # Restore REQUEST_QUEUE_TIME=0 if it was explicitly computed (zero queue + # time is a valid, meaningful observation). + if MetricNames.REQUEST_QUEUE_TIME in stat and stat[MetricNames.REQUEST_QUEUE_TIME] >= 0: + result[MetricNames.REQUEST_QUEUE_TIME] = stat[MetricNames.REQUEST_QUEUE_TIME] + return result diff --git a/tensorrt_llm/serve/openai_server.py b/tensorrt_llm/serve/openai_server.py index 1e1b6a401f51..adf05979ef60 100644 --- a/tensorrt_llm/serve/openai_server.py +++ b/tensorrt_llm/serve/openai_server.py @@ -369,15 +369,123 @@ def _init_llm(self, chat_template: Optional[str] = None): if self.generator.args.return_perf_metrics: set_prometheus_multiproc_dir() - self.metrics_collector = MetricsCollector({ - "model_name": "undefined", - "engine_type": "undefined" - }) + args = self.generator.args + pmc = getattr(args, "prometheus_metrics_config", None) + self.metrics_collector = MetricsCollector( + { + "model_name": self.model, + "engine_type": args.backend or "unknown" + }, + e2e_request_latency_buckets=(pmc.e2e_request_latency_buckets + if pmc else None), + time_to_first_token_buckets=(pmc.time_to_first_token_buckets + if pmc else None), + time_per_output_token_buckets=(pmc.time_per_output_token_buckets + if pmc else None), + request_queue_time_buckets=(pmc.request_queue_time_buckets + if pmc else None), + request_prefill_time_buckets=(pmc.request_prefill_time_buckets + if pmc else None), + request_decode_time_buckets=(pmc.request_decode_time_buckets + if pmc else None), + request_inference_time_buckets=( + pmc.request_inference_time_buckets if pmc else None), + ) + self._log_config_info_metrics() max_perf_metrics = self.generator.args.perf_metrics_max_requests if max_perf_metrics > 0: self.perf_metrics = deque(maxlen=max_perf_metrics) self.perf_metrics_lock = asyncio.Lock() + def _log_config_info_metrics(self) -> None: + """Extract configuration from generator args and log as Prometheus info gauges.""" + args = self.generator.args + + # Model config + model_config = { + "model": str(args.model), + "served_model_name": self.model, + "dtype": str(args.dtype), + } + quant_config = getattr(args, "quant_config", None) + if quant_config is not None: + quant_algo = getattr(quant_config, "quant_algo", None) + model_config["quantization"] = str( + quant_algo) if quant_algo else "none" + else: + model_config["quantization"] = "none" + max_seq_len = getattr(args, "max_seq_len", None) + if max_seq_len is not None: + model_config["max_model_len"] = str(max_seq_len) + try: + import torch + if torch.cuda.is_available(): + model_config["gpu_type"] = torch.cuda.get_device_name(0) + except (ImportError, RuntimeError) as e: + logger.debug("Could not detect GPU type for config metrics: %s", e) + + # Parallel config — prefer parallel_config from generator args + # for accurate values including cp_size and world_size. + par_cfg = getattr(args, "parallel_config", None) + if par_cfg is not None: + tp_size = getattr(par_cfg, "tp_size", 1) or 1 + pp_size = getattr(par_cfg, "pp_size", 1) or 1 + cp_size = getattr(par_cfg, "cp_size", 1) or 1 + world_size = getattr(par_cfg, "world_size", + tp_size * pp_size * cp_size) + else: + tp_size = getattr(args, "tensor_parallel_size", 1) or 1 + pp_size = getattr(args, "pipeline_parallel_size", 1) or 1 + cp_size = 1 + world_size = tp_size * pp_size * cp_size + parallel_config = { + "tensor_parallel_size": str(tp_size), + "pipeline_parallel_size": str(pp_size), + "context_parallel_size": str(cp_size), + "gpu_count": str(world_size), + } + ep_size = getattr(par_cfg, "moe_ep_size", None) if par_cfg else \ + getattr(args, "moe_expert_parallel_size", None) + if ep_size is not None and ep_size > 0: + parallel_config["expert_parallel_size"] = str(ep_size) + + # Speculative decoding config + spec_config_obj = getattr(args, "speculative_config", None) or getattr( + args, "decoding_config", None) + speculative_config = None + if spec_config_obj is not None: + speculative_config = {"spec_enabled": "true"} + decoding_type = getattr(spec_config_obj, "decoding_type", None) + if decoding_type is not None: + speculative_config["spec_method"] = str(decoding_type) + max_draft_len = getattr(spec_config_obj, "max_draft_len", None) + if max_draft_len is not None: + speculative_config["spec_num_tokens"] = str(max_draft_len) + draft_model = getattr(spec_config_obj, "speculative_model", None) + if draft_model is not None: + speculative_config["spec_draft_model"] = str(draft_model) + + # KV cache config + kv_cache_config_obj = getattr(args, "kv_cache_config", None) + kv_cache_config = None + if kv_cache_config_obj is not None: + kv_cache_config = {} + for field in ("page_size", "enable_block_reuse", + "enable_partial_reuse", "free_gpu_memory_fraction"): + val = getattr(kv_cache_config_obj, field, None) + if val is not None: + kv_cache_config[field] = str(val) + kv_dtype = getattr(kv_cache_config_obj, "dtype", None) + if kv_dtype is not None: + kv_cache_config["cache_dtype"] = str(kv_dtype) + + self.metrics_collector.log_config_info( + model_config=model_config, + parallel_config=parallel_config, + speculative_config=speculative_config, + kv_cache_config=kv_cache_config if kv_cache_config else None, + ) + async def await_disconnected(self, raw_request: Request, promise): if raw_request is None: return @@ -773,7 +881,14 @@ async def _extract_metrics(self, res: RequestOutput, raw_request: Request): if not res.finished: return if self.metrics_collector: - self.metrics_collector.log_request_metrics_dict(res.metrics_dict) + if res.candidate_metrics: + for candidate_m in res.candidate_metrics: + self.metrics_collector.log_request_metrics_dict(candidate_m) + elif res.metrics_dict: + # Fallback for paths that populate metrics_dict directly + # (e.g. PostprocWorker). + self.metrics_collector.log_request_metrics_dict( + res.metrics_dict) # Note: Iteration stats are collected by the background _iteration_stats_collector_loop task # Wake up the stats collector to drain iteration stats if getattr(self.generator.args, "enable_iter_perf_stats", True): diff --git a/tests/unittest/api_stability/references/llm.yaml b/tests/unittest/api_stability/references/llm.yaml index 3e5a4206279e..e3626793564e 100644 --- a/tests/unittest/api_stability/references/llm.yaml +++ b/tests/unittest/api_stability/references/llm.yaml @@ -159,6 +159,10 @@ methods: annotation: int default: 0 status: prototype + prometheus_metrics_config: + annotation: Optional[tensorrt_llm.llmapi.llm_args.PrometheusMetricsConfig] + default: null + status: prototype enable_energy_metrics: annotation: bool default: False diff --git a/tests/unittest/api_stability/references/request_output.yaml b/tests/unittest/api_stability/references/request_output.yaml index 684a178afb1c..e0a75bd5c25c 100644 --- a/tests/unittest/api_stability/references/request_output.yaml +++ b/tests/unittest/api_stability/references/request_output.yaml @@ -19,6 +19,9 @@ methods: stats: annotation: Optional[dict[str, float]] default: None + sequence_index: + annotation: int + default: 0 return_annotation: None do_tracing: parameters: diff --git a/tests/unittest/metrics/__init__.py b/tests/unittest/metrics/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/unittest/metrics/test_collector.py b/tests/unittest/metrics/test_collector.py index c9f80fd25513..c90f94545d07 100644 --- a/tests/unittest/metrics/test_collector.py +++ b/tests/unittest/metrics/test_collector.py @@ -12,22 +12,35 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Unit tests for MetricsCollector.log_iteration_stats with kvCacheIterationStats.""" +"""Unit tests for MetricsCollector and process_req_perf_metrics.""" import pytest +from prometheus_client import REGISTRY -prometheus_client = pytest.importorskip("prometheus_client") +from tensorrt_llm.metrics.collector import MetricsCollector +from tensorrt_llm.metrics.enums import MetricNames, RequestEventTiming +from tensorrt_llm.metrics.perf_utils import process_req_perf_metrics -from tensorrt_llm.metrics.collector import MetricsCollector # noqa: E402 -# Use a single module-level collector to avoid re-registering Prometheus metrics -# (Prometheus does not allow duplicate metric names in the same process). -_collector = MetricsCollector(labels={"test": "true"}) +@pytest.fixture(autouse=True) +def clean_registry(): + """Unregister all custom collectors between tests to avoid duplicate metric errors.""" + collectors_to_remove = [] + for collector in REGISTRY._names_to_collectors.values(): + if hasattr(collector, "_name") and collector._name.startswith("trtllm_"): + collectors_to_remove.append(collector) + for collector in set(collectors_to_remove): + try: + REGISTRY.unregister(collector) + except Exception: + pass + yield -def _make_collector() -> MetricsCollector: - """Return the shared collector instance.""" - return _collector +@pytest.fixture +def collector(): + labels = {"model_name": "test_model"} + return MetricsCollector(labels) def _get_gauge_value(collector, metric_name: str): @@ -42,17 +55,654 @@ def _get_counter_value(collector, metric_name: str): return metric.labels(**collector.labels)._value.get() +SAMPLE_ITERATION_STATS = { + "numActiveRequests": 5, + "numQueuedRequests": 3, + "numCompletedRequests": 2, + "maxNumActiveRequests": 10, + "iterLatencyMS": 15.5, + "gpuMemUsage": 4_000_000_000, + "cpuMemUsage": 2_000_000_000, + "pinnedMemUsage": 500_000_000, + "maxBatchSizeStatic": 64, + "maxBatchSizeRuntime": 32, + "maxNumTokensRuntime": 8192, + "kvCacheStats": { + "cacheHitRate": 0.85, + "maxNumBlocks": 1000, + "freeNumBlocks": 400, + "usedNumBlocks": 600, + "tokensPerBlock": 64, + }, + "inflightBatchingStats": { + "numContextRequests": 2, + "numGenRequests": 3, + "numPausedRequests": 1, + "numScheduledRequests": 5, + "numCtxTokens": 256, + "avgNumDecodedTokensPerIter": 4.5, + }, + "specDecodingStats": { + "numDraftTokens": 20, + "numAcceptedTokens": 15, + "acceptanceLength": 3.75, + "draftOverhead": 1.2, + }, +} + + +class TestIterationStatsTopLevel: + """Test top-level iteration stats are correctly exposed as Prometheus metrics.""" + + def test_queue_and_load_gauges(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_gauge_value(collector, "num_requests_running") == 5 + assert _get_gauge_value(collector, "num_requests_waiting") == 3 + assert _get_gauge_value(collector, "max_num_active_requests") == 10 + + def test_completed_requests_counter(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + assert _get_counter_value(collector, "counter_num_requests_completed") == 2 + + # Counter should accumulate across iterations + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + assert _get_counter_value(collector, "counter_num_requests_completed") == 4 + + def test_completed_requests_zero_not_incremented(self, collector): + stats = {**SAMPLE_ITERATION_STATS, "numCompletedRequests": 0} + collector.log_iteration_stats(stats) + assert _get_counter_value(collector, "counter_num_requests_completed") == 0 + + def test_iteration_latency_ms_to_seconds(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + assert _get_gauge_value(collector, "iteration_latency_seconds") == pytest.approx(0.0155) + + def test_memory_usage(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_gauge_value(collector, "gpu_memory_usage_bytes") == 4_000_000_000 + assert _get_gauge_value(collector, "cpu_memory_usage_bytes") == 2_000_000_000 + assert _get_gauge_value(collector, "pinned_memory_usage_bytes") == 500_000_000 + + def test_batch_size(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_gauge_value(collector, "max_batch_size_static") == 64 + assert _get_gauge_value(collector, "max_batch_size_runtime") == 32 + assert _get_gauge_value(collector, "max_num_tokens_runtime") == 8192 + + +class TestKVCacheStats: + """Test KV cache stats are correctly exposed.""" + + def test_kv_cache_block_gauges(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_gauge_value(collector, "kv_cache_max_blocks") == 1000 + assert _get_gauge_value(collector, "kv_cache_free_blocks") == 400 + assert _get_gauge_value(collector, "kv_cache_used_blocks") == 600 + assert _get_gauge_value(collector, "kv_cache_tokens_per_block") == 64 + + def test_kv_cache_utilization(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + assert _get_gauge_value(collector, "kv_cache_utilization") == pytest.approx(0.6) + + def test_kv_cache_hit_rate(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + assert _get_gauge_value(collector, "kv_cache_hit_rate") == pytest.approx(0.85) + + +class TestInflightBatchingStats: + """Test inflight batching stats are correctly exposed.""" + + def test_inflight_batching_gauges(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_gauge_value(collector, "num_context_requests") == 2 + assert _get_gauge_value(collector, "num_generation_requests") == 3 + assert _get_gauge_value(collector, "num_paused_requests") == 1 + assert _get_gauge_value(collector, "num_scheduled_requests") == 5 + assert _get_gauge_value(collector, "total_context_tokens") == 256 + assert _get_gauge_value(collector, "avg_decoded_tokens_per_iter") == pytest.approx(4.5) + + def test_missing_inflight_batching_stats(self, collector): + """No error when inflightBatchingStats is absent.""" + stats = {k: v for k, v in SAMPLE_ITERATION_STATS.items() if k != "inflightBatchingStats"} + collector.log_iteration_stats(stats) + + +class TestSpecDecodingStats: + """Test speculative decoding stats are correctly exposed.""" + + def test_spec_decoding_metrics(self, collector): + collector.log_iteration_stats(SAMPLE_ITERATION_STATS) + + assert _get_counter_value(collector, "counter_spec_decode_num_draft_tokens") == 20 + assert _get_counter_value(collector, "counter_spec_decode_num_accepted_tokens") == 15 + assert _get_gauge_value(collector, "spec_decode_acceptance_length") == pytest.approx(3.75) + assert _get_gauge_value(collector, "spec_decode_draft_overhead") == pytest.approx(1.2) + + def test_missing_spec_decoding_stats(self, collector): + """No error when specDecodingStats is absent.""" + stats = {k: v for k, v in SAMPLE_ITERATION_STATS.items() if k != "specDecodingStats"} + collector.log_iteration_stats(stats) + + +class TestPartialStats: + """Test that partial iteration stats don't cause errors.""" + + def test_empty_stats(self, collector): + collector.log_iteration_stats({}) + + def test_only_kv_cache(self, collector): + collector.log_iteration_stats( + { + "kvCacheStats": { + "maxNumBlocks": 100, + "usedNumBlocks": 50, + } + } + ) + assert _get_gauge_value(collector, "kv_cache_max_blocks") == 100 + assert _get_gauge_value(collector, "kv_cache_used_blocks") == 50 + + +class TestConfigInfoMetrics: + """Test config info gauges are correctly exposed.""" + + def test_model_config_info(self, collector): + model_config = { + "model": "meta-llama/Llama-3-8B", + "served_model_name": "Llama-3-8B", + "dtype": "float16", + "quantization": "none", + "max_model_len": "4096", + "gpu_type": "NVIDIA H100", + } + collector.log_config_info(model_config=model_config) + assert ( + REGISTRY.get_sample_value( + "trtllm_model_config_info", {"model_name": "test_model", **model_config} + ) + == 1.0 + ) + + def test_parallel_config_info(self, collector): + parallel_config = { + "tensor_parallel_size": "4", + "pipeline_parallel_size": "2", + "gpu_count": "8", + } + collector.log_config_info(parallel_config=parallel_config) + assert ( + REGISTRY.get_sample_value( + "trtllm_parallel_config_info", {"model_name": "test_model", **parallel_config} + ) + == 1.0 + ) + + def test_speculative_config_info(self, collector): + spec_config = { + "spec_enabled": "true", + "spec_method": "Eagle", + "spec_num_tokens": "5", + "spec_draft_model": "eagle-model", + } + collector.log_config_info(speculative_config=spec_config) + assert ( + REGISTRY.get_sample_value( + "trtllm_speculative_config_info", {"model_name": "test_model", **spec_config} + ) + == 1.0 + ) + + def test_kv_cache_config_info(self, collector): + kv_cache_config = { + "page_size": "64", + "enable_block_reuse": "True", + "cache_dtype": "auto", + } + collector.log_config_info(kv_cache_config=kv_cache_config) + assert ( + REGISTRY.get_sample_value( + "trtllm_kv_cache_config_info", {"model_name": "test_model", **kv_cache_config} + ) + == 1.0 + ) + + def test_no_config_no_error(self, collector): + """No error when all configs are None.""" + collector.log_config_info() + + def test_partial_config(self, collector): + """Only model config provided, others None.""" + collector.log_config_info(model_config={"model": "test", "dtype": "auto"}) + + +# --------------------------------------------------------------------------- +# Per-request token counters and phase histograms (Step 1 additions) +# --------------------------------------------------------------------------- + +SAMPLE_REQUEST_METRICS_FULL = { + MetricsCollector.labelname_finish_reason: "end_id", + # latency fields (seconds) — keys are MetricNames enum members + MetricNames.E2E: 2.5, + MetricNames.TTFT: 0.3, + MetricNames.TPOT: 0.05, + MetricNames.REQUEST_QUEUE_TIME: 0.1, + MetricNames.PREFILL_TIME: 0.2, + MetricNames.DECODE_TIME: 2.2, + MetricNames.INFERENCE_TIME: 2.4, + # token counts + MetricNames.PROMPT_TOKENS: 128, + MetricNames.GENERATION_TOKENS: 50, +} + + +class TestRequestSuccessCounter: + """Test counter_request_success increments with the correct finished_reason label.""" + + def test_success_counter_incremented(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + labels = {**collector.labels, "finished_reason": "end_id"} + assert collector.counter_request_success.labels(**labels)._value.get() == 1 + + def test_success_counter_tracks_finish_reason_separately(self, collector): + """Different finish_reason values must be tracked in separate label series.""" + metrics_stop = { + **SAMPLE_REQUEST_METRICS_FULL, + MetricsCollector.labelname_finish_reason: "stop_words", + } + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + collector.log_request_metrics_dict(metrics_stop) + end_id_labels = {**collector.labels, "finished_reason": "end_id"} + stop_labels = {**collector.labels, "finished_reason": "stop_words"} + assert collector.counter_request_success.labels(**end_id_labels)._value.get() == 1 + assert collector.counter_request_success.labels(**stop_labels)._value.get() == 1 + + +class TestPerRequestTokenCounters: + """Test prompt_tokens_total and generation_tokens_total counters.""" + + def test_prompt_tokens_incremented(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_counter_value(collector, "counter_prompt_tokens") == 128 + + def test_generation_tokens_incremented(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_counter_value(collector, "counter_generation_tokens") == 50 + + def test_token_counters_accumulate(self, collector): + """Counters should sum across multiple requests.""" + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_counter_value(collector, "counter_prompt_tokens") == 256 + assert _get_counter_value(collector, "counter_generation_tokens") == 100 + + def test_missing_token_counts_no_error(self, collector): + """No error and no increment when token counts are absent.""" + metrics_without_tokens = { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.E2E: 1.0, + MetricNames.TTFT: 0.1, + } + collector.log_request_metrics_dict(metrics_without_tokens) + assert _get_counter_value(collector, "counter_prompt_tokens") == 0 + assert _get_counter_value(collector, "counter_generation_tokens") == 0 + + def test_zero_tokens_not_incremented(self, collector): + """Zero token counts should not increment the counter.""" + metrics = { + **SAMPLE_REQUEST_METRICS_FULL, + MetricNames.PROMPT_TOKENS: 0, + MetricNames.GENERATION_TOKENS: 0, + } + collector.log_request_metrics_dict(metrics) + assert _get_counter_value(collector, "counter_prompt_tokens") == 0 + assert _get_counter_value(collector, "counter_generation_tokens") == 0 + + def test_n_greater_than_1_prompt_tokens_counted_once(self, collector): + """For n>1, PROMPT_TOKENS should be counted once (candidate 0 only). + + GENERATION_TOKENS accumulates per candidate. + + This simulates what record_stats in executor/result.py produces: + candidate 0 emits both PROMPT_TOKENS and GENERATION_TOKENS, + candidates 1+ emit only GENERATION_TOKENS (prompt is shared). + """ + # Candidate 0: prompt + generation tokens + collector.log_request_metrics_dict( + { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.PROMPT_TOKENS: 128, + MetricNames.GENERATION_TOKENS: 50, + } + ) + # Candidate 1: only generation tokens (shared prompt not re-counted) + collector.log_request_metrics_dict( + { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.GENERATION_TOKENS: 42, + } + ) + # Candidate 2: only generation tokens + collector.log_request_metrics_dict( + { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.GENERATION_TOKENS: 38, + } + ) + # Prompt counted once, generation tokens summed across all candidates + assert _get_counter_value(collector, "counter_prompt_tokens") == 128 + assert _get_counter_value(collector, "counter_generation_tokens") == 130 # 50 + 42 + 38 + + +def _get_histogram_sum(collector, metric_name: str): + """Return the sum of all observations in a Prometheus histogram.""" + histogram = getattr(collector, metric_name) + for metric in REGISTRY.collect(): + if metric.name == histogram._name: + for sample in metric.samples: + if sample.name.endswith("_sum") and sample.labels == collector.labels: + return sample.value + return 0.0 + + +def _get_histogram_count(collector, metric_name: str): + """Return the number of observations in a Prometheus histogram.""" + histogram = getattr(collector, metric_name) + for metric in REGISTRY.collect(): + if metric.name == histogram._name: + for sample in metric.samples: + if sample.name.endswith("_count") and sample.labels == collector.labels: + return int(sample.value) + return 0 + + +class TestPerRequestPhaseHistograms: + """Test request_prefill_time_seconds, _decode_time_seconds, _inference_time_seconds.""" + + def test_prefill_time_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_prefill_time_request") == 1 + assert _get_histogram_sum(collector, "histogram_prefill_time_request") == pytest.approx(0.2) + + def test_decode_time_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_decode_time_request") == 1 + assert _get_histogram_sum(collector, "histogram_decode_time_request") == pytest.approx(2.2) + + def test_inference_time_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_inference_time_request") == 1 + assert _get_histogram_sum(collector, "histogram_inference_time_request") == pytest.approx( + 2.4 + ) + + def test_e2e_time_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_e2e_time_request") == 1 + assert _get_histogram_sum(collector, "histogram_e2e_time_request") == pytest.approx(2.5) + + def test_ttft_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_time_to_first_token") == 1 + assert _get_histogram_sum(collector, "histogram_time_to_first_token") == pytest.approx(0.3) + + def test_tpot_observed(self, collector): + collector.log_request_metrics_dict(SAMPLE_REQUEST_METRICS_FULL) + assert _get_histogram_count(collector, "histogram_time_per_output_token") == 1 + assert _get_histogram_sum(collector, "histogram_time_per_output_token") == pytest.approx( + 0.05 + ) + + def test_missing_phase_times_no_observation(self, collector): + """No histogram observations when phase times are absent.""" + metrics = { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.E2E: 1.0, + MetricNames.TTFT: 0.1, + } + collector.log_request_metrics_dict(metrics) + assert _get_histogram_count(collector, "histogram_prefill_time_request") == 0 + assert _get_histogram_count(collector, "histogram_decode_time_request") == 0 + assert _get_histogram_count(collector, "histogram_inference_time_request") == 0 + + def test_no_observation_without_finish_reason(self, collector): + """Phase histograms must not be updated for in-progress requests.""" + metrics_no_finish = { + MetricNames.E2E: 1.0, + MetricNames.PREFILL_TIME: 0.2, + MetricNames.DECODE_TIME: 0.8, + MetricNames.INFERENCE_TIME: 1.0, + MetricNames.PROMPT_TOKENS: 10, + MetricNames.GENERATION_TOKENS: 5, + } + collector.log_request_metrics_dict(metrics_no_finish) + assert _get_histogram_count(collector, "histogram_prefill_time_request") == 0 + assert _get_counter_value(collector, "counter_prompt_tokens") == 0 + + def test_queue_time_zero_is_recorded_to_prometheus(self, collector): + """REQUEST_QUEUE_TIME=0 must reach the Prometheus histogram (not silently dropped).""" + metrics = { + MetricsCollector.labelname_finish_reason: "end_id", + MetricNames.E2E: 2.5, + MetricNames.TTFT: 0.3, + MetricNames.REQUEST_QUEUE_TIME: 0.0, # immediate scheduling + } + collector.log_request_metrics_dict(metrics) + assert _get_histogram_count(collector, "histogram_queue_time_request") == 1 + assert _get_histogram_sum(collector, "histogram_queue_time_request") == pytest.approx(0.0) + + +# Shared timing fixture used across TestProcessReqPerfMetrics tests. +_FULL_TIMESTAMPS = { + RequestEventTiming.ARRIVAL_TIME: 1000.0, + RequestEventTiming.FIRST_SCHEDULED_TIME: 1000.05, + RequestEventTiming.FIRST_TOKEN_TIME: 1000.3, + RequestEventTiming.LAST_TOKEN_TIME: 1002.5, +} + + +class TestProcessReqPerfMetrics: + """Unit tests for process_req_perf_metrics (new phase timings + token counts).""" + + def test_phase_timings_computed(self): + stat = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=50) + + assert stat[MetricNames.PREFILL_TIME] == pytest.approx(0.25) + assert stat[MetricNames.DECODE_TIME] == pytest.approx(2.2) + assert stat[MetricNames.INFERENCE_TIME] == pytest.approx(2.45) + + def test_base_latencies_computed(self): + stat = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=50) + + assert stat[MetricNames.TTFT] == pytest.approx(0.3) + assert stat[MetricNames.E2E] == pytest.approx(2.5) + + def test_zero_queue_time_is_included(self): + """REQUEST_QUEUE_TIME=0 (instant scheduling) must not be filtered out.""" + timestamps = { + RequestEventTiming.ARRIVAL_TIME: 1000.0, + RequestEventTiming.FIRST_SCHEDULED_TIME: 1000.0, # same as arrival + RequestEventTiming.FIRST_TOKEN_TIME: 1000.3, + RequestEventTiming.LAST_TOKEN_TIME: 1002.5, + } + stat = process_req_perf_metrics(timestamps, output_length=50) + assert MetricNames.REQUEST_QUEUE_TIME in stat + assert stat[MetricNames.REQUEST_QUEUE_TIME] == pytest.approx(0.0) + + def test_generation_tokens_in_stat(self): + stat = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=50) + assert stat[MetricNames.GENERATION_TOKENS] == 50 + + def test_output_length_one_has_tokens_but_no_tpot(self): + """GENERATION_TOKENS is present but TPOT is excluded for single-token output.""" + stat = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=1) + assert stat[MetricNames.GENERATION_TOKENS] == 1 + assert MetricNames.TPOT not in stat + + def test_per_candidate_metrics_computed_independently(self): + """Each candidate gets its own GENERATION_TOKENS and TPOT. + + Even when candidates have different output lengths. + """ + stat_a = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=50) + stat_b = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=30) + assert stat_a[MetricNames.GENERATION_TOKENS] == 50 + assert stat_b[MetricNames.GENERATION_TOKENS] == 30 + # TPOT differs because output_length differs + assert stat_a[MetricNames.TPOT] != stat_b[MetricNames.TPOT] + + def test_zero_output_length_excludes_tokens(self): + stat = process_req_perf_metrics(_FULL_TIMESTAMPS, output_length=0) + assert MetricNames.GENERATION_TOKENS not in stat + + def test_missing_timestamps_no_phase_timings(self): + """When only the arrival timestamp is present, phase metrics are absent.""" + # Key must be RequestEventTiming enum to match dict lookups. + raw = {RequestEventTiming.ARRIVAL_TIME: 1000.0} + stat = process_req_perf_metrics(raw, output_length=10) + assert MetricNames.PREFILL_TIME not in stat + assert MetricNames.DECODE_TIME not in stat + assert MetricNames.INFERENCE_TIME not in stat + assert MetricNames.TTFT not in stat + assert MetricNames.E2E not in stat + + def test_clock_skew_negative_phase_time_is_dropped(self): + """Negative phase durations (clock skew) must not appear in output. + + Non-negative metrics must still be present. + """ + skewed = { + RequestEventTiming.ARRIVAL_TIME: 1000.0, + RequestEventTiming.FIRST_SCHEDULED_TIME: 1000.05, + # first_token before first_scheduled — invalid clock ordering + RequestEventTiming.FIRST_TOKEN_TIME: 1000.02, + RequestEventTiming.LAST_TOKEN_TIME: 1002.5, + } + stat = process_req_perf_metrics(skewed, output_length=50) + # PREFILL_TIME = 1000.02 - 1000.05 = -0.03 → must be absent + assert MetricNames.PREFILL_TIME not in stat + # Non-negative metrics must still be present + assert MetricNames.TTFT in stat + assert stat[MetricNames.TTFT] == pytest.approx(0.02) + assert MetricNames.E2E in stat + assert stat[MetricNames.E2E] == pytest.approx(2.5) + assert MetricNames.DECODE_TIME in stat + assert stat[MetricNames.DECODE_TIME] == pytest.approx(2.48) + + def test_negative_queue_time_is_dropped(self): + """Negative REQUEST_QUEUE_TIME due to clock skew must not appear. + + When arrival is after first_scheduled due to clock skew, + the output should not contain the metric. + """ + skewed = { + RequestEventTiming.ARRIVAL_TIME: 1000.10, # arrives after scheduled + RequestEventTiming.FIRST_SCHEDULED_TIME: 1000.05, + RequestEventTiming.FIRST_TOKEN_TIME: 1000.5, + RequestEventTiming.LAST_TOKEN_TIME: 1002.5, + } + stat = process_req_perf_metrics(skewed, output_length=50) + # REQUEST_QUEUE_TIME = 1000.05 - 1000.10 = -0.05 → must be absent + assert MetricNames.REQUEST_QUEUE_TIME not in stat + # Other metrics are still valid + assert MetricNames.PREFILL_TIME in stat + + def test_empty_stats_returns_empty(self): + assert process_req_perf_metrics(None, output_length=10) == {} + + +class TestCustomHistogramBuckets: + """Tests for configurable Prometheus histogram bucket boundaries.""" + + def test_custom_buckets_applied(self): + """Custom bucket lists are reflected in the created histograms.""" + labels = {"model_name": "test_model"} + custom_e2e = [0.1, 0.5, 1.0, 5.0, 10.0] + custom_ttft = [0.001, 0.01, 0.1, 1.0] + c = MetricsCollector( + labels, e2e_request_latency_buckets=custom_e2e, time_to_first_token_buckets=custom_ttft + ) + # prometheus_client appends +Inf automatically + assert list(c.histogram_e2e_time_request._upper_bounds) == custom_e2e + [float("inf")] + assert list(c.histogram_time_to_first_token._upper_bounds) == custom_ttft + [float("inf")] + + def test_none_uses_defaults(self): + """None (unset) falls back to the built-in default bucket boundaries.""" + labels = {"model_name": "test_model"} + c = MetricsCollector(labels) + default_e2e = [ + 0.3, + 0.5, + 0.8, + 1.0, + 1.5, + 2.0, + 2.5, + 5.0, + 10.0, + 15.0, + 20.0, + 30.0, + 40.0, + 50.0, + 60.0, + 120.0, + 240.0, + 480.0, + 960.0, + 1920.0, + 7680.0, + ] + assert list(c.histogram_e2e_time_request._upper_bounds) == default_e2e + [float("inf")] + + def test_empty_bucket_list_raises(self): + """Passing an empty list must raise ValueError at MetricsCollector init.""" + labels = {"model_name": "test_model"} + with pytest.raises(ValueError, match="must not be empty"): + MetricsCollector(labels, e2e_request_latency_buckets=[]) + + def test_unsorted_bucket_list_raises(self): + """Passing an unsorted list must raise ValueError at MetricsCollector init.""" + labels = {"model_name": "test_model"} + with pytest.raises(ValueError, match="must be strictly increasing"): + MetricsCollector(labels, time_to_first_token_buckets=[1.0, 0.5, 2.0]) + + def test_duplicate_bucket_values_raises(self): + """Passing duplicate values must raise ValueError (not strictly increasing).""" + labels = {"model_name": "test_model"} + with pytest.raises(ValueError, match="must be strictly increasing"): + MetricsCollector(labels, e2e_request_latency_buckets=[1.0, 1.0, 2.0]) + + +# --------------------------------------------------------------------------- +# Tests for kvCacheIterationStats (from main — TRTLLM-11421) +# --------------------------------------------------------------------------- + + +def _make_kv_iter_collector() -> MetricsCollector: + """Construct a fresh MetricsCollector for kv_iter tests. + + Uses a unique label so the autouse clean_registry fixture does not + collide with collectors from other tests in this module. + """ + return MetricsCollector(labels={"kv_iter_test": "true"}) + + class TestLogIterationStatsKvCacheIteration: def test_no_kv_cache_iteration_stats(self): """When kvCacheIterationStats is absent, new metrics should not error.""" - collector = _make_collector() + collector = _make_kv_iter_collector() stats = {"kvCacheStats": {"cacheHitRate": 0.5, "usedNumBlocks": 10, "maxNumBlocks": 20}} # Should not raise collector.log_iteration_stats(stats) def test_gauges_updated(self): """Host utilization and iter reuse rate gauges should be set.""" - collector = _make_collector() + collector = _make_kv_iter_collector() stats = { "kvCacheIterationStats": { "16": { @@ -86,7 +736,7 @@ def test_gauges_updated(self): def test_counters_incremented(self): """Counter metrics should accumulate deltas across calls.""" - collector = _make_collector() + collector = _make_kv_iter_collector() stats = { "kvCacheIterationStats": { "16": { @@ -154,7 +804,7 @@ def test_counters_incremented(self): def test_multiple_windows_aggregated(self): """Stats from multiple window sizes should be summed.""" - collector = _make_collector() + collector = _make_kv_iter_collector() ws16 = { "primaryMaxNumBlocks": 50, "primaryFreeNumBlocks": 30, @@ -213,7 +863,7 @@ def test_multiple_windows_aggregated(self): def test_zero_deltas_no_counter_increment(self): """When all deltas are zero, counters should not increment.""" - collector = _make_collector() + collector = _make_kv_iter_collector() stats = { "kvCacheIterationStats": { "16": {