iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Core] Add PROMOTION_LATENCY histogram metric for tiering offload (#53910)

Browse Source 
Signed-off-by: Wen Zhou <wenzhou@redhat.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: OpenAI Codex <noreply@openai.com>
Co-authored-by: Kevin H. Luu <khluu000@gmail.com>
Co-authored-by: Or Ozeri <oro@il.ibm.com>
This commit is contained in:
Wen Zhou
2026-10-15 13:47:41 +03:00
committed by GitHub
co-authored by Claude Opus 4.6 OpenAI Codex Kevin H. Luu Or Ozeri
parent 6fe7401fac
commit e10c10a644
6 changed files with 234 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/v1/kv_offload/tiering/p2p/test_manager.py
+68 -1
View File
@@ -24,7 +24,12 @@ from vllm.v1.kv_offload.base import (
ReqContext,
ScheduleEndContext,
)
from vllm.v1.kv_offload.tiering.base import JobResult, TransferJob
from vllm.v1.kv_offload.tiering.base import (
JobResult,
TieringOffloadingMetrics,
TransferJob,
)
from vllm.v1.kv_offload.tiering.manager import TieringOffloadingManager
from vllm.v1.kv_offload.tiering.p2p import manager as manager_module
from vllm.v1.kv_offload.tiering.p2p.manager import (
_UNBOUND_STORE_TIMEOUT_S,
@@ -112,6 +117,8 @@ def _job_metadata(
def _make_manager() -> P2PSecondaryTierManager:
"""Create a manager with stubbed __init__."""
mgr = P2PSecondaryTierManager.__new__(P2PSecondaryTierManager)
mgr.tier_type = "p2p"
mgr._bp_detector = None
mgr._local_id = "127.0.0.1:7777"
mgr._hash_seed = "0"
mgr._finished_jobs = []
@@ -525,6 +532,66 @@ class TestSubmitLoad:
assert mgr._finished_jobs == [JobResult(job_id=1, success=False)]
# ---------------------------------------------------------------------------
# Tests for promotion latency metrics
# ---------------------------------------------------------------------------
class TestPromotionLatency:
def test_p2p_load_without_transfer_time_records_promotion_latency(self):
"""P2P load results omit transfer_time; the manager records promotion
latency from job creation to completion."""
p2p_tier = _make_manager()
primary_kv_view = memoryview(np.zeros((5, 16), dtype=np.uint8))
primary_tier = SimpleNamespace(
_num_chunks=5,
get_kv_memoryview=lambda: primary_kv_view,
complete_write=lambda *args: None,
get_stats=lambda: None,
)
tiering_manager = TieringOffloadingManager(
primary_tier=primary_tier,
secondary_tiers=[p2p_tier],
tier_poll_interval_s=0.0,
)
job = TransferJob(
job_id=42,
keys=[b"key1"],
chunk_ids=np.array([0]),
is_promotion=True,
req_context=_req_context(_remote_prefiller_kv_params()),
)
tiering_manager._register_job(job, tier_idx=0)
session = _FakeSession(
peer_id="10.0.0.1:8000",
loads=[LoadResult(job_id=42, kv_request_id="req-1", success=True)],
)
p2p_tier._sessions[session.peer_id] = session
p2p_tier._control = SimpleNamespace(poll=lambda: [])
reported_results: list[JobResult] = []
get_finished_jobs = p2p_tier.get_finished_jobs
def capture_finished_jobs():
results = list(get_finished_jobs())
reported_results.extend(results)
return results
p2p_tier.get_finished_jobs = capture_finished_jobs
p2p_tier.submit_load(job)
assert session.requests == [(42, "req-1")]
tiering_manager._process_finished_jobs()
assert reported_results == [JobResult(job_id=42, success=True)]
assert reported_results[0].transfer_time is None
stats = tiering_manager.get_stats()
assert stats is not None
observations = stats.data["data"][TieringOffloadingMetrics.PROMOTION_LATENCY]
assert len(observations[("1:p2p", "success")]) == 1
# ---------------------------------------------------------------------------
# Tests for on_request_finished
# ---------------------------------------------------------------------------
tests/v1/kv_offload/tiering/test_metrics.py
+91
View File
@@ -2,11 +2,13 @@
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
from collections.abc import Iterable
from types import SimpleNamespace
import numpy as np
import pytest
from vllm.v1.kv_offload.base import LookupResult, ReqContext, make_offload_key
from vllm.v1.kv_offload.tiering import metrics as tiering_metrics
from vllm.v1.kv_offload.tiering.base import (
JobResult,
TieringOffloadingMetrics,
@@ -211,6 +213,95 @@ def test_tiering_metrics_tracker_reports_active_job_and_primary_usage_gauges():
assert values[TieringOffloadingMetrics.ACTIVE_PROMOTION_JOBS][p2p_label] == 1
def test_tiering_metrics_tracker_records_promotion_latency_histogram(monkeypatch):
"""Tracker measures job-creation-to-completion, not transfer_time.
Includes failed promotions.
"""
clock = 100.0
monkeypatch.setattr(
tiering_metrics, "time", SimpleNamespace(monotonic=lambda: clock)
)
tracker = TieringMetricsTracker(
tier_types=["fs", "p2p"],
num_primary_chunks=5,
primary_chunk_size=16,
)
promotion_job = JobMetadata(
TransferJob(0, to_keys([0, 1]), np.array([0, 1]), True, _CTX),
1,
)
promotion_job.transfer_job.submit_time = clock
tracker.on_job_registered(promotion_job)
clock += 1.5
tracker.on_job_finished(
promotion_job, JobResult(job_id=0, success=True, transfer_time=0.42)
)
stats = tracker.take_stats()
assert stats is not None
observations = stats.data["data"][TieringOffloadingMetrics.PROMOTION_LATENCY]
assert observations == {("2:p2p", "success"): [1.5]}
cascade_job = JobMetadata(
TransferJob(1, to_keys([2]), np.array([2]), False, _CTX),
1,
)
tracker.on_job_registered(cascade_job)
clock += 0.1
tracker.on_job_finished(
cascade_job, JobResult(job_id=1, success=True, transfer_time=0.10)
)
stats = tracker.take_stats()
assert stats is not None
values = stats.data["data"]
assert TieringOffloadingMetrics.PROMOTION_LATENCY not in values
failed_promotion_job = JobMetadata(
TransferJob(2, to_keys([3]), np.array([3]), True, _CTX),
0,
)
failed_promotion_job.transfer_job.submit_time = clock
tracker.on_job_registered(failed_promotion_job)
clock += 0.2
tracker.on_job_finished(
failed_promotion_job,
JobResult(job_id=2, success=False, transfer_time=0.42),
)
stats = tracker.take_stats()
assert stats is not None
values = stats.data["data"]
observations = values[TieringOffloadingMetrics.PROMOTION_LATENCY]
assert observations == {("1:fs", "failure"): pytest.approx([0.2])}
tracker.assert_idle()
def test_tiering_metrics_tracker_records_promotion_latency_without_transfer_time():
"""A completion carrying no transfer_time still records a latency sample."""
tracker = TieringMetricsTracker(
tier_types=["p2p"],
num_primary_chunks=5,
primary_chunk_size=16,
)
promotion_job = JobMetadata(
TransferJob(0, to_keys([0]), np.array([0]), True, _CTX),
0,
)
tracker.on_job_registered(promotion_job)
tracker.on_job_finished(promotion_job, JobResult(job_id=0, success=True))
tracker.assert_idle()
stats = tracker.take_stats()
assert stats is not None
observations = stats.data["data"][TieringOffloadingMetrics.PROMOTION_LATENCY]
assert len(observations[("1:p2p", "success")]) == 1
def test_tiering_metrics_tracker_records_promotion_allocation_failures():
tracker = TieringMetricsTracker(
tier_types=["fs"],
tests/v1/kv_offload/tiering/test_tiering_offloading.py
+21
View File
@@ -535,6 +535,14 @@ class TestTieringOffloadingManager:
for chunk in chunks:
self.secondary_tier1.chunks[chunk] = True
original_submit_load = self.secondary_tier1.submit_load
def submit_load(job_metadata):
assert job_metadata.job_id in self.manager._jobs
original_submit_load(job_metadata)
self.secondary_tier1.submit_load = submit_load
# Lookup each chunk to initiate promotion for all of them
for chunk in chunks:
result = self.manager.lookup(chunk, _CTX)
@@ -546,6 +554,19 @@ class TestTieringOffloadingManager:
# End of step 2: processes the completed promotion jobs
self._simulate_on_schedule_end()
stats = self.manager.get_stats()
assert stats is not None
reduced = stats.reduce()
assert (
reduced[
histogram_count_key(
TieringOffloadingMetrics.PROMOTION_LATENCY,
("1:example", "success"),
)
]
== 1
)
# Now chunks should be in primary tier
assert count_hits(self.primary_tier, chunks) == 3
vllm/v1/kv_offload/tiering/base.py
+1
View File
@@ -65,6 +65,7 @@ class TieringOffloadingMetrics:
)
BACKPRESSURE_STORES_DROPPED = "vllm:kv_offload_tiering_backpressure_stores_dropped"
BACKPRESSURE_BLOCKS_DROPPED = "vllm:kv_offload_tiering_backpressure_blocks_dropped"
PROMOTION_LATENCY = "vllm:kv_offload_tiering_promotion_latency_seconds"
@dataclass
vllm/v1/kv_offload/tiering/metrics.py
+20
View File
@@ -124,6 +124,8 @@ class TieringMetricsTracker:
self, job_metadata: _JobMetadataLike, result: JobResult
) -> None:
self._observe_finished_job_stats(job_metadata, result)
if job_metadata.transfer_job.is_promotion:
self._observe_promotion_latency(job_metadata, result)
self._decrement_tier_state(job_metadata)
def on_promotion_allocation_failure(self) -> None:
@@ -237,6 +239,24 @@ class TieringMetricsTracker:
time_metric, completed_job.transfer_time, labelvalues
)
def _observe_promotion_latency(
self, job_metadata: _JobMetadataLike, result: JobResult
) -> None:
"""Record promotion time from job creation to reported completion.
This is broader than the tier-reported `JobResult.transfer_time`, which
covers only the active data movement, and so includes tier queueing.
Failed promotions are observed too; PROMOTION_JOB_FAILURES counts them.
"""
self._stats.observe_histogram(
TieringOffloadingMetrics.PROMOTION_LATENCY,
time.monotonic() - job_metadata.transfer_job.submit_time,
(
*self.tier_label(job_metadata.tier_idx),
"success" if result.success else "failure",
),
)
def _observe_active_transfer_stats(self, stats: OffloadingConnectorStats) -> None:
for tier_idx, state in enumerate(self._tier_states):
labelvalues = self.tier_label(tier_idx)
vllm/v1/kv_offload/tiering/spec.py
+33 -2
View File
@@ -159,7 +159,10 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
metrics[TieringOffloadingMetrics.READ_TIME] = OffloadingCounterMetadata(
documentation=(
"Total time spent reading from secondary tiers into the primary "
"tier, in seconds, labeled by tier."
"tier, in seconds, labeled by tier. Tiers that do not report "
"per-job transfer time have no series here. Use "
"vllm:kv_offload_tiering_promotion_latency_seconds for "
"tier-independent promotion timing."
),
labelnames=(TIER_LABEL,),
)
@@ -173,7 +176,8 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
metrics[TieringOffloadingMetrics.WRITE_TIME] = OffloadingCounterMetadata(
documentation=(
"Total time spent writing from the primary tier to secondary "
"tiers, in seconds, labeled by tier."
"tiers, in seconds, labeled by tier. Tiers that do not report "
"per-job transfer time have no series here."
),
labelnames=(TIER_LABEL,),
)
@@ -243,6 +247,33 @@ class TieringOffloadingSpec(CPUOffloadingSpec):
),
labelnames=(TIER_LABEL,),
)
metrics[TieringOffloadingMetrics.PROMOTION_LATENCY] = (
OffloadingHistogramMetadata(
documentation=(
"Histogram of per-job latency for secondary-tier promotion "
"attempts, measured from job creation until the tiering "
"manager observes completion so that tier queueing is "
"included, labeled by tier and result, in seconds. Includes "
"failed attempts."
),
labelnames=(TIER_LABEL, "result"),
buckets=(
0.005,
0.01,
0.025,
0.05,
0.1,
0.25,
0.5,
1,
2.5,
5,
10,
30,
60,
),
)
)
for tier_config, tier_cls in cls._get_secondary_tiers(extra_config):
metrics.update(tier_cls.build_metric_definitions(tier_config))