diff --git a/tests/v1/kv_offload/tiering/p2p/test_manager.py b/tests/v1/kv_offload/tiering/p2p/test_manager.py index 2282171fdb..15a3d64e23 100644 --- a/tests/v1/kv_offload/tiering/p2p/test_manager.py +++ b/tests/v1/kv_offload/tiering/p2p/test_manager.py @@ -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 # --------------------------------------------------------------------------- diff --git a/tests/v1/kv_offload/tiering/test_metrics.py b/tests/v1/kv_offload/tiering/test_metrics.py index 1a155892f8..97b0af051b 100644 --- a/tests/v1/kv_offload/tiering/test_metrics.py +++ b/tests/v1/kv_offload/tiering/test_metrics.py @@ -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"], diff --git a/tests/v1/kv_offload/tiering/test_tiering_offloading.py b/tests/v1/kv_offload/tiering/test_tiering_offloading.py index b6ee011b24..e8b38032bd 100644 --- a/tests/v1/kv_offload/tiering/test_tiering_offloading.py +++ b/tests/v1/kv_offload/tiering/test_tiering_offloading.py @@ -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 diff --git a/vllm/v1/kv_offload/tiering/base.py b/vllm/v1/kv_offload/tiering/base.py index 0be006d659..3e9fccf4c7 100644 --- a/vllm/v1/kv_offload/tiering/base.py +++ b/vllm/v1/kv_offload/tiering/base.py @@ -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 diff --git a/vllm/v1/kv_offload/tiering/metrics.py b/vllm/v1/kv_offload/tiering/metrics.py index 240ea57dd7..555077ec43 100644 --- a/vllm/v1/kv_offload/tiering/metrics.py +++ b/vllm/v1/kv_offload/tiering/metrics.py @@ -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) diff --git a/vllm/v1/kv_offload/tiering/spec.py b/vllm/v1/kv_offload/tiering/spec.py index cd8e375777..4b25597470 100644 --- a/vllm/v1/kv_offload/tiering/spec.py +++ b/vllm/v1/kv_offload/tiering/spec.py @@ -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))