[Metrics][KV Offload] Add vllm:kv_offload_config_info - generic info metrics for offloading the offloading connector (#56867)

Signed-off-by: Amir Friedman <Amir.Friedman1@ibm.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Or Ozeri <oro@il.ibm.com>
This commit is contained in:
Amir Friedman
2026-10-14 12:34:37 +00:00
committed by GitHub
co-authored by Claude Opus 5 Or Ozeri
parent 3a5a6b11a3
commit 775fefa804
17 changed files with 926 additions and 42 deletions
+17
View File
@@ -112,6 +112,23 @@ vllm serve <model> \
| `self_describing_kv_events` | no | `false` | both | Opt-in. When `true` *and* KV cache events are enabled (`--kv-events-config` with `enable_kv_cache_events`), the connector emits self-describing block-granular `BlockStored`/`BlockRemoved` payloads (constituent block hashes, whole-chunk `token_ids`, per-block `block_size`, parent hash, LoRA + group/cache-spec metadata) instead of the placeholder fallback, so external KV-event consumers can index offloaded blocks. Inert unless events are enabled. With `TieringOffloadingSpec`, a CPU promotion is self-describing when a local request observes its primary-tier `HIT` before event translation; otherwise its stored event may retain the placeholder, while a later `HIT` can backfill metadata for removal. Pending-removal/re-promotion races and externally initiated promotions may also produce placeholders, and consumers must ignore removals for unknown hashes. Partial recurrent tails emit the hash-aligned portion from the physical block start through the tail boundary. Other sliding-window/SSM chunks keep the placeholder fallback. In chunk mode (`block_size` > GPU block size, or `blocks_per_chunk` > 1), overlapping chunks re-announce shared per-block hashes, so consumers must reference-count (deduplicate) repeated store/remove announcements. |
| `spec_module_path` | no | — | both | Python import path for a custom `OffloadingSpec` not in the built-in registry. Required only when `spec_name` is not built-in (advanced). |
## Info Metric Labels
The connector publishes one Prometheus gauge, `vllm:kv_offload_config_info`, and always sets it to 1. The labels hold static facts of the offloading configuration, so a user reads the resolved numbers instead of the launch flags. The gauge appears from the first scheduler step, so an idle engine exposes no series. Each engine reports its own configuration, and not the instance total.
The gauge holds one series for each configuration that the manager reports. A manager that holds one configuration gives one series. A manager that holds several adds a label of its own, which tells the series apart.
| Label | Declared by | Meaning | Notes |
| --- | --- | --- | --- |
| `tier` | `TieringOffloadingSpec` | The tier that this series reports. | `<index>:<type>`. Index `0` is the CPU primary tier, and a secondary tier index starts at 1. `<type>` is the `type` key of the tier config, such as `1:fs`. |
| `cpu_num_chunks` | `CPUOffloadingSpec` or `TieringOffloadingSpec` | Chunk slots in the CPU tier. | Chunks, not GPU blocks. `cpu_bytes_to_use` divided by the bytes of one chunk. |
The spec declares the label names in the API server process, and the manager fills the values in the engine process. The names of every series bind once, at the declaration. A label that one series owns reads empty on every other series.
`CPUOffloadingSpec` publishes `cpu_num_chunks` on its one series. `TieringOffloadingSpec` publishes it on the `0:primary` series. For a worked example of a secondary tier, the `example` tier publishes one label, `example_info`.
A manager adds a label with `config_info_keys()` of its spec and `config_info()` of its own class. A name that a manager fills and the spec does not declare is dropped, and the engine log then holds one warning line.
## Custom Eviction Policies
`eviction_policy` resolves through `CachePolicyFactory` (`vllm/v1/kv_offload/cpu/policies/factory.py`), which pre-registers the built-in `lru` and `arc` policies.
@@ -2,12 +2,14 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from types import SimpleNamespace
from typing import Any
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
from prometheus_client import Counter, Gauge, Histogram
from vllm.distributed.kv_transfer.kv_connector.v1.metrics import KVConnectorLogging
from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
KV_OFFLOAD_CONFIG_INFO,
OffloadingConnectorStats,
OffloadPromMetrics,
_ConnectorMetricName,
@@ -115,15 +117,21 @@ class _FakeVllmConfig:
def _spec_cls_with_metric_definitions(
metric_definitions: dict[str, Any],
info_keys: tuple[str, ...] = (),
) -> type:
"""Build a fake offloading spec class reporting the given metric
definitions, so tests don't need to patch the real CPU spec."""
definitions and info label names, so tests don't need to patch the real
CPU spec."""
class _FakeOffloadingSpec:
@staticmethod
def build_metric_definitions(extra_config):
return metric_definitions
@staticmethod
def config_info_keys(extra_config):
return info_keys
return _FakeOffloadingSpec
@@ -784,3 +792,333 @@ def test_prom_metrics_rejects_undeclared_metric():
_StatsKey.DATA: {"unknown:metric": {(): 1}},
}
)
def _prom_metrics(
metric_definitions: dict[str, Any] | None = None,
info_keys: tuple[str, ...] = (),
):
"""Build OffloadPromMetrics over fake metric classes, with one engine."""
spec_cls = _spec_cls_with_metric_definitions(metric_definitions or {}, info_keys)
with patch.object(OffloadingSpecFactory, "get_spec_cls", return_value=spec_cls):
return 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"], 1: ["model", "1"]},
)
def test_prom_metrics_keeps_a_gauge_declared_multiprocess_mode():
"""A gauge that declares its own multiprocess_mode keeps it.
The value on OffloadingGaugeMetadata must stay a default. A hardcoded
"mostrecent" in _create_metric passes the info-gauge test below and
silently overrides a gauge that needs another merge, for example a real
cross-frontend sum.
"""
prom_metrics = _prom_metrics(
{
PENDING_STORES: OffloadingGaugeMetadata(
documentation="gauge declaring a non-default multiprocess mode.",
multiprocess_mode="sum",
),
}
)
gauge_def = prom_metrics._offloading_metric_defs[PENDING_STORES]
assert gauge_def.kwargs["multiprocess_mode"] == "sum"
def test_prom_metrics_omits_multiprocess_mode_outside_a_gauge():
"""multiprocess_mode reaches a gauge only.
prometheus_client accepts the argument on Gauge alone, so hoisting the
kwarg out of the gauge branch breaks construction of a real Counter or
Histogram.
"""
prom_metrics = _prom_metrics(
{
MY_COUNTER: OffloadingCounterMetadata(documentation="a counter."),
LOOKUP_LATENCY: OffloadingHistogramMetadata(
documentation="a histogram.",
buckets=(0.1, 1),
),
}
)
counter_def = prom_metrics._offloading_metric_defs[MY_COUNTER]
histogram_def = prom_metrics._offloading_metric_defs[LOOKUP_LATENCY]
assert "multiprocess_mode" not in counter_def.kwargs
assert "multiprocess_mode" not in histogram_def.kwargs
def test_prom_metrics_declares_the_info_gauge_from_the_spec_keys():
"""The API-server process holds no manager, but it does hold the spec, so
it declares the gauge at startup, before any payload."""
prom_metrics = _prom_metrics(info_keys=("cpu_num_chunks", "root_dir_name"))
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert gauge_def.kwargs["name"] == KV_OFFLOAD_CONFIG_INFO
assert gauge_def.kwargs["labelnames"] == [
"model_name",
"engine",
"cpu_num_chunks",
"root_dir_name",
]
# The frontend declares the metric and binds no label values, so an engine
# that never reports exposes no series.
assert gauge_def.children == []
prom_metrics.observe(
{_StatsKey.INFO: [{"cpu_num_chunks": 512, "root_dir_name": "a"}]}
)
# Every value renders as a label value, and the metric itself is pinned
# to 1.
(gauge,) = gauge_def.children
assert gauge.labelvalues == ("model", "0", "512", "a")
assert gauge.set_values == [1]
def test_prom_metrics_reads_the_info_payload_by_label_name():
"""A manager orders config_info() as it likes, and msgpack keeps that
order, so the frontend must key on the name rather than on the position."""
prom_metrics = _prom_metrics(info_keys=("first", "second"))
prom_metrics.observe({_StatsKey.INFO: [{"second": 2, "first": 1}]})
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
(gauge,) = gauge_def.children
assert gauge.labelvalues == ("model", "0", "1", "2")
def test_prom_metrics_reports_one_info_series_for_each_mapping():
"""A manager reports one mapping for each series, and a label of its own
tells the series apart. Two mappings of one kind agree on every other value,
so without that label they collapse into one series."""
prom_metrics = _prom_metrics(info_keys=("source", "root_dir_name"))
prom_metrics.observe(
{
_StatsKey.INFO: [
{"source": "1", "root_dir_name": "a"},
{"source": "2", "root_dir_name": "b"},
]
}
)
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert [child.labelvalues for child in gauge_def.children] == [
("model", "0", "1", "a"),
("model", "0", "2", "b"),
]
assert [child.set_values for child in gauge_def.children] == [[1], [1]]
def test_prom_metrics_declares_the_info_gauge_as_most_recent():
"""The info gauge must merge across frontends by freshest write.
Offloading stats reach one frontend per step as complete per-engine
snapshots, so summing would report the number of participating API-server
processes instead of the 1 an info gauge is pinned to -- wrong only under
multiprocess deployment, and silently, since the facts ride the labels.
"""
prom_metrics = _prom_metrics(info_keys=("cpu_num_chunks",))
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert gauge_def.kwargs["multiprocess_mode"] == "mostrecent"
def test_prom_metrics_reports_the_info_gauge_without_manager_labels():
"""A spec that declares no fact still yields the metric, so its presence
alone answers whether offloading runs."""
prom_metrics = _prom_metrics()
prom_metrics.observe({_StatsKey.INFO: [{}]})
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert gauge_def.kwargs["labelnames"] == ["model_name", "engine"]
(gauge,) = gauge_def.children
assert gauge.labelvalues == ("model", "0")
def test_prom_metrics_empties_a_declared_info_label_no_manager_fills():
"""A name the spec declares and the manager skips must publish empty.
PromQL reads an empty label value as an absent label, so the series still
matches both shapes. This code runs on the engine output path of the API
server, where an exception ends the output handler, so it warns and goes
on rather than drop the whole series.
"""
prom_metrics = _prom_metrics(info_keys=("filled", "skipped"))
with patch(
"vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics.logger"
) as mock_logger:
prom_metrics.observe({_StatsKey.INFO: [{"filled": 512}]})
assert mock_logger.warning_once.call_count == 1
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
(gauge,) = gauge_def.children
assert gauge.labelvalues == ("model", "0", "512", "")
def test_prom_metrics_keeps_quiet_when_a_name_belongs_to_another_series():
"""A name that one series owns reads empty on the other series by design.
The declaration holds the names of every series, so the frontend checks the
payload as a whole. A per-series check would read this normal shape as a
disagreement, and warn at every start of a server that reports several
series.
"""
prom_metrics = _prom_metrics(info_keys=("source", "root_dir_name", "bucket"))
with patch(
"vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics.logger"
) as mock_logger:
prom_metrics.observe(
{
_StatsKey.INFO: [
{"source": "1", "root_dir_name": "kv_a"},
{"source": "2", "bucket": "kv_bucket"},
]
}
)
mock_logger.warning_once.assert_not_called()
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert [child.labelvalues for child in gauge_def.children] == [
("model", "0", "1", "kv_a", ""),
("model", "0", "2", "", "kv_bucket"),
]
def test_prom_metrics_drops_an_info_label_the_spec_did_not_declare():
"""A Prometheus metric binds its label names once, so a name that the spec
did not declare cannot join the gauge. It warns and publishes the rest."""
prom_metrics = _prom_metrics(info_keys=("declared",))
with patch(
"vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics.logger"
) as mock_logger:
prom_metrics.observe({_StatsKey.INFO: [{"declared": 512, "undeclared": 1}]})
assert mock_logger.warning_once.call_count == 1
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert gauge_def.kwargs["labelnames"] == ["model_name", "engine", "declared"]
(gauge,) = gauge_def.children
assert gauge.labelvalues == ("model", "0", "512")
def test_prom_metrics_reports_the_info_of_every_engine():
"""Every engine of one instance runs one offloading config, so they share
the gauge and differ by the engine label alone."""
prom_metrics = _prom_metrics(info_keys=("cpu_num_chunks",))
prom_metrics.observe({_StatsKey.INFO: [{"cpu_num_chunks": 512}]}, engine_idx=0)
prom_metrics.observe({_StatsKey.INFO: [{"cpu_num_chunks": 512}]}, engine_idx=1)
gauge_def = prom_metrics._offloading_metric_defs[KV_OFFLOAD_CONFIG_INFO]
assert [child.labelvalues for child in gauge_def.children] == [
("model", "0", "512"),
("model", "1", "512"),
]
def test_stats_hold_the_config_info_without_an_observation():
"""An info-only payload must survive the scheduler gate, which drops an
empty payload, and must add no line to the interval log."""
stats = OffloadingConnectorStats()
assert stats.is_empty()
stats.set_info([{"cpu_num_chunks": 512}])
assert not stats.is_empty()
assert stats.reduce() == {}
def test_aggregate_keeps_the_config_info_of_the_other_payload():
"""The scheduler payload holds the info and merges into the worker payload
(v1/core/sched/scheduler.py), so a dropped info would hide the metric on
every step that also transfers, which is every step that matters."""
worker_stats = OffloadingConnectorStats()
worker_stats.increase_counter(LOAD_BYTES, 100)
scheduler_stats = OffloadingConnectorStats()
scheduler_stats.set_info([{"cpu_num_chunks": 512}])
worker_stats.aggregate(scheduler_stats)
assert worker_stats.data[_StatsKey.INFO] == [{"cpu_num_chunks": 512}]
assert worker_stats.data[_StatsKey.DATA][LOAD_BYTES] == {(): 100}
def test_aggregate_keeps_the_config_info_of_a_later_payload():
"""The scheduler sends the info once, so a later payload must not clear
it."""
stats = OffloadingConnectorStats()
stats.set_info([{"cpu_num_chunks": 512}])
later_stats = OffloadingConnectorStats()
later_stats.increase_counter(LOAD_BYTES, 100)
stats.aggregate(later_stats)
assert stats.data[_StatsKey.INFO] == [{"cpu_num_chunks": 512}]
def test_metric_sections_leave_out_the_config_info():
"""A manager accesses its own metrics by a walk over the metric sections, as
KVCRSecondaryTierManager does in v1/kv_offload/tiering/kvcr/manager.py. The
info section holds label names, not metric names, so it stays out of that
walk."""
stats = OffloadingConnectorStats()
stats.increase_counter(MY_COUNTER, 2, ("remote_deliver",))
stats.set_info([{"info_label": "local"}])
info = stats.data[_StatsKey.INFO]
sections = stats.metric_sections()
assert not any(section is info for section in sections)
assert sections == (stats.data[_StatsKey.TYPES], stats.data[_StatsKey.DATA])
def test_logging_writes_no_line_for_an_info_only_payload():
"""An info-only payload reaches the logging path, and reduce() drops the info.
The info belongs to the Prometheus frontend only. Without the guard in
KVConnectorLogging.log(), such a payload writes "KV Transfer metrics: "
with an empty body.
"""
stats = OffloadingConnectorStats()
stats.set_info([{"info_label": "local"}])
assert not stats.is_empty()
assert stats.reduce() == {}
kv_logging = KVConnectorLogging(None)
kv_logging.transfer_stats_accumulator = stats
log_fn = MagicMock()
kv_logging.log(log_fn=log_fn)
log_fn.assert_not_called()
assert kv_logging.transfer_stats_accumulator is None
def test_scheduler_sends_the_config_info_once(request_runner):
"""MockOffloadingSpec publishes no fact, so the payload holds one empty
mapping, which still yields the metric."""
runner = request_runner(
block_size=4,
num_gpu_blocks=10,
async_scheduling=False,
)
stats = runner.connector_scheduler.get_stats()
assert stats is not None
assert stats.data[_StatsKey.INFO] == [{}]
assert runner.connector_scheduler.get_stats() is None
@@ -132,6 +132,7 @@ class MockOffloadingSpec(OffloadingSpec):
self.manager.lookup.return_value = LookupResult.MISS
self.manager.get_load_source.return_value = CacheHitSource.EXTERNAL_UNSPECIFIED
self.manager.get_stats.return_value = None
self.manager.config_info.return_value = [{}]
self.manager.on_new_request.return_value = RequestOffloadingContext()
self.handler = MockOffloadingWorker()
+10
View File
@@ -24,6 +24,7 @@ from vllm.v1.kv_offload.cpu.common import (
from vllm.v1.kv_offload.cpu.manager import CPUOffloadingManager
from vllm.v1.kv_offload.cpu.policies.arc import ARCCachePolicy
from vllm.v1.kv_offload.cpu.policies.lru import LRUCachePolicy
from vllm.v1.kv_offload.cpu.spec import CPUOffloadingSpec
def make_req_context(
@@ -238,6 +239,15 @@ def test_filter_reused_manager_reports_stores_skipped_counter():
assert stats.reduce()[CPUOffloadingMetrics.STORES_SKIPPED] == 0
def test_cpu_manager_fills_the_config_info_keys_of_the_spec():
"""A name the declaration misses becomes a dropped label at runtime."""
manager = make_cpu_manager(num_chunks=4)
(info,) = manager.config_info()
assert tuple(info) == CPUOffloadingSpec.config_info_keys({})
assert info["cpu_num_chunks"] == 4
def test_cpu_manager_reports_cache_usage_gauge():
def check_usage_stats(manager: CPUOffloadingManager, value: float):
stats = manager.get_stats()
@@ -11,7 +11,7 @@ These tests verify:
"""
from collections.abc import Iterable
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
import numpy as np
import pytest
@@ -99,7 +99,8 @@ def has_histogram_count(reduced: dict[str, int | float], metric_name: str) -> bo
class MetricsSecondaryTierManager(SecondaryTierManager):
"""Test-only secondary tier that declares and emits one labeled metric."""
"""Test-only secondary tier that declares and emits one labeled metric,
and publishes one config fact."""
MY_TIER_METRIC = "my_tier_metric"
@@ -112,6 +113,13 @@ class MetricsSecondaryTierManager(SecondaryTierManager):
)
}
@classmethod
def config_info_keys(cls, extra_config):
return ("path",)
def config_info(self):
return {"path": f"/mnt/{self.tier_type}"}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.stats: OffloadingConnectorStats | None = None
@@ -248,6 +256,19 @@ class TestExampleSecondaryTierManager:
# Third chunk not present
assert tier.lookup(chunks[2], _CTX) is LookupResult.MISS
def test_config_info_keys_match_the_filled_values(self):
"""A name the declaration misses becomes a dropped label at runtime."""
mock_view = memoryview(torch.zeros((10, 16), dtype=torch.int8).numpy())
tier = ExampleSecondaryTierManager(
offloading_spec=_MOCK_OFFLOADING_SPEC,
primary_kv_view=mock_view,
tier_type="example",
custom_param=67,
)
info = tier.config_info()
assert tuple(info) == ExampleSecondaryTierManager.config_info_keys({})
assert info["example_info"] == 67
# What a request-level cascade does with a key already present in the primary
# tier, keyed by what primary lookup says about it. SUPPLY reaches the peer now,
@@ -1545,3 +1566,175 @@ def test_parse_tier_filter_skips_bad_entries():
if __name__ == "__main__":
pytest.main([__file__, "-v"])
def test_tiering_manager_reports_one_config_info_series_for_each_tier():
"""The info metric holds one series for each tier, and the tier label tells
the series apart. A tier cannot know its own index, so the manager adds the
label. A tier fills the names it owns only, so the CPU primary tier fills
its cpu_ labels on its own series alone."""
mock_region = _mock_mmap_region(5)
primary_tier = CPUPrimaryTierOffloadingManager(
num_chunks=5, mmap_region=mock_region
)
secondary_tiers = [
MetricsSecondaryTierManager(
offloading_spec=_MOCK_OFFLOADING_SPEC,
primary_kv_view=mock_region.create_kv_memoryview(),
tier_type="test_metrics",
)
for _ in range(2)
]
manager = TieringOffloadingManager(
primary_tier=primary_tier,
secondary_tiers=secondary_tiers,
)
assert manager.config_info() == [
{"tier": "0:primary", "cpu_num_chunks": 5},
{"tier": "1:test_metrics", "path": "/mnt/test_metrics"},
{"tier": "2:test_metrics", "path": "/mnt/test_metrics"},
]
def test_tiering_manager_reports_the_primary_series_with_no_secondary_tier():
"""A tiering manager runs with no secondary tier, and the info metric then
holds the primary series alone. The series still carries the tier label, so
the frontend publishes one series and not a bare metric."""
mock_region = _mock_mmap_region(5)
manager = TieringOffloadingManager(
primary_tier=CPUPrimaryTierOffloadingManager(
num_chunks=5, mmap_region=mock_region
),
secondary_tiers=[],
)
assert manager.config_info() == [{"tier": "0:primary", "cpu_num_chunks": 5}]
def test_tiering_manager_rejects_a_primary_tier_with_two_info_mappings():
"""This manager holds one primary tier, so two mappings would give two
series with one tier label. If the two mappings agree on every declared
name, the frontend cache then collapses them into one series and warns
about nothing. The unpack fails at engine start instead."""
class _TwoMappingPrimary(CPUPrimaryTierOffloadingManager):
def config_info(self):
return [{"path": "/a"}, {"path": "/b"}]
mock_region = _mock_mmap_region(5)
manager = TieringOffloadingManager(
primary_tier=_TwoMappingPrimary(num_chunks=5, mmap_region=mock_region),
secondary_tiers=[],
)
with pytest.raises(ValueError):
manager.config_info()
def test_tiering_manager_keeps_the_tier_label_out_of_the_tier_mapping():
"""The manager adds the tier label to a copy. A tier that returns a stored
mapping must not gain a foreign key that outlives the call."""
class _StoredInfoTier(MetricsSecondaryTierManager):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.info: dict[str, str] = {"path": "/mnt/a"}
def config_info(self):
return self.info
mock_region = _mock_mmap_region(5)
tier = _StoredInfoTier(
offloading_spec=_MOCK_OFFLOADING_SPEC,
primary_kv_view=mock_region.create_kv_memoryview(),
tier_type="test_metrics",
)
manager = TieringOffloadingManager(
primary_tier=CPUPrimaryTierOffloadingManager(
num_chunks=5, mmap_region=mock_region
),
secondary_tiers=[tier],
)
manager.config_info()
assert tier.info == {"path": "/mnt/a"}
def test_tiering_spec_declares_the_config_info_keys_the_manager_fills():
"""The spec declares the info label names in the API-server process, and
the manager fills the values in the engine process. A drift between the two
empties every declared label and drops every filled value, so this pins
both sides to the same literal names. The frontend reads a value by name,
so the order of the names does not have to agree."""
tier_configs = [{"type": "test_metrics"}, {"type": "test_metrics"}]
mock_region = _mock_mmap_region(5)
manager = TieringOffloadingManager(
primary_tier=CPUPrimaryTierOffloadingManager(
num_chunks=5, mmap_region=mock_region
),
secondary_tiers=[
MetricsSecondaryTierManager(
offloading_spec=_MOCK_OFFLOADING_SPEC,
primary_kv_view=mock_region.create_kv_memoryview(),
tier_type=tier_config["type"],
)
for tier_config in tier_configs
],
)
with patch.object(
SecondaryTierFactory,
"get_tier_class",
return_value=MetricsSecondaryTierManager,
):
keys = TieringOffloadingSpec.config_info_keys({"secondary_tiers": tier_configs})
assert keys == ("cpu_num_chunks", "tier", "path")
filled: set[str] = set()
for info in manager.config_info():
assert set(info) <= set(keys)
filled |= set(info)
assert filled == set(keys)
def test_tiering_spec_passes_each_tier_its_own_config():
"""A tier reads its own parameters out of its own config dict, not out of
the instance-wide kv_connector_extra_config. A tier that names a label
after one of its own keys declares nothing from the wrong dict."""
seen: list[dict] = []
class _RecordingTier(MetricsSecondaryTierManager):
@classmethod
def config_info_keys(cls, extra_config):
seen.append(extra_config)
return ()
tier_configs = [{"type": "test_metrics", "root_dir": "/mnt/a"}]
with patch.object(
SecondaryTierFactory, "get_tier_class", return_value=_RecordingTier
):
TieringOffloadingSpec.config_info_keys({"secondary_tiers": tier_configs})
assert seen == tier_configs
def test_tiering_spec_rejects_a_tier_that_declares_the_reserved_tier_label():
"""TieringOffloadingManager.config_info() writes the tier label over the
value of a tier, so a tier that declares the name loses its own value with
no message. The declaration must fail at start-up instead."""
class _ReservedLabelTier(MetricsSecondaryTierManager):
@classmethod
def config_info_keys(cls, extra_config):
return ("tier",)
tier_configs = [{"type": "test_metrics"}]
with (
patch.object(
SecondaryTierFactory, "get_tier_class", return_value=_ReservedLabelTier
),
pytest.raises(AssertionError, match="reserved label 'tier'"),
):
TieringOffloadingSpec.config_info_keys({"secondary_tiers": tier_configs})
@@ -99,8 +99,13 @@ class KVConnectorLogging:
# Produce a single cumulative stats object for the last time
# interval from the recorded observations.
xfer_metrics = self.transfer_stats_accumulator.reduce()
xfer_metrics_str = ", ".join(f"{k}={v}" for k, v in xfer_metrics.items())
log_fn("KV Transfer metrics: %s", xfer_metrics_str)
# A payload can hold static config facts alone (info metric), which reduce()
# drops, because they belong to the Prometheus metrics only.
if xfer_metrics:
xfer_metrics_str = ", ".join(
f"{k}={v}" for k, v in xfer_metrics.items()
)
log_fn("KV Transfer metrics: %s", xfer_metrics_str)
# Reset metrics for next interval
self.reset()
@@ -1,5 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
@@ -10,14 +11,30 @@ from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
PromMetric,
PromMetricT,
)
from vllm.logger import init_logger
from vllm.v1.kv_offload.base import (
ConfigInfoMapping,
OffloadingCounterMetadata,
OffloadingGaugeMetadata,
OffloadingHistogramMetadata,
OffloadingMetricMetadata,
OffloadingSpec,
)
from vllm.v1.kv_offload.factory import OffloadingSpecFactory
logger = init_logger(__name__)
KV_OFFLOAD_CONFIG_INFO = "vllm:kv_offload_config_info"
_INFO_METRIC_HELP = (
"Static configuration of the KV offload managers of this engine instance. "
"The configured offloading spec declares the label names, and each manager "
"fills the values, so a series appears from the first scheduler step of its "
"engine. A manager reports one series for each configuration it holds, and "
"a label of its own tells the series apart. Each engine reports its own "
"configuration, not the instance total."
)
class _TransferMetricName:
"""Flat metric names for GPU↔offload-medium transfer operations."""
@@ -166,6 +183,9 @@ class _StatsKey:
TYPES = "types"
# Maps metric name -> {label values tuple -> observed value (number or list)}
DATA = "data"
# One mapping of info metric label name -> label value for each series.
# None until a manager reports it, and one empty mapping for no fact.
INFO = "info"
@dataclass
@@ -177,6 +197,7 @@ class OffloadingConnectorStats(KVConnectorStats):
{
_StatsKey.TYPES: {name: _MetricType.*, ...},
_StatsKey.DATA: {name: {labelvalues: value, ...}, ...},
_StatsKey.INFO: [{labelname: labelvalue, ...}, ...] | None,
}
This structure is self-describing: it survives IPC serialization
@@ -187,6 +208,12 @@ class OffloadingConnectorStats(KVConnectorStats):
use the latest snapshot per-label-tuple, and histogram values are lists of
observed samples per-label-tuple. Unlabeled metrics use ``()`` as their
labelvalues tuple.
``INFO`` holds the static config facts of one series each, which the
scheduler sends once per process. It stays out of ``DATA``, because its
label names are known only when the payload arrives. It holds label names,
not metric names, so a manager that accesses its own metrics walks
``metric_sections()``.
"""
def __post_init__(self):
@@ -197,6 +224,7 @@ class OffloadingConnectorStats(KVConnectorStats):
self.data: dict[str, Any] = {
_StatsKey.TYPES: {},
_StatsKey.DATA: {},
_StatsKey.INFO: None,
}
@property
@@ -207,10 +235,28 @@ class OffloadingConnectorStats(KVConnectorStats):
def _values(self) -> dict[str, Any]:
return self.data[_StatsKey.DATA]
def metric_sections(self) -> tuple[dict[str, Any], ...]:
"""Return the type section and the value section, in that order.
A manager that accesses its own metrics walks these sections. Each one
is a mutable reference.
"""
return (self._types, self._values)
def aggregate(self, other: "KVConnectorStats") -> "KVConnectorStats":
if other.is_empty():
return self
assert isinstance(other, OffloadingConnectorStats)
# The scheduler payload merges into the worker payload
# (v1/core/sched/scheduler.py), so the info must survive the merge in
# this direction to reach the Prometheus frontend. reduce() drops it, so
# it never reaches a log line. The scheduler sends it once, so a later
# payload with no info must not clear it.
other_info = other.data.get(_StatsKey.INFO)
if other_info is not None:
self.data[_StatsKey.INFO] = [
dict(series_info) for series_info in other_info
]
other_types = other._types
other_values = other._values
for key, other_label_values in other_values.items():
@@ -268,7 +314,11 @@ class OffloadingConnectorStats(KVConnectorStats):
return return_dict
def is_empty(self) -> bool:
return not self.data.get(_StatsKey.DATA)
# An info-only payload is not empty. It carries no observation, but the
# caller drops an empty payload, and the info must reach the frontend.
return (
not self.data.get(_StatsKey.DATA) and self.data.get(_StatsKey.INFO) is None
)
def increase_counter(
self,
@@ -294,6 +344,18 @@ class OffloadingConnectorStats(KVConnectorStats):
gauge_values = self._values.setdefault(gauge_name, {})
gauge_values[labelvalues] = gauge_value
def set_info(self, info: Sequence[ConfigInfoMapping]) -> None:
"""Put the static config facts of this engine on the stats payload.
Args:
info: One mapping of info metric label name to label value for each
series, as OffloadingManager.config_info() returns it. An empty
mapping still gives one series, with an empty value on every
label the spec declared.
"""
self.data[_StatsKey.INFO] = [dict(series_info) for series_info in info]
def observe_histogram(
self,
histogram_name: str,
@@ -322,14 +384,22 @@ class OffloadPromMetrics(KVConnectorPromMetrics):
kv_transfer_config = vllm_config.kv_transfer_config
assert kv_transfer_config is not None
extra_config = kv_transfer_config.kv_connector_extra_config
spec_cls = OffloadingSpecFactory.get_spec_cls(extra_config)
self._spec_cls: type[OffloadingSpec] = OffloadingSpecFactory.get_spec_cls(
extra_config
)
# The spec declares the info label names here; a manager fills the
# values later, and _observe_info() aligns the two.
self._info_keys: tuple[str, ...] = self._spec_cls.config_info_keys(extra_config)
self._offloading_metric_metadata: dict[str, OffloadingMetricMetadata] = {
**spec_cls.build_metric_definitions(extra_config),
**self._spec_cls.build_metric_definitions(extra_config),
**get_connector_metric_definitions(),
KV_OFFLOAD_CONFIG_INFO: OffloadingGaugeMetadata(
documentation=_INFO_METRIC_HELP, labelnames=self._info_keys
),
}
from vllm.v1.kv_offload.cpu.spec import CPUOffloadingSpec
self._observe_deprecated_metrics = issubclass(spec_cls, CPUOffloadingSpec)
self._observe_deprecated_metrics = issubclass(self._spec_cls, CPUOffloadingSpec)
self._offloading_metric_defs: dict[str, PromMetricT] = {}
# (engine_idx, metric_name, labelvalues) -> metric with bound labels
self.offloading_metrics: dict[
@@ -393,6 +463,7 @@ class OffloadPromMetrics(KVConnectorPromMetrics):
metric_cls = self._counter_cls
elif isinstance(metadata, OffloadingGaugeMetadata):
metric_cls = self._gauge_cls
kwargs["multiprocess_mode"] = metadata.multiprocess_mode
elif isinstance(metadata, OffloadingHistogramMetadata):
metric_cls = self._histogram_cls
if metadata.buckets is not None:
@@ -476,8 +547,52 @@ class OffloadPromMetrics(KVConnectorPromMetrics):
observation
)
def _observe_info(self, info: Sequence[dict[str, Any]], engine_idx: int) -> None:
"""Publish the static config facts of one engine, one series per mapping.
The spec declares the label names, so a payload only fills them. A
declared name that no series fills reads empty on every series. A payload
name that the spec did not declare is dropped. Either gap logs once.
The check covers the payload as a whole, and not one series, because the
declaration holds the names of every series. A name that one series owns
reads empty on the other series by design. A series that leaves a name of
its own unfilled therefore stays hidden here.
Args:
info: One mapping of label name to label value for each series, as
OffloadingManager.config_info() returns it.
engine_idx: Index of the reporting engine.
"""
filled: dict[str, None] = {}
for series_info in info:
filled.update(dict.fromkeys(series_info))
labelvalues = tuple(
str(series_info.get(key, "")) for key in self._info_keys
)
self._set_gauge(KV_OFFLOAD_CONFIG_INFO, 1, labelvalues, engine_idx)
never_filled = tuple(key for key in self._info_keys if key not in filled)
dropped = tuple(key for key in filled if key not in self._info_keys)
if never_filled or dropped:
logger.warning_once(
"%s: spec %s and the manager of engine %d disagree on the KV "
"offload config labels. Declared, and no series filled them, so "
"empty on every series: %s. Filled, and the spec did not "
"declare them, so dropped: %s.",
KV_OFFLOAD_CONFIG_INFO,
self._spec_cls.__name__,
engine_idx,
never_filled,
dropped,
)
def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
"""Observe transfer statistics."""
info = transfer_stats_data.get(_StatsKey.INFO)
if info is not None:
self._observe_info(info, engine_idx)
metric_types = transfer_stats_data.get(_StatsKey.TYPES, {})
metric_data = transfer_stats_data.get(_StatsKey.DATA, {})
for key, label_value_map in metric_data.items():
@@ -574,6 +574,10 @@ class OffloadingConnectorScheduler:
)
self.manager: OffloadingManager = spec.get_manager()
self._connector_stats = OffloadingConnectorStats()
# The static config facts ride on the first stats payload of this
# process. A Prometheus child holds the values it received, so one
# payload is enough.
self._info_sent = False
full_attention_groups: list[int] = []
sliding_window_groups: list[int] = []
@@ -2004,6 +2008,14 @@ class OffloadingConnectorScheduler:
else:
stats.aggregate(manager_stats)
if not self._info_sent:
if stats is None:
stats = OffloadingConnectorStats()
# Sent even when the manager reports no facts, so that the metric
# exists whenever offloading runs.
stats.set_info(self.manager.config_info())
self._info_sent = True
return stats
def request_finished(
+57 -3
View File
@@ -3,8 +3,8 @@
"""Core abstractions for KV cache offloading in vLLM v1."""
from abc import ABC, abstractmethod
from collections.abc import Collection, Iterable, Sequence
from dataclasses import dataclass, field
from collections.abc import Collection, Iterable, Mapping, Sequence
from dataclasses import dataclass, field, fields
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, ClassVar, NamedTuple, NewType, TypeVar
@@ -214,7 +214,12 @@ class OffloadingCounterMetadata(OffloadingMetricMetadata):
@dataclass(frozen=True)
class OffloadingGaugeMetadata(OffloadingMetricMetadata):
pass
# Gauge-only in prometheus_client: how MultiProcessCollector merges samples
# written by different API-server processes. Offloading stats reach one
# frontend per step as complete per-engine snapshots, so the freshest write
# is the correct value and summing would multiply it by the number of
# participating frontends.
multiprocess_mode: str = "mostrecent"
@dataclass(frozen=True)
@@ -232,6 +237,36 @@ class OffloadingKVEventsConfig:
self_describing_kv_events: bool
# The labels of one info metric series: a label name mapped to its value.
ConfigInfoMapping = Mapping[str, str | int | float | bool]
@dataclass(frozen=True)
class ConfigInfo:
"""Static facts that one component publishes on vllm:kv_offload_config_info.
Each field is one label. The OffloadingSpec declares the names with
config_info_keys(), and the OffloadingManager fills the values with
as_config_info(). A declared name that the manager does not fill reads
empty, and an undeclared name is dropped. The values must stay fixed for
the process lifetime. Document each field in
docs/features/kv_offloading_usage.md. See CPUOffloadingInfo for an example.
"""
@classmethod
def config_info_keys(cls) -> tuple[str, ...]:
"""Return one label name for each field, in field order."""
return tuple(info.name for info in fields(cls))
def as_config_info(self) -> ConfigInfoMapping:
"""Return the label values. A None value becomes "None", not empty."""
values = (getattr(self, info.name) for info in fields(self))
return {
key: "None" if value is None else value
for key, value in zip(self.config_info_keys(), values)
}
class OffloadingManager(ABC):
@abstractmethod
def lookup(self, key: OffloadKey, req_context: ReqContext) -> LookupResult:
@@ -416,6 +451,13 @@ class OffloadingManager(ABC):
"""Return collected metrics since last call, or None if disabled."""
return None
def config_info(self) -> Sequence[ConfigInfoMapping]:
"""Return the info metric labels, one mapping for each series.
The scheduler reads this once. See ConfigInfo.
"""
return [{}]
def shutdown(self) -> None:
"""Shutdown the manager and release any resources."""
return
@@ -602,6 +644,18 @@ class OffloadingSpec(ABC):
"""Return Prometheus metric definitions emitted by this spec."""
return {}
@classmethod
def config_info_keys(cls, extra_config: dict[str, Any]) -> tuple[str, ...]:
"""Return the info metric label names that the manager fills.
See ConfigInfo.
Args:
extra_config: kv_connector_extra_config of this instance.
"""
return ()
def __init__(self, config: OffloadingConfig):
self.config = config
self.extra_config = config.extra_config
+11 -1
View File
@@ -1,8 +1,10 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from dataclasses import dataclass
import numpy as np
from vllm.v1.kv_offload.base import BlockIDsLoadStoreSpec
from vllm.v1.kv_offload.base import BlockIDsLoadStoreSpec, ConfigInfo
class CPUOffloadingMetrics:
@@ -24,3 +26,11 @@ class CPULoadStoreSpec(BlockIDsLoadStoreSpec):
@property
def chunk_ids(self) -> np.ndarray:
return self.block_ids
@dataclass(frozen=True)
class CPUOffloadingInfo(ConfigInfo):
"""Static facts about the CPU offload tier."""
# Chunk slots in the tier. Chunks, not GPU blocks.
cpu_num_chunks: int
+11 -1
View File
@@ -1,7 +1,7 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections import OrderedDict
from collections.abc import Collection, Iterable
from collections.abc import Collection, Iterable, Sequence
from dataclasses import dataclass, field
from typing_extensions import override
@@ -10,6 +10,7 @@ from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
OffloadingConnectorStats,
)
from vllm.v1.kv_offload.base import (
ConfigInfoMapping,
LoadStoreSpec,
LookupResult,
Medium,
@@ -23,6 +24,7 @@ from vllm.v1.kv_offload.base import (
)
from vllm.v1.kv_offload.cpu.common import (
CPULoadStoreSpec,
CPUOffloadingInfo,
CPUOffloadingMetrics,
)
from vllm.v1.kv_offload.cpu.policies.base import CachePolicy, ChunkStatus
@@ -66,6 +68,9 @@ class CPUOffloadingManager(OffloadingManager):
):
self.medium: Medium = Medium.CPU
self._num_chunks: int = num_chunks
self._config_info: ConfigInfoMapping = CPUOffloadingInfo(
cpu_num_chunks=num_chunks
).as_config_info()
self._num_allocated_chunks: int = 0
self._free_list: list[int] = []
self.events: list[OffloadingEvent] | None = [] if enable_events else None
@@ -446,6 +451,11 @@ class CPUOffloadingManager(OffloadingManager):
yield from self.events
self.events.clear()
@override
def config_info(self) -> Sequence[ConfigInfoMapping]:
"""Report the CPU cache facts of the one cache this manager holds."""
return [self._config_info]
def get_stats(self) -> OffloadingConnectorStats | None:
stats = OffloadingConnectorStats()
+7 -1
View File
@@ -17,7 +17,7 @@ from vllm.v1.kv_offload.base import (
OffloadingWorker,
)
from vllm.v1.kv_offload.config import OffloadingConfig
from vllm.v1.kv_offload.cpu.common import CPUOffloadingMetrics
from vllm.v1.kv_offload.cpu.common import CPUOffloadingInfo, CPUOffloadingMetrics
from vllm.v1.kv_offload.cpu.gpu_worker import CPUOffloadingWorker
from vllm.v1.kv_offload.cpu.manager import CPUOffloadingManager
from vllm.v1.kv_offload.cpu.shared_offload_region import SharedOffloadRegion
@@ -136,6 +136,12 @@ class CPUOffloadingSpec(OffloadingSpec):
"cache_policy_module_path"
)
@classmethod
@override
def config_info_keys(cls, extra_config: dict[str, Any]) -> tuple[str, ...]:
"""Return the label names CPUOffloadingManager.config_info() fills."""
return CPUOffloadingInfo.config_info_keys()
@override
def get_manager(self) -> OffloadingManager:
if not self._manager:
+23
View File
@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, ClassVar
import numpy as np
from vllm.v1.kv_offload.base import (
ConfigInfoMapping,
Locality,
LookupResult,
Medium,
@@ -36,6 +37,8 @@ if TYPE_CHECKING:
# Type alias for job IDs used in async transfer tracking
JobId = int
TIER_LABEL = "tier"
class TieringOffloadingMetrics:
"""Metric names for TieringOffloadingManager."""
@@ -370,3 +373,23 @@ class SecondaryTierManager(ABC):
def get_stats(self) -> "OffloadingConnectorStats | None":
"""Return and reset metric observations collected by this tier."""
return None
@classmethod
def config_info_keys(cls, tier_config: dict[str, Any]) -> tuple[str, ...]:
"""Return the info metric label names of this tier.
TieringOffloadingSpec merges the names of all tiers and adds
TIER_LABEL, so a tier must not declare it. See ConfigInfo.
Args:
tier_config: Configuration dict of this tier.
"""
return ()
def config_info(self) -> ConfigInfoMapping:
"""Return the info metric labels of this tier, for its one series.
TieringOffloadingManager adds TIER_LABEL. See ConfigInfo.
"""
return {}
+26 -1
View File
@@ -10,11 +10,14 @@ TieringOffloadingManager without requiring actual storage or network backends.
import logging
from collections.abc import Iterable
from typing import TYPE_CHECKING, ClassVar
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar
from typing_extensions import override
from vllm.v1.kv_offload.base import (
ConfigInfo,
ConfigInfoMapping,
LookupResult,
Medium,
OffloadKey,
@@ -34,6 +37,17 @@ if TYPE_CHECKING:
from vllm.v1.kv_offload.tiering.backpressure import BackpressureDetector
@dataclass(frozen=True)
class ExampleTierInfo(ConfigInfo):
"""Config facts of one ExampleSecondaryTierManager.
A field name is the label name. A secondary tier needs no name of its own
for the tier, because the tier label tells the series of the tiers apart.
"""
example_info: int
class ExampleSecondaryTierManager(SecondaryTierManager):
"""A simple in-memory secondary tier.
@@ -69,6 +83,8 @@ class ExampleSecondaryTierManager(SecondaryTierManager):
backpressure_detector=backpressure_detector,
)
self._config_info = ExampleTierInfo(example_info=custom_param)
logger.info(
"ExampleSecondaryTierManager initialized with custom_param=%d", custom_param
)
@@ -81,6 +97,15 @@ class ExampleSecondaryTierManager(SecondaryTierManager):
assert primary_kv_view.strides is not None
self._chunk_size = primary_kv_view.strides[0]
@classmethod
@override
def config_info_keys(cls, tier_config: dict[str, Any]) -> tuple[str, ...]:
return ExampleTierInfo.config_info_keys()
@override
def config_info(self) -> ConfigInfoMapping:
return self._config_info.as_config_info()
@override
def lookup(self, key: OffloadKey, req_context: ReqContext) -> LookupResult:
"""Check whether a chunk exists in this secondary tier.
+1 -1
View File
@@ -523,7 +523,7 @@ class KVCRSecondaryTierManager(SecondaryTierManager):
stats = cast(OffloadingConnectorStats | None, self._kvcr.get_stats())
if stats is not None:
# Add the framework namespace only at the vLLM boundary.
for metrics in stats.data.values():
for metrics in stats.metric_sections():
for name in tuple(metrics):
metrics[_vllm_metric_name(name)] = metrics.pop(name)
return stats
+22
View File
@@ -32,6 +32,7 @@ from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
)
from vllm.logger import init_logger
from vllm.v1.kv_offload.base import (
ConfigInfoMapping,
LoadStoreSpec,
LookupResult,
OffloadingEvent,
@@ -47,6 +48,7 @@ from vllm.v1.kv_offload.cpu.common import CPULoadStoreSpec
from vllm.v1.kv_offload.cpu.manager import CPUOffloadingManager
from vllm.v1.kv_offload.cpu.shared_offload_region import SharedOffloadRegion
from vllm.v1.kv_offload.tiering.base import (
TIER_LABEL,
JobId,
JobResult,
ParentManager,
@@ -1007,6 +1009,26 @@ class TieringOffloadingManager(OffloadingManager):
return stats
@override
def config_info(self) -> Sequence[ConfigInfoMapping]:
"""Return one mapping for each tier, primary tier first, with TIER_LABEL.
Raises:
ValueError: If the primary tier returns more than one mapping.
"""
(primary_info,) = self.primary_tier.config_info()
primary_tier_info = dict(primary_info)
primary_tier_info[TIER_LABEL] = self._metrics.primary_tier_label[0]
tier_infos = [primary_tier_info]
for tier in self.secondary_tiers:
tier_info = dict(tier.config_info())
tier_info[TIER_LABEL] = self._metrics.tier_label(self._tier_index[tier])[0]
tier_infos.append(tier_info)
return tier_infos
@override
def shutdown(self) -> None:
"""Shut down secondary tiers before releasing primary resources.
+67 -24
View File
@@ -69,7 +69,11 @@ from vllm.v1.kv_offload.config import OffloadingConfig
from vllm.v1.kv_offload.cpu.gpu_worker import CPUOffloadingWorker
from vllm.v1.kv_offload.cpu.shared_offload_region import SharedOffloadRegion
from vllm.v1.kv_offload.cpu.spec import CPUOffloadingSpec
from vllm.v1.kv_offload.tiering.base import TieringOffloadingMetrics
from vllm.v1.kv_offload.tiering.base import (
TIER_LABEL,
SecondaryTierManager,
TieringOffloadingMetrics,
)
from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory
from vllm.v1.kv_offload.tiering.manager import (
CPUPrimaryTierOffloadingManager,
@@ -105,7 +109,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Histogram of blocking time spent in a per-chunk tier lookup "
"that resolved as a hit or miss, labeled by tier, in seconds."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
buckets=(
0.00001,
0.00005,
@@ -128,7 +132,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"first returning retry until that same tier lookup resolves "
"as a hit or miss, labeled by tier, in seconds."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
buckets=(
0.0001,
0.0005,
@@ -149,35 +153,35 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Total bytes read from secondary tiers into the primary tier, "
"labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.READ_TIME] = OffloadingCounterMetadata(
documentation=(
"Total time spent reading from secondary tiers into the primary "
"tier, in seconds, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.WRITE_BYTES] = OffloadingCounterMetadata(
documentation=(
"Total bytes written from the primary tier to secondary tiers, "
"labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.WRITE_TIME] = OffloadingCounterMetadata(
documentation=(
"Total time spent writing from the primary tier to secondary "
"tiers, in seconds, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.PROMOTION_JOB_FAILURES] = (
OffloadingCounterMetadata(
documentation=(
"Number of failed secondary-tier promotion jobs, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.CASCADE_JOB_FAILURES] = (
@@ -185,18 +189,18 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
documentation=(
"Number of failed secondary-tier cascade jobs, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.CHUNK_QUERIES] = OffloadingCounterMetadata(
documentation=(
"Number of chunk lookup queries sent to a tier, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.CHUNK_HITS] = OffloadingCounterMetadata(
documentation="Number of chunk lookup hits in a tier, labeled by tier.",
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.PROMOTION_ALLOCATION_FAILURES] = (
OffloadingCounterMetadata(
@@ -212,7 +216,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Current fraction of primary-tier space used by writes from "
"secondary tiers, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.PRIMARY_READ_USAGE_PERC] = (
@@ -221,7 +225,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Current fraction of primary-tier space used by reads to "
"secondary tiers, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.ACTIVE_PROMOTION_JOBS] = (
@@ -229,22 +233,17 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
documentation=(
"Number of active secondary-tier promotion jobs, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.ACTIVE_CASCADE_JOBS] = OffloadingGaugeMetadata(
documentation=(
"Number of active secondary-tier cascade jobs, labeled by tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
secondary_tier_configs = extra_config.get("secondary_tiers", [])
if not isinstance(secondary_tier_configs, list):
raise ValueError("secondary_tiers must be a list of tier configurations")
for tier_config in secondary_tier_configs:
assert isinstance(tier_config, dict)
tier_cls = SecondaryTierFactory.get_tier_class(tier_config)
for tier_config, tier_cls in cls._get_secondary_tiers(extra_config):
metrics.update(tier_cls.build_metric_definitions(tier_config))
metrics[TieringOffloadingMetrics.BACKPRESSURE_STORE_LATENCY_EMA] = (
@@ -253,7 +252,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Exponential moving average of store latency "
"for back-pressure detection, in s/MiB."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.BACKPRESSURE_STORES_DROPPED] = (
@@ -262,7 +261,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
"Number of store operations dropped due to "
"back-pressure on a secondary tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
metrics[TieringOffloadingMetrics.BACKPRESSURE_BLOCKS_DROPPED] = (
@@ -270,7 +269,7 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
documentation=(
"Number of blocks dropped due to back-pressure on a secondary tier."
),
labelnames=("tier",),
labelnames=(TIER_LABEL,),
)
)
@@ -426,6 +425,50 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
return self._manager
@classmethod
@override
def config_info_keys(cls, extra_config: dict[str, Any]) -> tuple[str, ...]:
"""Return TIER_LABEL and the label names of all tiers, with no repeat."""
# dict.fromkeys() keeps one copy of a name that two tier types share.
keys = dict.fromkeys(super().config_info_keys(extra_config))
keys[TIER_LABEL] = None
for tier_config, tier_cls in cls._get_secondary_tiers(extra_config):
tier_keys = tier_cls.config_info_keys(tier_config)
assert TIER_LABEL not in tier_keys, (
f"{tier_cls.__name__} must not declare the reserved label "
f"{TIER_LABEL!r}"
)
keys.update(dict.fromkeys(tier_keys))
return tuple(keys)
@classmethod
def _get_secondary_tiers(
cls, extra_config: dict[str, Any]
) -> list[tuple[dict[str, Any], type[SecondaryTierManager]]]:
"""Resolve every configured secondary tier to its manager class.
Args:
extra_config: kv_connector_extra_config of this instance.
Returns:
One (tier config, tier class) pair for each secondary tier, in
configuration order. That order sets the tier index.
Raises:
ValueError: If secondary_tiers is not a list.
"""
secondary_tier_configs = extra_config.get("secondary_tiers", [])
if not isinstance(secondary_tier_configs, list):
raise ValueError("secondary_tiers must be a list of tier configurations")
tiers = []
for tier_config in secondary_tier_configs:
assert isinstance(tier_config, dict)
tier_cls = SecondaryTierFactory.get_tier_class(tier_config)
tiers.append((tier_config, tier_cls))
return tiers
@override
def _uses_shared_region(self) -> bool:
# Tiering always allocates on the shared region (every platform), so the