[Metrics] Count finished requests by watermark status (#58715)

Signed-off-by: Simon Veitner <sveitner@redhat.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Lucas Wilkinson <LucasWilkinson@users.noreply.github.com>

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

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."""

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()

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))

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,

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
         )

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."""