From a7b7bc1fbe1f5c4e47c53b0f02a3ecfbcd8054b9 Mon Sep 17 00:00:00 2001 From: simon-veitner-redhat Date: Mon, 12 Oct 2026 19:21:22 +0200 Subject: [PATCH] [Metrics] Count finished requests by watermark status (#58715) Signed-off-by: Simon Veitner Co-authored-by: Claude Co-authored-by: Lucas Wilkinson --- docs/features/watermarking.md | 28 ++++++ tests/v1/engine/test_output_processor.py | 94 +++++++++++++++++++++ tests/v1/metrics/test_histogram_buckets.py | 14 +++ tests/watermarking/test_watermarking_e2e.py | 16 ++++ vllm/v1/engine/output_processor.py | 6 +- vllm/v1/metrics/loggers.py | 30 +++++++ vllm/v1/metrics/stats.py | 16 ++++ 7 files changed, 203 insertions(+), 1 deletion(-) diff --git a/docs/features/watermarking.md b/docs/features/watermarking.md index 66b6465eb8..f51ec72eb1 100644 --- a/docs/features/watermarking.md +++ b/docs/features/watermarking.md @@ -56,6 +56,34 @@ watermarking algorithm that does not support it natively, at the cost of weaker detectability. See [Speculative decoding](#speculative-decoding). +## Monitoring + +When `watermark_config` is set, the Prometheus counter +`vllm:watermark_requests_total` counts finished requests by `status`: + +- `watermarked`: the request was sampled with the watermark switched on. +- `skipped`: the request asked for the watermark (omitted or `true`) but could + not use it, such as `temperature=0` and trace replay requests. A beam search + request that asks for the watermark runs without it and logs a warning. Its + beam steps count as `not_watermarked`, not `skipped`. +- `not_watermarked`: all other finished requests. This includes explicit + opt-outs, pooling requests, and the internal requests of beam search (one + per beam step), generative scoring and language detection. + +Each finished request has exactly one status, so the sum over `status` equals +`vllm:request_success_total` on the same instance. Requests that the client +aborts are in neither counter. The counter does not exist without +`watermark_config`. With `n > 1`, each sequence counts once. In a +disaggregated prefill/decode deployment, both instances count a request: sum +over decode instances only. For streaming input, the first input chunk sets +the status. + +The status is per request, not per token. A `watermarked` request can contain +tokens without the watermark: accepted draft tokens with +`allow_target_only_watermarking`, tokens whose context repeats +(`deduplicate_contexts`), the first `context_width` generated tokens with +`deduplicate_contexts: "all"`, and tokens that a grammar forces. + ## Architecture `WatermarkConfig` selects an algorithm and PRF. Model Runner V2 constructs the diff --git a/tests/v1/engine/test_output_processor.py b/tests/v1/engine/test_output_processor.py index 8eb8576599..3c186a4e73 100644 --- a/tests/v1/engine/test_output_processor.py +++ b/tests/v1/engine/test_output_processor.py @@ -6,6 +6,7 @@ import time from unittest.mock import MagicMock, Mock import numpy as np +import prometheus_client import pytest import torch @@ -17,6 +18,7 @@ from tests.v1.engine.utils import ( MockEngineCore, ) from vllm import PoolingParams +from vllm.config.watermarking import WatermarkConfig from vllm.logprobs import FlatLogprobs, Logprob, PromptLogprobs, SampleLogprobs from vllm.lora.request import LoRARequest from vllm.outputs import ( @@ -41,6 +43,9 @@ from vllm.v1.engine.output_processor import ( RequestOutputCollector, RequestState, ) +from vllm.v1.engine.parallel_sampling import ParentRequest +from vllm.v1.metrics.loggers import PrometheusStatLogger +from vllm.v1.metrics.prometheus import unregister_vllm_metrics from vllm.v1.metrics.stats import IterationStats, PrefillStats, SchedulerStats from vllm.v1.outputs import SamplingMaskLists @@ -1169,6 +1174,95 @@ def test_iteration_stats(dummy_test_vectors): assert iteration_stats.num_generation_tokens == num_active +def test_watermark_status_partitions_finished_requests(dummy_test_vectors): + """Each finished request gets exactly one watermark status.""" + output_processor = OutputProcessor(dummy_test_vectors.tokenizer, log_stats=True) + prompt_token_ids = dummy_test_vectors.prompt_tokens[0] + + def make_request( + request_id: str, + sampling_params: SamplingParams | None = None, + pooling_params: PoolingParams | None = None, + ) -> EngineCoreRequest: + return EngineCoreRequest( + request_id=request_id, + external_req_id=f"{request_id}-ext", + prompt_token_ids=prompt_token_ids, + mm_features=None, + arrival_time=0, + lora_request=None, + cache_salt=None, + data_parallel_rank=None, + sampling_params=sampling_params, + pooling_params=pooling_params, + ) + + # The frontend resolves the flags before the request reaches the + # OutputProcessor; set them as it does. + def resolved(watermarking: bool, skipped: bool, n: int = 1) -> SamplingParams: + params = SamplingParams(n=n, watermarking=watermarking) + params._watermarking_skipped = skipped + return params + + finish_reasons = { + "wm": FinishReason.ABORT, + "skip": FinishReason.STOP, + # A re-resolved request can carry both flags; it is watermarked. + "sticky": FinishReason.STOP, + "off": FinishReason.ERROR, + "pool": FinishReason.STOP, + } + output_processor.add_request(make_request("wm", resolved(True, False)), None) + output_processor.add_request(make_request("skip", resolved(False, True)), None) + output_processor.add_request(make_request("sticky", resolved(True, True)), None) + output_processor.add_request(make_request("off", resolved(False, False)), None) + output_processor.add_request( + make_request("pool", pooling_params=PoolingParams(task="embed")), None + ) + parent_req = ParentRequest(make_request("parent", resolved(True, False, n=2))) + for idx in range(2): + child_id, child_params = parent_req.get_child_info(idx) + output_processor.add_request( + make_request(child_id, child_params), None, parent_req, idx + ) + finish_reasons[child_id] = FinishReason.LENGTH + + outputs = [ + EngineCoreOutput( + request_id=request_id, + new_token_ids=[] if request_id == "pool" else [prompt_token_ids[0]], + pooling_output=torch.tensor([1.0]) if request_id == "pool" else None, + finish_reason=finish_reason, + ) + for request_id, finish_reason in finish_reasons.items() + ] + iteration_stats = IterationStats() + output_processor.process_outputs(outputs, time.monotonic(), iteration_stats) + + vllm_config = dummy_test_vectors.vllm_config + vllm_config.watermark_config = WatermarkConfig(key=42) + try: + PrometheusStatLogger(vllm_config).record( + scheduler_stats=None, iteration_stats=iteration_stats + ) + totals: dict[str, dict[str, float]] = {} + for metric in prometheus_client.REGISTRY.collect(): + for sample in metric.samples: + if sample.name in ( + "vllm:watermark_requests_total", + "vllm:request_success_total", + ): + label = sample.labels.get("status", "all") + by_label = totals.setdefault(sample.name, {}) + by_label[label] = by_label.get(label, 0) + sample.value + finally: + unregister_vllm_metrics() + + status = totals["vllm:watermark_requests_total"] + assert status == {"watermarked": 4, "skipped": 1, "not_watermarked": 2} + assert sum(status.values()) == totals["vllm:request_success_total"]["all"] + + @pytest.mark.parametrize("log_stats", [True, False]) def test_lora_request_tracking(log_stats: bool, dummy_test_vectors): """Test LoRA request lifecycle tracking through waiting -> running -> finished.""" diff --git a/tests/v1/metrics/test_histogram_buckets.py b/tests/v1/metrics/test_histogram_buckets.py index cf905b5d8b..c5d41db054 100644 --- a/tests/v1/metrics/test_histogram_buckets.py +++ b/tests/v1/metrics/test_histogram_buckets.py @@ -11,6 +11,7 @@ import pytest from vllm.config import ModelConfig, VllmConfig from vllm.config.observability import ObservabilityConfig +from vllm.config.watermarking import WatermarkConfig from vllm.v1.metrics.buckets import ( BUCKET_FAMILY_KEYS, build_1_2_5_buckets, @@ -317,3 +318,16 @@ def test_prometheus_logger_applies_overrides(): assert found[metric_name] == overridden[family], metric_name finally: unregister_vllm_metrics() + + +@pytest.mark.parametrize("with_config", [True, False]) +def test_watermark_requests_counter_registration(with_config: bool): + """The watermark counter must exist only with watermark_config.""" + config = build_logger_config(ObservabilityConfig()) + config.watermark_config = WatermarkConfig(key=42) if with_config else None + try: + PrometheusStatLogger(config) + names = {m.name for m in prometheus_client.REGISTRY.collect()} + assert ("vllm:watermark_requests" in names) == with_config + finally: + unregister_vllm_metrics() diff --git a/tests/watermarking/test_watermarking_e2e.py b/tests/watermarking/test_watermarking_e2e.py index c73d48f6b6..ce6433e613 100644 --- a/tests/watermarking/test_watermarking_e2e.py +++ b/tests/watermarking/test_watermarking_e2e.py @@ -5,6 +5,7 @@ import pytest from vllm import SamplingParams from vllm.platforms import current_platform +from vllm.v1.metrics.reader import Counter from vllm.v1.watermarking import GumbelWatermarkDetector @@ -23,14 +24,29 @@ def test_llm_generated_sequence_is_watermarked(vllm_runner): enforce_eager=True, max_model_len=512, gpu_memory_utilization=0.2, + disable_log_stats=False, ) request = "Tell me a story about an explorer who discovers a mysterious island." params_use_wm = SamplingParams(temperature=1.0, max_tokens=256, watermarking=True) params_no_wm = SamplingParams(temperature=1.0, max_tokens=256, watermarking=False) + params_greedy = SamplingParams(temperature=0.0, max_tokens=16) with runner: output_use_wm = runner.llm.generate(request, params_use_wm) output_no_wm = runner.llm.generate(request, params_no_wm) + runner.llm.generate(request, params_greedy) + metrics = runner.llm.get_metrics() + + # The greedy request is skipped; the opt-out one is not watermarked. + counters = [m for m in metrics if isinstance(m, Counter)] + status = { + m.labels["status"]: m.value + for m in counters + if m.name == "vllm:watermark_requests" + } + assert status == {"watermarked": 1, "skipped": 1, "not_watermarked": 1} + num_success = sum(m.value for m in counters if m.name == "vllm:request_success") + assert sum(status.values()) == num_success result_use_wm = detector.detect(list(output_use_wm[0].outputs[0].token_ids)) result_no_wm = detector.detect(list(output_no_wm[0].outputs[0].token_ids)) diff --git a/vllm/v1/engine/output_processor.py b/vllm/v1/engine/output_processor.py index 263a28fb39..eb499f6cc7 100644 --- a/vllm/v1/engine/output_processor.py +++ b/vllm/v1/engine/output_processor.py @@ -276,7 +276,7 @@ class RequestState: output_kind = request.pooling_params.output_kind assert request.external_req_id is not None - return cls( + state = cls( request_id=request.request_id, external_req_id=request.external_req_id, parent_req=parent_req, @@ -299,6 +299,10 @@ class RequestState: stream_input=request.resumable, remote_prefill_cached_tokens=remote_prefill_cached_tokens, ) + if state.stats is not None and sampling_params is not None: + state.stats.is_watermarked = bool(sampling_params.watermarking) + state.stats.is_watermark_skipped = sampling_params._watermarking_skipped + return state def make_request_output( self, diff --git a/vllm/v1/metrics/loggers.py b/vllm/v1/metrics/loggers.py index c42d177909..65acc49ac0 100644 --- a/vllm/v1/metrics/loggers.py +++ b/vllm/v1/metrics/loggers.py @@ -639,6 +639,25 @@ class PrometheusStatLogger(AggregateStatLoggerBase): counter_corrupted_requests, per_engine_labelvalues ) + if vllm_config.watermark_config is not None: + counter_watermark_requests = self._counter_cls( + name="vllm:watermark_requests", + documentation=( + "Finished requests by watermark status " + "(only with watermark_config)." + ), + labelnames=labelnames + ["status"], + ) + self.counter_watermark_requests: dict[str, dict[int, Counter]] = {} + for status in ("watermarked", "skipped", "not_watermarked"): + per_engine_labelvalues_with_status = { + idx: labelvalues + [status] + for idx, labelvalues in per_engine_labelvalues.items() + } + self.counter_watermark_requests[status] = create_metric_per_engine( + counter_watermark_requests, per_engine_labelvalues_with_status + ) + counter_prefix_cache_queries = self._counter_cls( name="vllm:prefix_cache_queries", documentation=( @@ -1226,6 +1245,17 @@ class PrometheusStatLogger(AggregateStatLoggerBase): self.counter_corrupted_requests[engine_idx].inc( iteration_stats.num_corrupted_reqs ) + if self.vllm_config.watermark_config is not None: + counter_watermark = self.counter_watermark_requests + counter_watermark["watermarked"][engine_idx].inc( + iteration_stats.num_watermarked_reqs + ) + counter_watermark["skipped"][engine_idx].inc( + iteration_stats.num_watermark_skipped_reqs + ) + counter_watermark["not_watermarked"][engine_idx].inc( + iteration_stats.num_not_watermarked_reqs + ) self.counter_num_preempted_reqs[engine_idx].inc( iteration_stats.num_preempted_reqs ) diff --git a/vllm/v1/metrics/stats.py b/vllm/v1/metrics/stats.py index 0046dcefa9..87277f6545 100644 --- a/vllm/v1/metrics/stats.py +++ b/vllm/v1/metrics/stats.py @@ -277,6 +277,12 @@ class RequestStateStats: # Track if this request is corrupted (NaNs in logits) is_corrupted: bool = False + # Track if this request was sampled with the engine watermark + is_watermarked: bool = False + + # Track if this request asked for the watermark but could not use it + is_watermark_skipped: bool = False + @dataclass class FinishedRequestStats: @@ -479,6 +485,9 @@ class IterationStats: self.time_to_first_tokens_iter: list[float] = [] self.inter_token_latencies_iter: list[float] = [] self.num_corrupted_reqs: int = 0 + self.num_watermarked_reqs: int = 0 + self.num_watermark_skipped_reqs: int = 0 + self.num_not_watermarked_reqs: int = 0 def __repr__(self) -> str: field_to_value_str = ", ".join(f"{k}={v}" for k, v in vars(self).items()) @@ -624,6 +633,13 @@ class IterationStats: if req_stats.is_corrupted: self.num_corrupted_reqs += 1 + if req_stats.is_watermarked: + self.num_watermarked_reqs += 1 + elif req_stats.is_watermark_skipped: + self.num_watermark_skipped_reqs += 1 + else: + self.num_not_watermarked_reqs += 1 + class LoRAStats: """Tracks waiting and running request IDs for a single LoRA."""