From c98517ce7fab78221e7806b56dd862a5fb69eae4 Mon Sep 17 00:00:00 2001 From: vraiti Date: Thu, 13 Aug 2026 02:59:13 -0400 Subject: [PATCH] [Metrics] Add Diffusion Metrics (#4755) Signed-off-by: vraiti Co-authored-by: Claude Opus 4.6 --- docs/design/metrics.md | 2 +- docs/usage/metrics.md | 76 ++++++--- .../test_diffusion_output_formatter.py | 1 + tests/metrics/test_modality.py | 150 +++++++++++++++++- vllm_omni/diffusion/output_formatter.py | 1 + vllm_omni/entrypoints/omni_base.py | 1 + vllm_omni/metrics/definitions.py | 9 ++ vllm_omni/metrics/modality.py | 135 ++++++++++++---- vllm_omni/metrics/stats.py | 17 +- 9 files changed, 328 insertions(+), 64 deletions(-) diff --git a/docs/design/metrics.md b/docs/design/metrics.md index 6e4946f4e..7ee35813f 100644 --- a/docs/design/metrics.md +++ b/docs/design/metrics.md @@ -1,6 +1,6 @@ # Prometheus Metrics Design -This document describes how vLLM-Omni exposes Prometheus metrics for multi-stage pipelines, the constraints that shaped the design, and how the pipeline-level metrics coexist with upstream vLLM per-engine metrics. This iteration covers the text and audio surface; diffusion / image / video families land in a follow-up. +This document describes how vLLM-Omni exposes Prometheus metrics for multi-stage pipelines, the constraints that shaped the design, and how the pipeline-level metrics coexist with upstream vLLM per-engine metrics. ## Objectives diff --git a/docs/usage/metrics.md b/docs/usage/metrics.md index 68a38043e..184a76eef 100644 --- a/docs/usage/metrics.md +++ b/docs/usage/metrics.md @@ -1,23 +1,23 @@ # Production Metrics -vLLM-Omni exposes Prometheus metrics via the `/metrics` endpoint on the OpenAI-compatible API server. This page covers the text and audio surface; diffusion / image / video metrics are tracked in a follow-up PR. +vLLM-Omni exposes Prometheus metrics via the `/metrics` endpoint on the OpenAI-compatible API server. ```bash vllm-omni serve Qwen/Qwen3-Omni-30B-A3B-Instruct --port 8000 --log-stats curl http://localhost:8000/metrics ``` -**`--log-stats` is required to populate metric data.** Without the flag, the endpoint still returns `200 OK` but the upstream `vllm:*` wrap is not registered at all, and the 15 `vllm:omni_*` families are registered as placeholders with no data written to them. This keeps the runtime cost essentially zero for deployments that don't need monitoring. With the flag, all ~80 families populate. +**`--log-stats` is required to populate metric data.** Without the flag, the endpoint still returns `200 OK` but the upstream `vllm:*` wrap is not registered at all, and the 15 `vllm_omni:*` families are registered as placeholders with no data written to them. This keeps the runtime cost essentially zero for deployments that don't need monitoring. With the flag, all ~80 families populate. ## Metric Namespaces | Prefix | Source | Present when | |--------|--------|--------------| -| `vllm:omni_` | vLLM-Omni orchestrator / audio modality / cross-stage transfer | Pipeline-dependent | +| `vllm_omni:` | vLLM-Omni orchestrator / audio modality / cross-stage transfer | Pipeline-dependent | | `vllm:` | Upstream vLLM engine, wrapped by `OmniPrometheusStatLogger` to expose `{stage, replica}` | Pipeline includes an LLM (AR) stage | | `http_` / `process_` | Uvicorn / Python runtime | Always | -## Pipeline-Level Metrics (`vllm:omni_`) +## Pipeline-Level Metrics (`vllm_omni:`) Defined in `vllm_omni/metrics/prometheus.py`. Track request lifecycle across the full multi-stage pipeline. @@ -25,42 +25,66 @@ Defined in `vllm_omni/metrics/prometheus.py`. Track request lifecycle across the | Metric | Type | Labels | Description | |--------|------|--------|-------------| -| `vllm:omni_num_requests_running` | Gauge | `model_name` | Pipeline-global in-flight requests (dispatched to engine, not yet finalized) | -| `vllm:omni_num_requests_waiting` | Gauge | `model_name` | Requests waiting in the Orchestrator queue | -| `vllm:omni_requests_success_total` | Counter | `model_name`, `finished_reason` | Total requests by completion reason. `finished_reason` ∈ {`stop`, `length`, `abort`, ...} mirroring upstream `vllm:request_success_total`; aborts cover client disconnect / cancellation paths in addition to upstream `FinishReason.ABORT` | +| `vllm_omni:num_requests_running` | Gauge | `model_name` | Pipeline-global in-flight requests (dispatched to engine, not yet finalized) | +| `vllm_omni:num_requests_waiting` | Gauge | `model_name` | Requests waiting in the Orchestrator queue | +| `vllm_omni:requests_success_total` | Counter | `model_name`, `finished_reason` | Total requests by completion reason. `finished_reason` ∈ {`stop`, `length`, `abort`, ...} mirroring upstream `vllm:request_success_total`; aborts cover client disconnect / cancellation paths in addition to upstream `FinishReason.ABORT` | ### Latency | Metric | Type | Labels | Description | |--------|------|--------|-------------| -| `vllm:omni_e2e_request_latency_s` | Histogram | `model_name` | Pipeline-global end-to-end request latency in seconds | +| `vllm_omni:e2e_request_latency_s` | Histogram | `model_name` | Pipeline-global end-to-end request latency in seconds | -## Audio Modality Metrics (`vllm:omni_`) +## Audio Modality Metrics (`vllm_omni:`) Emitted at request finalize, except for `audio_ttfp_s` (streaming-hook at the first audio packet) and `audio_underrun_s` / `audio_continuity_ok_total` (streaming finalize, after the chunk stream is exhausted). All carry `{model_name, stage, replica}` plus the listed extra label. | Metric | Type | Extra label | Description | |--------|------|-------------|-------------| -| `vllm:omni_audio_ttfp_s` | Histogram | — | Time from request arrival to first audio packet/frame | -| `vllm:omni_audio_duration_s` | Histogram | — | Audio content duration (`audio_frames / sample_rate`) | -| `vllm:omni_audio_rtf` | Histogram | — | Real-time factor (`stage_gen_time_s / audio_duration_s`); streaming TTS SLO red line `< 1`; uses `RTF_BUCKETS` | -| `vllm:omni_audio_frames_total` | Counter | — | Cumulative audio frame count; throughput via `rate()` | -| `vllm:omni_audio_underrun_s` | Histogram | — | Per-request worst-case player deficit; `> 0` indicates listener heard silent gaps | -| `vllm:omni_audio_continuity_ok_total` | Counter | `threshold_ms` | Incremented when the request's worst underrun stayed below `threshold_ms` | -| `vllm:omni_audio_skipped_requests_total` | Counter | `reason` | Silent-loss counter — code2wav rejected malformed codec input and returned `200 OK` with empty audio | +| `vllm_omni:audio_ttfp_s` | Histogram | — | Time from request arrival to first audio packet/frame | +| `vllm_omni:audio_duration_s` | Histogram | — | Audio content duration (`audio_frames / sample_rate`) | +| `vllm_omni:audio_rtf` | Histogram | — | Real-time factor (`stage_gen_time_s / audio_duration_s`); streaming TTS SLO red line `< 1`; uses `RTF_BUCKETS` | +| `vllm_omni:audio_frames_total` | Counter | — | Cumulative audio frame count; throughput via `rate()` | +| `vllm_omni:audio_underrun_s` | Histogram | — | Per-request worst-case player deficit; `> 0` indicates listener heard silent gaps | +| `vllm_omni:audio_continuity_ok_total` | Counter | `threshold_ms` | Incremented when the request's worst underrun stayed below `threshold_ms` | +| `vllm_omni:audio_skipped_requests_total` | Counter | `reason` | Silent-loss counter — code2wav rejected malformed codec input and returned `200 OK` with empty audio | The continuity math comes from `vllm_omni/benchmarks/audio_continuity.py::compute_continuity_stats` so the server-side observation aligns with the bench-side definition. -## Cross-Stage Transfer Metrics (`vllm:omni_`) +## Diffusion Metrics (`vllm_omni:`) + +Per-request timing breakdowns for diffusion (image/video) stages. Emitted at request finalize when `stage_metrics.diffusion_metrics` is present. All carry `{model_name, stage, replica}`. + +| Metric | Type | Description | +|--------|------|-------------| +| `vllm_omni:diffusion_exec_s` | Histogram | DiT forward pass execution time per request in seconds | +| `vllm_omni:diffusion_exec_per_step_s` | Histogram | DiT forward pass execution time per denoising step in seconds (`exec_s / num_inference_steps`) | +| `vllm_omni:diffusion_preprocess_s` | Histogram | Diffusion input preprocessing time per request in seconds | +| `vllm_omni:diffusion_postprocess_s` | Histogram | Diffusion output postprocessing (VAE decode) time per request in seconds | + +`diffusion_exec_s` uses `SECONDS_BUCKETS` (50 ms → 300 s); the other three use `SECONDS_FAST_BUCKETS` (1 ms → 60 s). `diffusion_exec_per_step_s` is skipped when `num_inference_steps` is not set. + +### Example PromQL + +```promql +# P99 DiT exec time by stage +histogram_quantile(0.99, rate(vllm_omni:diffusion_exec_s_bucket[5m])) + +# Ratio of preprocess to total exec wall time +rate(vllm_omni:diffusion_preprocess_s_sum[5m]) + / rate(vllm_omni:diffusion_exec_s_sum[5m]) +``` + +## Cross-Stage Transfer Metrics (`vllm_omni:`) Per-physical-transfer histograms tracking the data hop between adjacent stages. Labels `{model_name, from_stage, from_replica, to_stage, to_replica}` let dashboards attribute latency to specific replica edges. `from_replica` / `to_replica` are resolved from the orchestrator's sticky-routing binding (`stage_pool.get_bound_replica_id(request_id)`), so no extra plumbing through `TransferEdgeStats` is needed. | Metric | Type | Description | |--------|------|-------------| -| `vllm:omni_transfer_size_bytes` | Histogram | Per-transfer payload size in bytes | -| `vllm:omni_transfer_tx_s` | Histogram | Sender-side time (serialize + submit to connector) | -| `vllm:omni_transfer_rx_s` | Histogram | Receiver-side time (recv + deserialize) | -| `vllm:omni_transfer_in_flight_s` | Histogram | Network in-flight time (TX done → RX recv start) | +| `vllm_omni:transfer_size_bytes` | Histogram | Per-transfer payload size in bytes | +| `vllm_omni:transfer_tx_s` | Histogram | Sender-side time (serialize + submit to connector) | +| `vllm_omni:transfer_rx_s` | Histogram | Receiver-side time (recv + deserialize) | +| `vllm_omni:transfer_in_flight_s` | Histogram | Network in-flight time (TX done → RX recv start) | ## vLLM Engine Metrics (`vllm:`) @@ -83,9 +107,9 @@ For the full list of upstream metrics, see [the vLLM docs](https://github.com/vl | Metric group | Multi-stage LLM (Qwen3-Omni) | |---|---| -| `vllm:omni_` request tracking + latency | With `--log-stats` | -| `vllm:omni_` audio modality | With `--log-stats`, if pipeline has a talker stage | -| `vllm:omni_` transfer | With `--log-stats`, if pipeline has ≥ 2 stages | +| `vllm_omni:` request tracking + latency | With `--log-stats` | +| `vllm_omni:` audio modality | With `--log-stats`, if pipeline has a talker stage | +| `vllm_omni:` transfer | With `--log-stats`, if pipeline has ≥ 2 stages | | `vllm:` engine metrics (per `(stage, replica)`) | With `--log-stats` | | `vllm:` MFU metrics | With `--log-stats --enable-mfu-metrics` | @@ -94,5 +118,5 @@ For the full list of upstream metrics, see [the vLLM docs](https://github.com/vl - All time-bearing metrics use the `_s` suffix (values in seconds). Buckets are `SECONDS_BUCKETS` for e2e / generation-style values and `SECONDS_FAST_BUCKETS` (1 ms → 60 s) for the fine-grained transfer and audio-underrun values. - Counters use the `_total` suffix (auto-appended by `prometheus_client`). - Sizes use the `_bytes` suffix. -- All omni-specific families are prefixed `vllm:omni_`. The upstream `unregister_vllm_metrics()` function is monkey-patched (see `vllm_omni/patch.py`) to a scoped version that still strips upstream `vllm:*` collectors so multi-engine init within one process does not crash on duplicate registration, but preserves anything prefixed `vllm:omni_` / `vllm_omni`. -- Text and audio first-output use distinct families (`vllm:time_to_first_token_seconds` reused from upstream for text; `vllm:omni_audio_ttfp_s` for audio) rather than a single metric with a `modality` label. +- All omni-specific families are prefixed `vllm_omni:`. The upstream `unregister_vllm_metrics()` function is monkey-patched (see `vllm_omni/patch.py`) to a scoped version that still strips upstream `vllm:*` collectors so multi-engine init within one process does not crash on duplicate registration, but preserves anything prefixed `vllm_omni:`. +- Text and audio first-output use distinct families (`vllm:time_to_first_token_seconds` reused from upstream for text; `vllm_omni:audio_ttfp_s` for audio) rather than a single metric with a `modality` label. diff --git a/tests/diffusion/test_diffusion_output_formatter.py b/tests/diffusion/test_diffusion_output_formatter.py index eea53210c..dc21351ff 100644 --- a/tests/diffusion/test_diffusion_output_formatter.py +++ b/tests/diffusion/test_diffusion_output_formatter.py @@ -95,6 +95,7 @@ def test_formatter_preserves_single_video_audio_actions_and_metadata( "resolution": 512, "width": None, "height": None, + "num_inference_steps": 1, } diff --git a/tests/metrics/test_modality.py b/tests/metrics/test_modality.py index ba1785639..315594233 100644 --- a/tests/metrics/test_modality.py +++ b/tests/metrics/test_modality.py @@ -42,6 +42,10 @@ _EXPECTED_FAMILIES = [ defs.AUDIO_UNDERRUN_S, defs.AUDIO_CONTINUITY_OK_METRIC, defs.AUDIO_SKIPPED_REQUESTS_METRIC, + defs.DIFFUSION_EXEC_S, + defs.DIFFUSION_EXEC_PER_STEP_S, + defs.DIFFUSION_PREPROCESS_S, + defs.DIFFUSION_POSTPROCESS_S, ] @@ -55,6 +59,10 @@ class TestRegistration: mod.observe_audio_underrun("s", "r", 0.01) mod.inc_audio_continuity_ok("s", "r", 100) mod.inc_audio_skipped("s", "r", "malformed_codec") + mod.observe_diffusion_exec("s", "r", 0.5) + mod.observe_diffusion_exec_per_step("s", "r", 0.01) + mod.observe_diffusion_preprocess("s", "r", 0.01) + mod.observe_diffusion_postprocess("s", "r", 0.2) out = generate_latest(REGISTRY).decode() for name in _EXPECTED_FAMILIES: @@ -156,6 +164,50 @@ class TestAudio: assert unknown == 1.0 +# --------------------------------------------------------------------------- +# Diffusion observe API +# --------------------------------------------------------------------------- + + +class TestDiffusion: + def test_diffusion_exec_observed(self, mod: OmniModalityMetrics) -> None: + stage, replica = "dit_exec", "0" + mod.observe_diffusion_exec(stage, replica, 1.25) + out = generate_latest(REGISTRY).decode() + prefix = f'{defs.DIFFUSION_EXEC_S}_sum{{model_name="{_MODEL}",replica="{replica}",stage="{stage}"}}' + assert _sample_value(out, prefix) == 1.25 + + def test_diffusion_preprocess_observed(self, mod: OmniModalityMetrics) -> None: + stage, replica = "dit_pre", "0" + mod.observe_diffusion_preprocess(stage, replica, 0.03) + out = generate_latest(REGISTRY).decode() + prefix = f'{defs.DIFFUSION_PREPROCESS_S}_sum{{model_name="{_MODEL}",replica="{replica}",stage="{stage}"}}' + assert _sample_value(out, prefix) == pytest.approx(0.03) + + def test_diffusion_postprocess_observed(self, mod: OmniModalityMetrics) -> None: + stage, replica = "dit_post", "0" + mod.observe_diffusion_postprocess(stage, replica, 0.8) + out = generate_latest(REGISTRY).decode() + prefix = f'{defs.DIFFUSION_POSTPROCESS_S}_sum{{model_name="{_MODEL}",replica="{replica}",stage="{stage}"}}' + assert _sample_value(out, prefix) == 0.8 + + def test_diffusion_exec_per_step_observed(self, mod: OmniModalityMetrics) -> None: + stage, replica = "dit_per_step", "0" + mod.observe_diffusion_exec_per_step(stage, replica, 0.078) + out = generate_latest(REGISTRY).decode() + prefix = f'{defs.DIFFUSION_EXEC_PER_STEP_S}_sum{{model_name="{_MODEL}",replica="{replica}",stage="{stage}"}}' + assert _sample_value(out, prefix) == pytest.approx(0.078) + + def test_log_stats_false_skips(self) -> None: + silent = OmniModalityMetrics(model_name="silent-model", log_stats=False) + silent.observe_diffusion_exec("s", "r", 1.0) + silent.observe_diffusion_exec_per_step("s", "r", 0.02) + silent.observe_diffusion_preprocess("s", "r", 0.01) + silent.observe_diffusion_postprocess("s", "r", 0.5) + out = generate_latest(REGISTRY).decode() + assert 'model_name="silent-model"' not in out + + # --------------------------------------------------------------------------- # Stub-driven routing tests for the finalize / streaming helpers. # --------------------------------------------------------------------------- @@ -183,6 +235,18 @@ class _StubModMetrics: def inc_audio_skipped(self, s, r, reason): self.calls.append(("inc_audio_skipped", s, r, reason)) + def observe_diffusion_exec(self, s, r, v): + self.calls.append(("observe_diffusion_exec", s, r, v)) + + def observe_diffusion_exec_per_step(self, s, r, v): + self.calls.append(("observe_diffusion_exec_per_step", s, r, v)) + + def observe_diffusion_preprocess(self, s, r, v): + self.calls.append(("observe_diffusion_preprocess", s, r, v)) + + def observe_diffusion_postprocess(self, s, r, v): + self.calls.append(("observe_diffusion_postprocess", s, r, v)) + class _Bag: def __init__(self, **kw): @@ -239,8 +303,88 @@ class TestObserveModalityAtFinalize: ) assert ("observe_audio_duration", "1", "0", 1.0) in stub.calls - def test_non_audio_output_type_skipped(self): - # Image / video / text output types are out of scope for this module. + def test_diffusion_path_full(self): + stub = _StubModMetrics() + stage_metrics = _Bag( + stage_gen_time_ms=1000.0, + diffusion_metrics={ + "preprocess_time_s": 0.05, + "diffusion_engine_exec_time_s": 1.2, + "postprocess_time_s": 0.3, + "image_num": 1, + "resolution": 1024, + "num_inference_steps": 50, + }, + ) + observe_modality_at_finalize( + stub, + output_type="image", + stage_id=2, + replica_id=0, + stage_metrics=stage_metrics, + engine_outputs=_Bag(), + ) + assert ("observe_diffusion_preprocess", "2", "0", 0.05) in stub.calls + assert ("observe_diffusion_exec", "2", "0", 1.2) in stub.calls + assert ("observe_diffusion_postprocess", "2", "0", 0.3) in stub.calls + assert ("observe_diffusion_exec_per_step", "2", "0", pytest.approx(0.024)) in stub.calls + + def test_diffusion_per_step_skipped_without_num_inference_steps(self): + stub = _StubModMetrics() + stage_metrics = _Bag( + stage_gen_time_ms=1000.0, + diffusion_metrics={ + "diffusion_engine_exec_time_s": 1.2, + "preprocess_time_s": 0.05, + "postprocess_time_s": 0.3, + }, + ) + observe_modality_at_finalize( + stub, + output_type="image", + stage_id=2, + replica_id=0, + stage_metrics=stage_metrics, + engine_outputs=_Bag(), + ) + assert ("observe_diffusion_exec", "2", "0", 1.2) in stub.calls + assert not any(c[0] == "observe_diffusion_exec_per_step" for c in stub.calls) + + def test_diffusion_metrics_none_skipped(self): + stub = _StubModMetrics() + observe_modality_at_finalize( + stub, + output_type="image", + stage_id=2, + replica_id=0, + stage_metrics=_Bag(stage_gen_time_ms=500.0, diffusion_metrics=None), + engine_outputs=_Bag(), + ) + assert not any(c[0].startswith("observe_diffusion") for c in stub.calls) + + def test_diffusion_zero_values_observed(self): + stub = _StubModMetrics() + stage_metrics = _Bag( + stage_gen_time_ms=500.0, + diffusion_metrics={ + "preprocess_time_s": 0.0, + "diffusion_engine_exec_time_s": 0.0, + "postprocess_time_s": 0.0, + }, + ) + observe_modality_at_finalize( + stub, + output_type="image", + stage_id=2, + replica_id=0, + stage_metrics=stage_metrics, + engine_outputs=_Bag(), + ) + assert ("observe_diffusion_preprocess", "2", "0", 0.0) in stub.calls + assert ("observe_diffusion_exec", "2", "0", 0.0) in stub.calls + assert ("observe_diffusion_postprocess", "2", "0", 0.0) in stub.calls + + def test_non_audio_output_type_skips_audio_but_still_checks_diffusion(self): stub = _StubModMetrics() for output_type in ("text", "image", "video"): observe_modality_at_finalize( @@ -251,7 +395,7 @@ class TestObserveModalityAtFinalize: stage_metrics=_Bag(stage_gen_time_ms=100.0), engine_outputs=_Bag(), ) - assert stub.calls == [] + assert not any(c[0].startswith("observe_audio") or c[0].startswith("inc_audio") for c in stub.calls) def test_replica_id_none_skipped(self): stub = _StubModMetrics() diff --git a/vllm_omni/diffusion/output_formatter.py b/vllm_omni/diffusion/output_formatter.py index d5b03443c..4e68bfd25 100644 --- a/vllm_omni/diffusion/output_formatter.py +++ b/vllm_omni/diffusion/output_formatter.py @@ -138,6 +138,7 @@ def format_diffusion_outputs( ), "width": request.sampling_params.width, "height": request.sampling_params.height, + "num_inference_steps": request.sampling_params.num_inference_steps, } # Only the primary payload determines the response type. Some image models diff --git a/vllm_omni/entrypoints/omni_base.py b/vllm_omni/entrypoints/omni_base.py index 88e5a5edf..b787c44a4 100644 --- a/vllm_omni/entrypoints/omni_base.py +++ b/vllm_omni/entrypoints/omni_base.py @@ -568,6 +568,7 @@ class OmniBase(PDDisaggregationMixin): msg_id = id(result) consumed = self._consumed_metric_message_ids(req_id) if msg_id not in consumed: + metrics.accumulate_diffusion_metrics(stage_meta.stage_type, req_id, engine_outputs) metrics.on_stage_metrics(stage_id, req_id, _m, output_type) consumed.add(msg_id) diff --git a/vllm_omni/metrics/definitions.py b/vllm_omni/metrics/definitions.py index 23a1364a4..b5d5217f5 100644 --- a/vllm_omni/metrics/definitions.py +++ b/vllm_omni/metrics/definitions.py @@ -134,6 +134,15 @@ AUDIO_CONTINUITY_OK_METRIC = METRIC_PREFIX + AUDIO_CONTINUITY_OK AUDIO_SKIPPED_REQUESTS_METRIC = METRIC_PREFIX + AUDIO_SKIPPED_REQUESTS +# ============================================================================ +# Diffusion family (per-stage + per-replica diffusion timing breakdowns) +# ============================================================================ +DIFFUSION_EXEC_S = METRIC_PREFIX + "diffusion_exec_s" +DIFFUSION_EXEC_PER_STEP_S = METRIC_PREFIX + "diffusion_exec_per_step_s" +DIFFUSION_PREPROCESS_S = METRIC_PREFIX + "diffusion_preprocess_s" +DIFFUSION_POSTPROCESS_S = METRIC_PREFIX + "diffusion_postprocess_s" + + # ============================================================================ # Cross-stage Transfer family (per-physical-hop TX/RX/in-flight timings) # ============================================================================ diff --git a/vllm_omni/metrics/modality.py b/vllm_omni/metrics/modality.py index fd455ad36..b83912ed3 100644 --- a/vllm_omni/metrics/modality.py +++ b/vllm_omni/metrics/modality.py @@ -78,6 +78,35 @@ _audio_skipped_family = Counter( ) +# ---------------------------------------------------------------------------- +# Diffusion family +# ---------------------------------------------------------------------------- +_diffusion_exec_family = Histogram( + defs.DIFFUSION_EXEC_S, + "DiT forward pass execution time per request in seconds.", + labelnames=_stage_labels, + buckets=defs.SECONDS_BUCKETS, +) +_diffusion_exec_per_step_family = Histogram( + defs.DIFFUSION_EXEC_PER_STEP_S, + "DiT forward pass execution time per denoising step in seconds.", + labelnames=_stage_labels, + buckets=defs.SECONDS_FAST_BUCKETS, +) +_diffusion_preprocess_family = Histogram( + defs.DIFFUSION_PREPROCESS_S, + "Diffusion input preprocessing time per request in seconds.", + labelnames=_stage_labels, + buckets=defs.SECONDS_FAST_BUCKETS, +) +_diffusion_postprocess_family = Histogram( + defs.DIFFUSION_POSTPROCESS_S, + "Diffusion output postprocessing (VAE decode) time per request in seconds.", + labelnames=_stage_labels, + buckets=defs.SECONDS_FAST_BUCKETS, +) + + class OmniModalityMetrics: """Per-modality observe API. Stage/replica are passed at observe time because a single OmniModalityMetrics instance per pipeline serves all @@ -139,6 +168,44 @@ class OmniModalityMetrics: reason=reason or "unknown", ).inc() + # ---- Diffusion -------------------------------------------------------- + + def observe_diffusion_exec(self, stage: str, replica: str, seconds: float) -> None: + if not self._log_stats: + return + _diffusion_exec_family.labels( + model_name=self._model_name, + stage=stage, + replica=replica, + ).observe(seconds) + + def observe_diffusion_exec_per_step(self, stage: str, replica: str, seconds: float) -> None: + if not self._log_stats: + return + _diffusion_exec_per_step_family.labels( + model_name=self._model_name, + stage=stage, + replica=replica, + ).observe(seconds) + + def observe_diffusion_preprocess(self, stage: str, replica: str, seconds: float) -> None: + if not self._log_stats: + return + _diffusion_preprocess_family.labels( + model_name=self._model_name, + stage=stage, + replica=replica, + ).observe(seconds) + + def observe_diffusion_postprocess(self, stage: str, replica: str, seconds: float) -> None: + if not self._log_stats: + return + _diffusion_postprocess_family.labels( + model_name=self._model_name, + stage=stage, + replica=replica, + ).observe(seconds) + def observe_modality_at_finalize( mod_metrics: OmniModalityMetrics, @@ -161,43 +228,49 @@ def observe_modality_at_finalize( """ if replica_id is None or stage_metrics is None or output_type is None: return - if output_type != "audio": - return stage_label = str(stage_id) replica_label = str(replica_id) gen_time_s = float(getattr(stage_metrics, "stage_gen_time_ms", 0.0)) / 1000.0 mm_out = extract_mm_output(engine_outputs) - sample_rate = defs.resolve_audio_sample_rate(mm_out) - # `stage_metrics.audio_generated_frames` is the legacy per-chunk - # accumulator field on StageRequestStats. No production path currently - # fills it, so the fallback below is the live source — but we leave the - # field lookup in place in case the accumulator gets re-wired upstream. - n_frames = int(getattr(stage_metrics, "audio_generated_frames", 0) or 0) - if n_frames == 0: - n_frames = count_audio_frames(mm_out) - mod_metrics.inc_audio_frames(stage_label, replica_label, n_frames) - duration_s = n_frames / sample_rate if sample_rate > 0 else 0.0 - if duration_s > 0: - mod_metrics.observe_audio_duration(stage_label, replica_label, duration_s) - mod_metrics.observe_audio_rtf( - stage_label, - replica_label, - defs.compute_audio_rtf(gen_time_s, duration_s), - ) - else: - # Request completed (finish_reason ∈ {stop, length} — error paths - # don't reach finalize) but no audio samples were produced. Covers - # silent `return None` skips in the talker→code2wav stage - # processors and the `parsed.append((0,0))` malformed-length path - # in qwen3-tts code2wav. raise-paths surface via the upstream - # vllm:request_success_total{finished_reason="error"} channel and - # never reach this branch. - mod_metrics.inc_audio_skipped(stage_label, replica_label, "no_audio_data") - # audio_underrun / continuity are emitted from the streaming path in - # observe_audio_streaming_finalize; finalize is too late for the - # per-chunk timeline they need. + if output_type == "audio": + sample_rate = defs.resolve_audio_sample_rate(mm_out) + n_frames = int(getattr(stage_metrics, "audio_generated_frames", 0) or 0) + if n_frames == 0: + n_frames = count_audio_frames(mm_out) + mod_metrics.inc_audio_frames(stage_label, replica_label, n_frames) + duration_s = n_frames / sample_rate if sample_rate > 0 else 0.0 + if duration_s > 0: + mod_metrics.observe_audio_duration(stage_label, replica_label, duration_s) + mod_metrics.observe_audio_rtf( + stage_label, + replica_label, + defs.compute_audio_rtf(gen_time_s, duration_s), + ) + else: + mod_metrics.inc_audio_skipped(stage_label, replica_label, "no_audio_data") + + dm = getattr(stage_metrics, "diffusion_metrics", None) + if dm: + _key_map = { + "diffusion_engine_exec_time_s": mod_metrics.observe_diffusion_exec, + "preprocess_time_s": mod_metrics.observe_diffusion_preprocess, + "postprocess_time_s": mod_metrics.observe_diffusion_postprocess, + } + for key, observe_fn in _key_map.items(): + val = dm.get(key) + if val is not None: + observe_fn(stage_label, replica_label, float(val)) + + exec_time = dm.get("diffusion_engine_exec_time_s") + num_steps = dm.get("num_inference_steps") + if exec_time is not None and num_steps and num_steps > 0: + mod_metrics.observe_diffusion_exec_per_step( + stage_label, + replica_label, + float(exec_time) / int(num_steps), + ) def observe_audio_first_packet( diff --git a/vllm_omni/metrics/stats.py b/vllm_omni/metrics/stats.py index 3d08adae0..95198ef6a 100644 --- a/vllm_omni/metrics/stats.py +++ b/vllm_omni/metrics/stats.py @@ -44,7 +44,7 @@ class StageRequestStats: final_output_type: str | None = None request_id: str | None = None postprocess_time_ms: float = 0.0 - diffusion_metrics: dict[str, int] = None + diffusion_metrics: dict[str, float] = None audio_generated_frames: int = 0 audio_sample_rate: int = 0 audio_duration_s: float = 0.0 @@ -579,7 +579,7 @@ class OrchestratorAggregator: if final_output_type is not None: stats.final_output_type = final_output_type stats.diffusion_metrics = ( - {k: int(v) for k, v in self.diffusion_metrics.pop(req_id, {}).items()} + {k: float(v) for k, v in self.diffusion_metrics.pop(req_id, {}).items()} if req_id in self.diffusion_metrics else None ) @@ -644,8 +644,19 @@ class OrchestratorAggregator: if isinstance(diffusion_metrics, list): diffusion_metrics = diffusion_metrics[0] if diffusion_metrics: + _MS_TO_S = { + "diffusion_engine_exec_time_ms": "diffusion_engine_exec_time_s", + "preprocess_time_ms": "preprocess_time_s", + "postprocess_time_ms": "postprocess_time_s", + "diffusion_engine_total_time_ms": "diffusion_engine_total_time_s", + } for key, value in diffusion_metrics.items(): - self.diffusion_metrics[req_id][key] += value + if value is None: + continue + if key in _MS_TO_S: + self.diffusion_metrics[req_id][_MS_TO_S[key]] = float(value) / 1000.0 + else: + self.diffusion_metrics[req_id][key] = value def on_forward( self,