[EC Connector] Add Metrics Collection (#54960)
Signed-off-by: omerpaz95 <omerpaz95@gmail.com>
omerpaz95 committed
Sep 17, 2026 at 05:36 UTC
fa3622a4a1cd16294a7843895a3246bf29c452c4
12 files changed
+472
-3
tests/v1/core/test_scheduler.py
+82
@@ -19,6 +19,7 @@ from vllm.config import (
19
SpeculativeConfig,
20
VllmConfig,
21
)
22
+from vllm.distributed.ec_transfer.ec_connector.metrics import ECConnectorStats
23
from vllm.distributed.kv_transfer.kv_connector.v1.hisparse.connector import (
24
HiSparseConnector,
25
HiSparseConnectorScheduler,
@@ -5296,6 +5297,87 @@ def test_scheduler_kv_connector_stats():
5297
assert final_stats == expected_data
5298
5299
5300
+def test_scheduler_ec_connector_stats():
5301
+ """Test worker-side, scheduler-side, and combined EC connector stats."""
5302
+
5303
+ class GenericECConnectorStats(ECConnectorStats):
5304
+ def reset(self):
5305
+ self.data = {}
5306
+
5307
+ def aggregate(self, other: ECConnectorStats) -> ECConnectorStats:
5308
+ self.data.update(other.data)
5309
+ return self
5310
+
5311
+ def reduce(self) -> dict[str, int | float]:
5312
+ return {}
5313
+
5314
+ def is_empty(self) -> bool:
5315
+ return not self.data
5316
+
5317
+ test_cases = (
5318
+ ({"worker": 1}, None, {"worker": 1}),
5319
+ (None, {"scheduler": 2}, {"scheduler": 2}),
5320
+ ({"worker": 1}, {"scheduler": 2}, {"worker": 1, "scheduler": 2}),
5321
+ )
5322
+
5323
+ for worker_data, scheduler_data, expected_data in test_cases:
5324
+ scheduler = create_scheduler()
5325
+ worker_stats = (
5326
+ GenericECConnectorStats(data=worker_data) if worker_data else None
5327
+ )
5328
+ scheduler_stats = (
5329
+ GenericECConnectorStats(data=scheduler_data) if scheduler_data else None
5330
+ )
5331
+ scheduler.ec_connector = Mock()
5332
+ scheduler.ec_connector.take_unavailable_requests.return_value = set()
5333
+ scheduler.ec_connector.get_ec_connector_stats.return_value = (
5334
+ scheduler_stats if worker_stats is None else None
5335
+ )
5336
+
5337
+ def update_connector_output(
5338
+ ec_connector_output: ECConnectorOutput,
5339
+ scheduler=scheduler,
5340
+ scheduler_stats=scheduler_stats,
5341
+ ):
5342
+ scheduler.ec_connector.get_ec_connector_stats.return_value = scheduler_stats
5343
+
5344
+ scheduler.ec_connector.update_connector_output.side_effect = (
5345
+ update_connector_output
5346
+ )
5347
+
5348
+ model_output = ModelRunnerOutput(
5349
+ req_ids=["req_0"],
5350
+ req_id_to_index={"req_0": 0},
5351
+ sampled_token_ids=[[123]],
5352
+ logprobs=None,
5353
+ prompt_logprobs_dict={},
5354
+ pooler_output=[None],
5355
+ ec_connector_output=ECConnectorOutput(ec_connector_stats=worker_stats)
5356
+ if worker_stats
5357
+ else None,
5358
+ )
5359
+ scheduler_output = SchedulerOutput(
5360
+ scheduled_new_reqs=[],
5361
+ scheduled_cached_reqs=None,
5362
+ num_scheduled_tokens={"req_0": 1},
5363
+ total_num_scheduled_tokens=1,
5364
+ scheduled_spec_decode_tokens={},
5365
+ scheduled_encoder_inputs={},
5366
+ num_common_prefix_blocks=[0],
5367
+ finished_req_ids=set(),
5368
+ free_encoder_mm_hashes=[],
5369
+ )
5370
+
5371
+ engine_core_outputs = scheduler.update_from_output(
5372
+ scheduler_output, model_output
5373
+ )
5374
+
5375
+ final_stats = next(
5376
+ iter(engine_core_outputs.values())
5377
+ ).scheduler_stats.ec_connector_stats
5378
+ assert final_stats == expected_data
5379
+
5380
+
5381
def test_ec_connector_update_connector_output_called():
5382
"""Test that worker-side EC connector output is forwarded to the
5383
EC connector's update_connector_output hook."""
tests/v1/ec_connector/unit/test_ec_connector_metrics.py
new
+85
@@ -0,0 +1,85 @@
1
+# SPDX-License-Identifier: Apache-2.0
2
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+"""Unit tests for the connector-agnostic EC connector metrics plumbing.
4
+
5
+Covers only the base infra: no concrete connector type implements these
6
+hooks yet, so every assertion here is about the documented no-op defaults
7
+and the container contracts, mirroring the equivalent KV connector tests.
8
+"""
9
+
10
+from unittest.mock import Mock
11
+
12
+import pytest
13
+
14
+from vllm.config import VllmConfig
15
+from vllm.distributed.ec_transfer.ec_connector.base import ECConnectorRole
16
+from vllm.distributed.ec_transfer.ec_connector.example_connector import (
17
+ ECExampleConnector,
18
+)
19
+from vllm.distributed.ec_transfer.ec_connector.metrics import (
20
+ ECConnectorLogging,
21
+ ECConnectorProm,
22
+ ECConnectorStats,
23
+)
24
+from vllm.v1.outputs import ECConnectorOutput
25
+
26
+pytestmark = pytest.mark.cpu_test
27
+
28
+
29
+@pytest.fixture
30
+def example_connector(tmp_path):
31
+ config = Mock(spec=VllmConfig)
32
+ config.ec_transfer_config = Mock()
33
+ config.ec_transfer_config.get_from_extra_config = Mock(return_value=str(tmp_path))
34
+ config.ec_transfer_config.is_ec_producer = True
35
+ config.ec_transfer_config.is_ec_consumer = False
36
+ return ECExampleConnector(vllm_config=config, role=ECConnectorRole.SCHEDULER)
37
+
38
+
39
+def test_base_stats_methods_raise_not_implemented():
40
+ """The base container has no aggregation logic of its own."""
41
+ stats = ECConnectorStats()
42
+
43
+ with pytest.raises(NotImplementedError):
44
+ stats.reset()
45
+ with pytest.raises(NotImplementedError):
46
+ stats.aggregate(stats)
47
+ with pytest.raises(NotImplementedError):
48
+ stats.reduce()
49
+ with pytest.raises(NotImplementedError):
50
+ stats.is_empty()
51
+
52
+
53
+def test_connector_hooks_default_to_none(example_connector):
54
+ """No connector type overrides these yet, so every hook is a no-op."""
55
+ assert example_connector.get_ec_connector_stats() is None
56
+ assert ECExampleConnector.build_ec_connector_stats() is None
57
+ assert (
58
+ ECExampleConnector.build_prom_metrics(
59
+ vllm_config=Mock(spec=VllmConfig),
60
+ metric_types={},
61
+ labelnames=[],
62
+ per_engine_labelvalues={},
63
+ )
64
+ is None
65
+ )
66
+
67
+
68
+def test_ec_connector_output_is_empty_reflects_stats_field():
69
+ output = ECConnectorOutput()
70
+ assert output.is_empty()
71
+
72
+ output.ec_connector_stats = ECConnectorStats(data={"x": 1})
73
+ assert not output.is_empty()
74
+
75
+
76
+def test_logging_and_prom_are_no_ops_without_ec_transfer_config():
77
+ """A normal run with no EC connector configured must stay silent."""
78
+ logging = ECConnectorLogging(ec_transfer_config=None)
79
+ logging.log() # Should not raise, nothing to log.
80
+
81
+ vllm_config = Mock(spec=VllmConfig)
82
+ vllm_config.ec_transfer_config = None
83
+ prom = ECConnectorProm(vllm_config, labelnames=[], per_engine_labelvalues={})
84
+ assert prom.prom_metrics is None
85
+ prom.observe({}) # Should not raise.
tests/v1/ec_connector/unit/test_ec_output_aggregator.py
+19
-3
@@ -5,6 +5,7 @@
5
import pytest
6
7
from vllm.distributed.ec_transfer.ec_connector.base import ECConnectorWorkerMetadata
8
+from vllm.distributed.ec_transfer.ec_connector.metrics import ECConnectorStats
9
from vllm.distributed.ec_transfer.ec_connector.utils import ECOutputAggregator
10
from vllm.distributed.kv_transfer.kv_connector.utils import KVOutputAggregator
11
from vllm.v1.outputs import (
@@ -29,6 +30,16 @@ class FakeWorkerMeta(ECConnectorWorkerMetadata):
30
return FakeWorkerMeta(self.saves + other.saves)
31
32
33
+class FakeConnectorStats(ECConnectorStats):
34
+ """Records merge order, mirroring FakeWorkerMeta above."""
35
+
36
+ def __init__(self, saves: list[str]):
37
+ self.saves = saves
38
+
39
+ def aggregate(self, other: "FakeConnectorStats") -> "FakeConnectorStats":
40
+ return FakeConnectorStats(self.saves + other.saves)
41
+
42
+
43
def _worker_output(ec_output: ECConnectorOutput | None) -> ModelRunnerOutput:
44
return ModelRunnerOutput(
45
req_ids=[], req_id_to_index={}, ec_connector_output=ec_output
@@ -38,19 +49,23 @@ def _worker_output(ec_output: ECConnectorOutput | None) -> ModelRunnerOutput:
49
def test_aggregate_folds_every_rank_onto_output_rank():
50
"""EC work done on any rank reaches the scheduler via output_rank's output.
51
41
- The middle rank reports no worker metadata: it must neither seed nor clobber
42
- the accumulator.
52
+ The middle rank reports no worker metadata or stats: it must neither seed
53
+ nor clobber either accumulator.
54
"""
55
outputs = [
56
_worker_output(
57
ECConnectorOutput(
58
finished_sending={"mm0"},
59
+ ec_connector_stats=FakeConnectorStats(["mm0"]),
60
ec_connector_worker_meta=FakeWorkerMeta(["mm0"]),
61
)
62
),
63
_worker_output(ECConnectorOutput(finished_recving={"mm1"})),
64
_worker_output(
53
- ECConnectorOutput(ec_connector_worker_meta=FakeWorkerMeta(["mm2"]))
65
+ ECConnectorOutput(
66
+ ec_connector_stats=FakeConnectorStats(["mm2"]),
67
+ ec_connector_worker_meta=FakeWorkerMeta(["mm2"]),
68
+ )
69
),
70
]
71
@@ -59,6 +74,7 @@ def test_aggregate_folds_every_rank_onto_output_rank():
74
assert result is outputs[2]
75
assert result.ec_connector_output.finished_sending == {"mm0"}
76
assert result.ec_connector_output.finished_recving == {"mm1"}
77
+ assert result.ec_connector_output.ec_connector_stats.saves == ["mm0", "mm2"]
78
assert result.ec_connector_output.ec_connector_worker_meta.saves == ["mm0", "mm2"]
79
80
tests/v1/ec_connector/unit/test_worker_ec_connector.py
+12
@@ -20,6 +20,7 @@ from vllm.v1.worker.gpu.ec_connector import NO_OP_EC_CONNECTOR, ActiveECConnecto
20
pytestmark = pytest.mark.cpu_test
21
22
WORKER_META = object()
23
+CONNECTOR_STATS = object()
24
25
26
def _scheduler_output() -> SimpleNamespace:
@@ -38,6 +39,7 @@ def _connector(
39
fake.is_consumer = is_consumer
40
fake.get_finished.return_value = (None, None)
41
fake.build_connector_worker_meta.return_value = WORKER_META
42
+ fake.get_ec_connector_stats.return_value = CONNECTOR_STATS
43
with patch("vllm.v1.worker.gpu.ec_connector.get_ec_transfer", return_value=fake):
44
return ActiveECConnector(SimpleNamespace(), encoder_cache or {}), fake
45
@@ -72,6 +74,16 @@ def test_worker_meta_is_reported_on_context_exit():
74
assert fake.clear_connector_metadata.called
75
76
77
+def test_connector_stats_are_reported_on_context_exit():
78
+ """Reported in the finally block, alongside the worker meta."""
79
+ connector, fake = _connector()
80
+
81
+ with connector.maybe_get_output(_scheduler_output()) as output:
82
+ assert output.ec_connector_stats is None
83
+
84
+ assert output.ec_connector_stats is CONNECTOR_STATS
85
+
86
+
87
def test_v1_producer_receives_existing_encoder_cache():
88
encoder_cache = {"mm_cached": None}
89
fake = MagicMock(spec=ECConnectorBase)
vllm/distributed/ec_transfer/ec_connector/base.py
+43
@@ -35,6 +35,12 @@ from vllm.v1.outputs import ECConnectorOutput
35
36
if TYPE_CHECKING:
37
from vllm.config import VllmConfig
38
+ from vllm.distributed.ec_transfer.ec_connector.metrics import (
39
+ ECConnectorPromMetrics,
40
+ ECConnectorStats,
41
+ PromMetric,
42
+ PromMetricT,
43
+ )
44
from vllm.v1.request import Request
45
46
logger = init_logger(__name__)
@@ -219,6 +225,17 @@ class ECConnectorBase(ABC):
225
"""
226
return None
227
228
+ def get_ec_connector_stats(self) -> "ECConnectorStats | None":
229
+ """
230
+ Get the EC connector stats collected during the last interval.
231
+
232
+ Callable on either a worker-side or scheduler-side connector
233
+ instance: the worker-side instance reports stats gathered during
234
+ cache save/load, while the scheduler-side instance reports stats
235
+ gathered during scheduling decisions.
236
+ """
237
+ return None
238
+
239
# ==============================
240
# Scheduler-side methods
241
# ==============================
@@ -336,3 +353,29 @@ class ECConnectorBase(ABC):
353
leave this as False.
354
"""
355
return False
356
+
357
+ @classmethod
358
+ def build_ec_connector_stats(
359
+ cls, data: dict[str, Any] | None = None
360
+ ) -> "ECConnectorStats | None":
361
+ """
362
+ ECConnectorStats resolution method. This method allows dynamically
363
+ registered connectors to return their own ECConnectorStats object,
364
+ which can implement custom aggregation logic on the data dict.
365
+ """
366
+ return None
367
+
368
+ @classmethod
369
+ def build_prom_metrics(
370
+ cls,
371
+ vllm_config: "VllmConfig",
372
+ metric_types: dict[type["PromMetric"], type["PromMetricT"]],
373
+ labelnames: list[str],
374
+ per_engine_labelvalues: dict[int, list[object]],
375
+ ) -> "ECConnectorPromMetrics | None":
376
+ """
377
+ Create an ECConnectorPromMetrics subclass which should register
378
+ per-connector Prometheus metrics and implement observe() to
379
+ expose connector transfer stats via Prometheus.
380
+ """
381
+ return None
vllm/distributed/ec_transfer/ec_connector/metrics.py
new
+175
@@ -0,0 +1,175 @@
1
+# SPDX-License-Identifier: Apache-2.0
2
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+from dataclasses import dataclass, field
4
+from typing import Any, TypeAlias, TypeVar
5
+
6
+from prometheus_client import Counter, Gauge, Histogram
7
+
8
+from vllm.config import ECTransferConfig, VllmConfig
9
+from vllm.distributed.ec_transfer.ec_connector.factory import ECConnectorFactory
10
+from vllm.logger import init_logger
11
+
12
+PromMetric: TypeAlias = Gauge | Counter | Histogram
13
+PromMetricT = TypeVar("PromMetricT", bound=PromMetric)
14
+
15
+logger = init_logger(__name__)
16
+
17
+
18
+@dataclass
19
+class ECConnectorStats:
20
+ """
21
+ Base class for EC Connector Stats, a container for transfer performance
22
+ metrics or otherwise important telemetry from the connector.
23
+ All sub-classes need to be serializable as stats are sent from worker to
24
+ logger process.
25
+ """
26
+
27
+ data: dict[str, Any] = field(default_factory=dict)
28
+
29
+ def reset(self):
30
+ """Reset the stats, clear the state."""
31
+ raise NotImplementedError
32
+
33
+ def aggregate(self, other: "ECConnectorStats") -> "ECConnectorStats":
34
+ """
35
+ Aggregate stats with another `ECConnectorStats` object.
36
+ """
37
+ raise NotImplementedError
38
+
39
+ def reduce(self) -> dict[str, int | float]:
40
+ """
41
+ Reduce the observations collected during a time interval to one or
42
+ more representative values (eg avg/median/sum of the series).
43
+ This is meant to be called by the logger to produce a summary of the
44
+ stats for the last time interval.
45
+ """
46
+ raise NotImplementedError
47
+
48
+ def is_empty(self) -> bool:
49
+ """Return True if the stats are empty."""
50
+ raise NotImplementedError
51
+
52
+
53
+class ECConnectorLogging:
54
+ def __init__(self, ec_transfer_config: ECTransferConfig | None):
55
+ # Instantiate the connector's stats class.
56
+ if ec_transfer_config and ec_transfer_config.ec_connector:
57
+ self.connector_cls = ECConnectorFactory.get_connector_class(
58
+ ec_transfer_config
59
+ )
60
+ self.reset()
61
+
62
+ def reset(self):
63
+ self.transfer_stats_accumulator: ECConnectorStats | None = None
64
+
65
+ def observe(self, transfer_stats_data: dict[str, Any]):
66
+ # Should not be called when an ECConnector is not configured.
67
+ assert self.connector_cls is not None
68
+ # Called periodically when connector syncs with the scheduler.
69
+ # Note that this is not the same as the logging interval.
70
+ # We expect transfer_stats_data to be aggregated across all workers and
71
+ # consist of observations from a single connector.
72
+ transfer_stats = self.connector_cls.build_ec_connector_stats(
73
+ transfer_stats_data
74
+ )
75
+ if transfer_stats is None:
76
+ logger.warning_once(
77
+ "The connector %s is collecting stats but "
78
+ "does not implement the "
79
+ "`build_ec_connector_stats` method. "
80
+ "Stats will not be logged.",
81
+ self.connector_cls,
82
+ )
83
+ return
84
+
85
+ if self.transfer_stats_accumulator is None:
86
+ self.transfer_stats_accumulator = transfer_stats
87
+ else:
88
+ # Accumulate last interval stats.
89
+ self.transfer_stats_accumulator = self.transfer_stats_accumulator.aggregate(
90
+ transfer_stats
91
+ )
92
+
93
+ def log(self, log_fn=logger.info):
94
+ """Log transfer metrics periodically, similar to throughput logging"""
95
+ if (
96
+ self.transfer_stats_accumulator
97
+ and not self.transfer_stats_accumulator.is_empty()
98
+ ):
99
+ # Produce a single cumulative stats object for the last time
100
+ # interval from the recorded observations.
101
+ xfer_metrics = self.transfer_stats_accumulator.reduce()
102
+ xfer_metrics_str = ", ".join(f"{k}={v}" for k, v in xfer_metrics.items())
103
+ log_fn("EC Transfer metrics: %s", xfer_metrics_str)
104
+
105
+ # Reset metrics for next interval
106
+ self.reset()
107
+
108
+
109
+class ECConnectorPromMetrics:
110
+ """
111
+ A base class for per-connector Prometheus metric registration
112
+ and recording.
113
+ """
114
+
115
+ def __init__(
116
+ self,
117
+ vllm_config: VllmConfig,
118
+ metric_types: dict[type[PromMetric], type[PromMetricT]],
119
+ labelnames: list[str],
120
+ per_engine_labelvalues: dict[int, list[object]],
121
+ ):
122
+ self._ec_transfer_config = vllm_config.ec_transfer_config
123
+ self._gauge_cls = metric_types[Gauge]
124
+ self._counter_cls = metric_types[Counter]
125
+ self._histogram_cls = metric_types[Histogram]
126
+ self._labelnames = labelnames
127
+ self.per_engine_labelvalues = per_engine_labelvalues
128
+
129
+ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
130
+ """
131
+ Record the supplied transfer statistics to Prometheus metrics. These
132
+ statistics are engine-specific, and should be recorded to a metric
133
+ with the appropriate 'engine' label. These metric instances can be
134
+ created using the create_metric_per_engine() helper method.
135
+ """
136
+ raise NotImplementedError
137
+
138
+
139
+class ECConnectorProm:
140
+ """
141
+ Support for registering per-connector Prometheus metrics, and
142
+ recording transfer statistics to those metrics. Uses
143
+ ECConnectorBase.build_prom_metrics().
144
+ """
145
+
146
+ _gauge_cls = Gauge
147
+ _counter_cls = Counter
148
+ _histogram_cls = Histogram
149
+
150
+ def __init__(
151
+ self,
152
+ vllm_config: VllmConfig,
153
+ labelnames: list[str],
154
+ per_engine_labelvalues: dict[int, list[object]],
155
+ ):
156
+ self.prom_metrics: ECConnectorPromMetrics | None = None
157
+ ec_transfer_config = vllm_config.ec_transfer_config
158
+ if ec_transfer_config and ec_transfer_config.ec_connector:
159
+ connector_cls = ECConnectorFactory.get_connector_class(ec_transfer_config)
160
+ metric_types = {
161
+ Gauge: self._gauge_cls,
162
+ Counter: self._counter_cls,
163
+ Histogram: self._histogram_cls,
164
+ }
165
+ self.prom_metrics = connector_cls.build_prom_metrics(
166
+ vllm_config,
167
+ metric_types,
168
+ labelnames,
169
+ per_engine_labelvalues,
170
+ )
171
+
172
+ def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
173
+ if self.prom_metrics is None:
174
+ return
175
+ self.prom_metrics.observe(transfer_stats_data, engine_idx)
vllm/distributed/ec_transfer/ec_connector/utils.py
+9
@@ -103,6 +103,7 @@ class ECOutputAggregator:
103
finished_sending = set[str]()
104
finished_recving = set[str]()
105
worker_meta = None
106
+ connector_stats = None
107
for model_runner_output in outputs:
108
assert model_runner_output is not None
109
ec_output = model_runner_output.ec_connector_output
@@ -117,9 +118,17 @@ class ECOutputAggregator:
118
meta if worker_meta is None else worker_meta.aggregate(meta)
119
)
120
121
+ if stats := ec_output.ec_connector_stats:
122
+ connector_stats = (
123
+ stats
124
+ if connector_stats is None
125
+ else connector_stats.aggregate(stats)
126
+ )
127
+
128
aggregated = ECConnectorOutput(
129
finished_sending=finished_sending or None,
130
finished_recving=finished_recving or None,
131
+ ec_connector_stats=connector_stats,
132
ec_connector_worker_meta=worker_meta,
133
)
134
if aggregated.is_empty():
vllm/v1/core/sched/scheduler.py
+24
@@ -15,6 +15,7 @@ from vllm.distributed.ec_transfer.ec_connector.base import (
15
ECConnectorRole,
16
)
17
from vllm.distributed.ec_transfer.ec_connector.factory import ECConnectorFactory
18
+from vllm.distributed.ec_transfer.ec_connector.metrics import ECConnectorStats
19
from vllm.distributed.kv_events import EventPublisherFactory, KVEventBatch
20
from vllm.distributed.kv_transfer.kv_connector.factory import KVConnectorFactory
21
from vllm.distributed.kv_transfer.kv_connector.v1 import (
@@ -2243,6 +2244,23 @@ class Scheduler(SchedulerInterface):
2244
else scheduler_kv_connector_stats
2245
)
2246
2247
+ # Worker-side EC connector stats from the model runner output.
2248
+ ec_connector_stats: ECConnectorStats | None = (
2249
+ ec_connector_output.ec_connector_stats if ec_connector_output else None
2250
+ )
2251
+ if self.ec_connector:
2252
+ # Scheduler-side EC connector stats collected after connector update.
2253
+ scheduler_ec_connector_stats = self.ec_connector.get_ec_connector_stats()
2254
+ if (
2255
+ scheduler_ec_connector_stats is not None
2256
+ and not scheduler_ec_connector_stats.is_empty()
2257
+ ):
2258
+ ec_connector_stats = (
2259
+ ec_connector_stats.aggregate(scheduler_ec_connector_stats)
2260
+ if ec_connector_stats is not None
2261
+ else scheduler_ec_connector_stats
2262
+ )
2263
+
2264
# collect KV cache events from KV cache manager
2265
events = self.kv_cache_manager.take_events()
2266
@@ -2287,6 +2305,7 @@ class Scheduler(SchedulerInterface):
2305
kv_connector_stats,
2306
cudagraph_stats,
2307
perf_stats,
2308
+ ec_connector_stats=ec_connector_stats,
2309
)
2310
) is not None:
2311
# Return stats to only one of the front-ends.
@@ -2772,6 +2791,7 @@ class Scheduler(SchedulerInterface):
2791
kv_connector_stats: KVConnectorStats | None = None,
2792
cudagraph_stats: CUDAGraphStat | None = None,
2793
perf_stats: PerfStats | None = None,
2794
+ ec_connector_stats: ECConnectorStats | None = None,
2795
) -> SchedulerStats | None:
2796
if not self.log_stats:
2797
return None
@@ -2790,6 +2810,9 @@ class Scheduler(SchedulerInterface):
2810
connector_stats_payload = (
2811
kv_connector_stats.to_dict() if kv_connector_stats else None
2812
)
2813
+ ec_connector_stats_payload = (
2814
+ ec_connector_stats.data if ec_connector_stats else None
2815
+ )
2816
return SchedulerStats(
2817
num_running_reqs=len(self.running),
2818
num_waiting_reqs=len(self.waiting),
@@ -2802,6 +2825,7 @@ class Scheduler(SchedulerInterface):
2825
kv_connector_stats=connector_stats_payload,
2826
cudagraph_stats=cudagraph_stats,
2827
perf_stats=perf_stats,
2828
+ ec_connector_stats=ec_connector_stats_payload,
2829
)
2830
2831
def make_spec_decoding_stats(
vllm/v1/metrics/loggers.py
+18
@@ -11,6 +11,10 @@ from prometheus_client import Counter, Gauge, Histogram
11
import vllm.envs as envs
12
from vllm.compilation.cuda_graph import CUDAGraphLogging
13
from vllm.config import SupportsMetricsInfo, VllmConfig
14
+from vllm.distributed.ec_transfer.ec_connector.metrics import (
15
+ ECConnectorLogging,
16
+ ECConnectorProm,
17
+)
18
from vllm.distributed.kv_transfer.kv_connector.v1.metrics import (
19
KVConnectorLogging,
20
KVConnectorProm,
@@ -118,6 +122,8 @@ class LoggingStatLogger(StatLoggerBase):
122
self.spec_decoding_logging = SpecDecodingLogging(is_diffusion=is_diffusion)
123
kv_transfer_config = self.vllm_config.kv_transfer_config
124
self.kv_connector_logging = KVConnectorLogging(kv_transfer_config)
125
+ ec_transfer_config = self.vllm_config.ec_transfer_config
126
+ self.ec_connector_logging = ECConnectorLogging(ec_transfer_config)
127
self.cudagraph_logging = None
128
if self.vllm_config.observability_config.cudagraph_metrics:
129
self.cudagraph_logging = CUDAGraphLogging(
@@ -224,6 +230,8 @@ class LoggingStatLogger(StatLoggerBase):
230
self.spec_decoding_logging.observe(scheduler_stats.spec_decoding_stats)
231
if kv_connector_stats := scheduler_stats.kv_connector_stats:
232
self.kv_connector_logging.observe(kv_connector_stats)
233
+ if ec_connector_stats := scheduler_stats.ec_connector_stats:
234
+ self.ec_connector_logging.observe(ec_connector_stats)
235
if (
236
self.cudagraph_logging is not None
237
and scheduler_stats.cudagraph_stats is not None
@@ -319,6 +327,7 @@ class LoggingStatLogger(StatLoggerBase):
327
328
self.spec_decoding_logging.log(log_fn=log_fn)
329
self.kv_connector_logging.log(log_fn=log_fn)
330
+ self.ec_connector_logging.log(log_fn=log_fn)
331
if self.cudagraph_logging is not None:
332
self.cudagraph_logging.log(log_fn=log_fn)
333
if self._enable_perf_stats():
@@ -451,6 +460,7 @@ class PrometheusStatLogger(AggregateStatLoggerBase):
460
_histogram_cls = Histogram
461
_spec_decoding_cls = SpecDecodingProm
462
_kv_connector_cls = KVConnectorProm
463
+ _ec_connector_cls = ECConnectorProm
464
_perf_metrics_cls = PerfMetricsProm
465
466
def __init__(
@@ -488,6 +498,9 @@ class PrometheusStatLogger(AggregateStatLoggerBase):
498
self.kv_connector_prom = self._kv_connector_cls(
499
vllm_config, labelnames, per_engine_labelvalues
500
)
501
+ self.ec_connector_prom = self._ec_connector_cls(
502
+ vllm_config, labelnames, per_engine_labelvalues
503
+ )
504
self.perf_metrics_prom = self._perf_metrics_cls(
505
vllm_config, labelnames, per_engine_labelvalues
506
)
@@ -1064,6 +1077,11 @@ class PrometheusStatLogger(AggregateStatLoggerBase):
1077
scheduler_stats.kv_connector_stats, engine_idx
1078
)
1079
1080
+ if scheduler_stats.ec_connector_stats is not None:
1081
+ self.ec_connector_prom.observe(
1082
+ scheduler_stats.ec_connector_stats, engine_idx
1083
+ )
1084
+
1085
if scheduler_stats.perf_stats is not None:
1086
self.perf_metrics_prom.observe(scheduler_stats.perf_stats, engine_idx)
1087
vllm/v1/metrics/stats.py
+1
@@ -205,6 +205,7 @@ class SchedulerStats:
205
206
spec_decoding_stats: SpecDecodingStats | None = None
207
kv_connector_stats: dict[str, Any] | None = None
208
+ ec_connector_stats: dict[str, Any] | None = None
209
210
waiting_lora_adapters: dict[str, int] = field(default_factory=dict)
211
running_lora_adapters: dict[str, int] = field(default_factory=dict)
vllm/v1/outputs.py
+3
@@ -16,6 +16,7 @@ from vllm.v1.core.sched.output import SchedulerOutput
16
17
if TYPE_CHECKING:
18
from vllm.distributed.ec_transfer.ec_connector.base import ECConnectorWorkerMetadata
19
+ from vllm.distributed.ec_transfer.ec_connector.metrics import ECConnectorStats
20
from vllm.distributed.kv_events import KVConnectorKVEvents
21
from vllm.distributed.kv_transfer.kv_connector.v1.base import (
22
KVConnectorWorkerMetadata,
@@ -307,12 +308,14 @@ class ECConnectorOutput:
308
# [mm_hash]
309
finished_sending: set[str] | None = None
310
finished_recving: set[str] | None = None
311
+ ec_connector_stats: "ECConnectorStats | None" = None
312
ec_connector_worker_meta: ECConnectorWorkerMetadata | None = None
313
314
def is_empty(self):
315
return (
316
not self.finished_sending
317
and not self.finished_recving
318
+ and not self.ec_connector_stats
319
and not self.ec_connector_worker_meta
320
)
321
vllm/v1/worker/gpu/ec_connector.py
+1
@@ -80,6 +80,7 @@ class ActiveECConnector(ECConnector):
80
output.finished_sending, output.finished_recving = (
81
ec_connector.get_finished(scheduler_output.finished_req_ids)
82
)
83
+ output.ec_connector_stats = ec_connector.get_ec_connector_stats()
84
output.ec_connector_worker_meta = ec_connector.build_connector_worker_meta()
85
ec_connector.clear_connector_metadata()
86