Add vllm:kv_offload_cpu_total_blocks capacity metric

tests/v1/kv_connector/unit/offloading_connector/test_metrics.py

@@ -24,8 +24,13 @@
     OffloadingGaugeMetadata,
     OffloadingHistogramMetadata,
 )
+from vllm.v1.kv_offload.cpu.common import (
+    CPU_CONFIG_INFO_LABELS,
+    CPUOffloadingMetrics,
+)
 from vllm.v1.kv_offload.factory import OffloadingSpecFactory
 
+CPU_CONFIG_INFO = CPUOffloadingMetrics.CPU_CONFIG_INFO
 LOAD_BYTES = _TransferMetricName.LOAD_BYTES
 LOAD_TIME = _TransferMetricName.LOAD_TIME
 LOAD_SIZE = _TransferMetricName.LOAD_SIZE
@@ -593,6 +598,70 @@ def test_prom_metrics_observes_manager_gauge_and_histogram():
     assert histogram_def.kwargs["buckets"] == (0.1, 1.0)
 
 
+def test_prom_metrics_gauges_collapse_multiprocess_fanout():
+    """Gauges must not fan out per pid: they are point-in-time values for one
+    engine that every API-server process reports. Counters and histograms are
+    per-process cumulative and stay summed."""
+    metric_definitions = {
+        PENDING_STORES: OffloadingGaugeMetadata(documentation="pending stores"),
+        STORES_SKIPPED: OffloadingCounterMetadata(documentation="stores skipped"),
+        LOOKUP_LATENCY: OffloadingHistogramMetadata(documentation="lookup latency"),
+    }
+    with patch.object(
+        OffloadingSpecFactory,
+        "get_spec_cls",
+        return_value=_spec_cls_with_metric_definitions(metric_definitions),
+    ):
+        prom_metrics = OffloadPromMetrics(
+            vllm_config=_FakeVllmConfig(store_threshold=0),  # type: ignore[arg-type]
+            metric_types={
+                Gauge: _FakeMetric,
+                Counter: _FakeMetric,
+                Histogram: _FakeMetric,
+            },
+            labelnames=["model_name", "engine"],
+            per_engine_labelvalues={0: ["model", "0"]},
+        )
+
+    metric_defs = prom_metrics._offloading_metric_defs
+    assert metric_defs[PENDING_STORES].kwargs["multiprocess_mode"] == "mostrecent"
+    assert "multiprocess_mode" not in metric_defs[STORES_SKIPPED].kwargs
+    assert "multiprocess_mode" not in metric_defs[LOOKUP_LATENCY].kwargs
+
+
+def test_prom_metrics_record_config_info():
+    """Static config is emitted at startup as a gauge pinned to 1, carrying the
+    config in its labels (cache_config_info-style), with no stats/observe."""
+    prom_metrics = OffloadPromMetrics(
+        vllm_config=_FakeVllmConfig(store_threshold=0),  # type: ignore[arg-type]
+        metric_types={
+            Gauge: _FakeMetric,
+            Counter: _FakeMetric,
+            Histogram: _FakeMetric,
+        },
+        labelnames=["model_name", "engine"],
+        per_engine_labelvalues={0: ["model", "0"]},
+    )
+
+    labels = {
+        "num_blocks": "4",
+        "blocks_per_chunk": "8",
+        "kv_bytes_per_chunk": "1024",
+        "cpu_page_size_per_worker": "512",
+        "eviction_policy": "lru",
+    }
+    # Metrics this prom does not own are skipped, so MultiConnector can fan the
+    # merged config info out to every child prom.
+    prom_metrics.record_config_info(
+        {CPU_CONFIG_INFO: labels, "vllm:someone_elses_config_info": {}}
+    )
+
+    labelvalues = tuple(labels[name] for name in CPU_CONFIG_INFO_LABELS)
+    gauge = prom_metrics.offloading_metrics[(0, CPU_CONFIG_INFO, labelvalues)]
+    assert gauge.set_values == [1]
+    assert gauge.labelvalues == ("model", "0", *labelvalues)
+
+
 def test_prom_metrics_lazily_observes_labeled_metric():
     metric_definitions = {
         MY_COUNTER: OffloadingCounterMetadata(

tests/v1/kv_connector/unit/test_multi_connector.py

@@ -879,6 +879,23 @@ def test_multi_connector_prefer_cross_layer_blocks(mc):
     assert mc.prefer_cross_layer_blocks is True
 
 
+def test_multi_connector_aggregates_config_info(mc):
+    """Nested connectors' startup config info is merged flat (metric_name ->
+    labels), so a wrapped connector still emits it at startup."""
+    mc._connectors[0].get_config_info = lambda: None
+    mc._connectors[1].get_config_info = lambda: None
+    assert mc.get_config_info() is None
+
+    mc._connectors[0].get_config_info = lambda: {"vllm:first_info": {"a": "1"}}
+    assert mc.get_config_info() == {"vllm:first_info": {"a": "1"}}
+
+    mc._connectors[1].get_config_info = lambda: {"vllm:second_info": {"b": "2"}}
+    assert mc.get_config_info() == {
+        "vllm:first_info": {"a": "1"},
+        "vllm:second_info": {"b": "2"},
+    }
+
+
 def test_multi_connector_worker_metadata(mc):
     class MockConnectorWorkerMetadata(KVConnectorWorkerMetadata):
         def __init__(self, data: set[str]):

tests/v1/kv_offload/cpu/test_manager.py

@@ -41,9 +41,11 @@ def make_cpu_manager(
     enable_events: bool = False,
     store_threshold: int = 0,
     max_tracker_size: int = 64_000,
+    config_info: dict[str, str] | None = None,
 ) -> CPUOffloadingManager:
     return CPUOffloadingManager(
         num_blocks=num_blocks,
+        config_info=config_info,
         cache_policy=cache_policy,
         cache_policy_module_path=cache_policy_module_path,
         enable_events=enable_events,
@@ -261,6 +263,33 @@ def check_usage_stats(manager: CPUOffloadingManager, value: float):
     check_usage_stats(manager, 0.0)
 
 
+def test_cpu_manager_reports_config_info():
+    config_info = {
+        "num_blocks": "4",
+        "blocks_per_chunk": "8",
+        "kv_bytes_per_chunk": "1024",
+        "cpu_page_size_per_worker": "512",
+        "eviction_policy": "lru",
+    }
+    manager = make_cpu_manager(num_blocks=4, config_info=config_info)
+
+    # Exposed for one-shot startup emission via get_config_info(), keyed by
+    # metric name.
+    assert manager.get_config_info() == {
+        CPUOffloadingMetrics.CPU_CONFIG_INFO: config_info
+    }
+
+    # It is NOT put on the per-interval stats path: emitting a constant gauge
+    # every interval would re-serialize it over IPC for no reason; startup
+    # emission covers it instead (and the gauge persists).
+    stats = manager.get_stats()
+    assert stats is not None
+    assert CPUOffloadingMetrics.CPU_CONFIG_INFO not in stats._values
+
+    # A manager created without config info exposes nothing.
+    assert make_cpu_manager().get_config_info() is None
+
+
 def test_cpu_manager_reports_allocation_size_histogram():
     manager = make_cpu_manager(num_blocks=4, cache_policy="lru")
 

tests/v1/kv_offload/test_factory.py

@@ -588,3 +588,65 @@ def test_build_metric_definitions_returns_counter_at_threshold():
     metrics = spec_cls.build_metric_definitions(extra_config)
 
     assert CPUOffloadingMetrics.STORES_SKIPPED in metrics
+
+
+def test_cpu_config_info_wiring_matches_declared_labels():
+    """The config info the spec hands the manager must line up exactly with the
+    declared CPU_CONFIG_INFO labels.
+
+    Guards two things record_config_info() relies on but the manager-level test
+    cannot see: the keys equal the label set (a drift would KeyError at emit
+    time) and the values reflect the spec's computed static config.
+    """
+    from vllm.v1.kv_offload.cpu.common import (
+        CPU_CONFIG_INFO_LABELS,
+        CPUOffloadingMetrics,
+    )
+
+    config = _make_offloading_config()
+    spec = OffloadingSpecFactory.create_spec(config)
+    config_info = spec.get_manager().get_config_info()
+
+    assert config_info is not None
+    labels = config_info[CPUOffloadingMetrics.CPU_CONFIG_INFO]
+    assert set(labels) == set(CPU_CONFIG_INFO_LABELS)
+    assert labels == {
+        "num_blocks": str(spec.num_blocks),
+        "blocks_per_chunk": str(spec.blocks_per_chunk),
+        "kv_bytes_per_chunk": str(spec.kv_bytes_per_chunk),
+        "cpu_page_size_per_worker": str(spec.cpu_page_size_per_worker),
+        "eviction_policy": spec.eviction_policy,
+    }
+
+    metrics = type(spec).build_metric_definitions(config.extra_config)
+    assert (
+        tuple(metrics[CPUOffloadingMetrics.CPU_CONFIG_INFO].labelnames)
+        == CPU_CONFIG_INFO_LABELS
+    )
+
+
+def test_cpu_config_info_reflects_world_size_sizing():
+    """The emitted labels carry the world-size-folded sizing (and the mmap
+    padding), not the single-rank numbers -- same geometry as
+    test_cpu_spec_sizes_normalized_worker_layout."""
+    from vllm.v1.kv_offload.cpu.common import CPUOffloadingMetrics
+
+    alignment = SharedOffloadRegion.BLOCK_SIZE_ALIGNMENT
+    spec = _create_spec(
+        cpu_bytes_to_use=alignment * 3,
+        worker_kv_bytes_per_block=16,
+        blocks_per_chunk=2,
+        world_size=6,
+        tp_size=3,
+        pp_size=2,
+    )
+
+    assert spec.get_manager().get_config_info() == {
+        CPUOffloadingMetrics.CPU_CONFIG_INFO: {
+            "num_blocks": "3",
+            "blocks_per_chunk": "2",
+            "kv_bytes_per_chunk": str(alignment),
+            "cpu_page_size_per_worker": "32",
+            "eviction_policy": "lru",
+        }
+    }

vllm/config/kv_transfer.py

@@ -71,6 +71,13 @@ class KVTransferConfig:
     'recompute': reschedule the request to recompute failed blocks
     'fail': immediately fail the request with an error finish reason (default)"""
 
+    kv_connector_config_info: dict[str, dict[str, str]] | None = field(
+        default=None, init=False
+    )
+    """Runtime-populated (not a CLI arg): static per-connector config, mapping
+    metric_name -> {label: value}, propagated from the engine-core handshake so
+    it can be emitted as Info-style metrics at startup."""
+
     def compute_hash(self) -> str:
         """
         WARNING: Whenever a new field is added to this config,

vllm/distributed/kv_transfer/kv_connector/v1/base.py

@@ -418,6 +418,16 @@ def get_kv_connector_stats(self) -> "KVConnectorStats | None":
         """
         return None
 
+    def get_config_info(self) -> dict[str, dict[str, str]] | None:
+        """
+        Static, per-connector configuration to emit once at engine startup as
+        Info-style gauges, mapping ``metric_name -> {label: value}``. Mirrors
+        ``vllm:cache_config_info`` so a connector's static config is observable
+        on ``/metrics`` before any request. Returns None when there is nothing
+        to expose.
+        """
+        return None
+
     def get_kv_connector_kv_cache_events(self) -> "KVConnectorKVEvents | None":
         """
         Get the KV connector kv cache events collected during the last interval.

vllm/distributed/kv_transfer/kv_connector/v1/metrics.py

@@ -135,6 +135,15 @@ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
         """
         raise NotImplementedError
 
+    def record_config_info(
+        self, config_info: dict[str, dict[str, str]], engine_idx: int = 0
+    ):
+        """
+        Emit the connector's static config as Info-style gauges once at engine
+        startup, mapping ``metric_name -> {label: value}`` (each set to 1).
+        Default no-op; connectors with startup config override this.
+        """
+
 
 class KVConnectorProm:
     """
@@ -173,3 +182,10 @@ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
         if self.prom_metrics is None:
             return
         self.prom_metrics.observe(transfer_stats_data, engine_idx)
+
+    def record_config_info(
+        self, config_info: dict[str, dict[str, str]], engine_idx: int = 0
+    ):
+        if self.prom_metrics is None:
+            return
+        self.prom_metrics.record_config_info(config_info, engine_idx)

vllm/distributed/kv_transfer/kv_connector/v1/multi_connector.py

@@ -124,6 +124,15 @@ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
             )
             self._prom_metrics[connector_id].observe(stats_data["data"], engine_idx)
 
+    def record_config_info(
+        self, config_info: dict[str, dict[str, str]], engine_idx: int = 0
+    ):
+        # config_info is flat (metric_name -> labels), merged across children;
+        # hand it to every child prom, each of which emits only the metrics it
+        # owns (unknown metric names are skipped).
+        for prom in self._prom_metrics.values():
+            prom.record_config_info(config_info, engine_idx)
+
 
 class MultiConnector(KVConnectorBase_V1, SupportsHMA):
     """
@@ -637,6 +646,19 @@ def get_kv_connector_stats(self) -> MultiKVConnectorStats | None:
                 stats_by_connector[connector_id] = stats
         return stats_by_connector
 
+    def get_config_info(self) -> dict[str, dict[str, str]] | None:
+        # Merge nested connectors' startup config-info (metric_name -> labels)
+        # so a connector wrapped inside MultiConnector still emits at startup.
+        merged: dict[str, dict[str, str]] | None = None
+        for c in self._connectors:
+            info = c.get_config_info()
+            if info is None:
+                continue
+            if merged is None:
+                merged = {}
+            merged.update(info)
+        return merged
+
     @classmethod
     def build_prom_metrics(
         cls,

vllm/distributed/kv_transfer/kv_connector/v1/offloading/metrics.py

@@ -395,6 +395,9 @@ def _create_metric(
             metric_cls = self._counter_cls
         elif isinstance(metadata, OffloadingGaugeMetadata):
             metric_cls = self._gauge_cls
+            # Counters and histograms are per-process cumulative and must stay
+            # summed; only gauges collapse (see OffloadingGaugeMetadata).
+            kwargs["multiprocess_mode"] = metadata.multiprocess_mode
         elif isinstance(metadata, OffloadingHistogramMetadata):
             metric_cls = self._histogram_cls
             if metadata.buckets is not None:
@@ -502,3 +505,15 @@ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
                     raise AssertionError(
                         f"Unknown metric type '{type_str}' for key: {key}"
                     )
+
+    def record_config_info(
+        self, config_info: dict[str, dict[str, str]], engine_idx: int = 0
+    ):
+        """Emit each config-info metric once as a gauge pinned to 1, with the
+        static config carried in the labels (cache_config_info-style)."""
+        for metric_name, labels in config_info.items():
+            metadata = self._offloading_metric_metadata.get(metric_name)
+            if metadata is None:
+                continue
+            labelvalues = tuple(labels[name] for name in metadata.labelnames)
+            self._set_gauge(metric_name, 1, labelvalues, engine_idx)

vllm/distributed/kv_transfer/kv_connector/v1/offloading_connector.py

@@ -206,6 +206,11 @@ def get_kv_connector_stats(self) -> KVConnectorStats | None:
             return self.connector_scheduler.get_stats()
         return None
 
+    def get_config_info(self) -> dict[str, dict[str, str]] | None:
+        if self.connector_scheduler is not None:
+            return self.connector_scheduler.manager.get_config_info()
+        return None
+
     @classmethod
     def build_kv_connector_stats(
         cls, data: dict[str, Any] | None = None

vllm/v1/engine/__init__.py

@@ -92,6 +92,10 @@ class EngineCoreReadyResponse:
     kv_cache_size_tokens: int | None = None
     kv_cache_max_concurrency: float | None = None
     kv_events_config: KVEventsConfig | None = None
+    # Static per-connector config to emit as Info-style metrics at startup,
+    # mapping metric_name -> {label: value}. None when no KV connector or the
+    # connector exposes no startup config.
+    kv_connector_config_info: dict[str, dict[str, str]] | None = None
 
 
 class EngineCoreRequest(

vllm/v1/engine/core.py

@@ -1619,6 +1619,7 @@ def _send_engine_dead(self):
     def _make_ready_response(self) -> EngineCoreReadyResponse:
         parallel_config = self.vllm_config.parallel_config
         scheduler_config = self.vllm_config.scheduler_config
+        connector = self.scheduler.get_kv_connector()
         return EngineCoreReadyResponse(
             max_model_len=self.vllm_config.model_config.max_model_len,
             num_gpu_blocks=self.vllm_config.cache_config.num_gpu_blocks or 0,
@@ -1640,6 +1641,9 @@ def _make_ready_response(self) -> EngineCoreReadyResponse:
             max_num_batched_tokens=scheduler_config.max_num_batched_tokens,
             instance_id=self.vllm_config.instance_id,
             kv_events_config=self.scheduler.get_kv_event_publisher_config(),
+            kv_connector_config_info=(
+                connector.get_config_info() if connector is not None else None
+            ),
         )
 
     def process_input_sockets(

vllm/v1/engine/core_client.py

@@ -767,6 +767,17 @@ def _apply_ready_response(self, payload: bytes) -> None:
             else response.kv_cache_max_concurrency
         )
 
+        # Static per-connector config (metric_name -> {label: value}), emitted
+        # as Info-style metrics at startup. Per-engine (last writer wins), not
+        # summed across DP; identical across replicas.
+        if (
+            response.kv_connector_config_info is not None
+            and vllm_config.kv_transfer_config is not None
+        ):
+            vllm_config.kv_transfer_config.kv_connector_config_info = (
+                response.kv_connector_config_info
+            )
+
         # In external DP LB mode, the coordinator address that the
         # front-end procs connect to is obtained by each engine via it's
         # initial handshake with the rank 0 front-end.