Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 24 additions & 12 deletions components/src/dynamo/vllm/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -412,7 +412,7 @@ def setup_vllm_engine(
config: Config,
stat_logger: Optional[StatLoggerFactory] = None,
fpm_worker_id: Optional[str] = None,
) -> tuple[AsyncLLM, VllmConfig, Any, Any, LLMBackendMetrics]:
) -> tuple[AsyncLLM, VllmConfig, Any, Any, Optional[LLMBackendMetrics]]:
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
# vLLM v0.11.0 bug: vllm/v1.metrics/prometheus.py:79 passes TemporaryDirectory object
# instead of .name string, causing false error on exit. Set PROMETHEUS_MULTIPROC_DIR
# ourselves to avoid this and handle cleanup properly.
Expand All @@ -437,16 +437,24 @@ def setup_vllm_engine(

# Construct Prometheus gauges AFTER setup_multiprocess_prometheus() so Gauge objects
# see the correct ValueClass (multiprocess vs in-memory).
component_gauges = LLMBackendMetrics(
registry=DYNAMO_COMPONENT_REGISTRY,
model_name=config.served_model_name or "",
component_name=config.component or "",
)
#
# Embedding workers (pooling engines) have no KV cache, no scheduler
# gauges, and no model_load_time hook -- registering the chat-shaped
# LLMBackendMetrics on them publishes zeros forever. Skip the
# construction entirely on that path so /metrics stays clean.
embedding_worker = stat_logger is not None and stat_logger.embedding_worker
component_gauges: Optional[LLMBackendMetrics] = None
if not embedding_worker:
component_gauges = LLMBackendMetrics(
registry=DYNAMO_COMPONENT_REGISTRY,
model_name=config.served_model_name or "",
component_name=config.component or "",
)

# If a StatLoggerFactory was provided, give it the gauges so the loggers
# it creates can publish Prometheus metrics.
if stat_logger is not None:
stat_logger.component_gauges = component_gauges
# If a StatLoggerFactory was provided, give it the gauges so the loggers
# it creates can publish Prometheus metrics.
if stat_logger is not None:
stat_logger.component_gauges = component_gauges

os.environ["VLLM_NO_USAGE_STATS"] = "1" # Avoid internal HTTP requests
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
Expand Down Expand Up @@ -575,8 +583,12 @@ def setup_vllm_engine(
)
load_time = time.time() - start_time

# Record model load time
component_gauges.set_model_load_time(load_time)
# Record model load time. ``component_gauges`` is None on the
# embedding-worker path -- pooling engines have no chat-shaped gauges
# registered, so model_load_time has no collector to publish to.
# Skip rather than fabricating a zero-valued sample.
if component_gauges is not None:
component_gauges.set_model_load_time(load_time)

logger.info(f"VllmWorker for {config.served_model_name} has been initialized")

Expand Down
44 changes: 44 additions & 0 deletions components/src/dynamo/vllm/publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,19 +85,63 @@ def log_engine_initialized(self) -> None:
pass


class _NoopStatLogger(StatLoggerBase):
Comment thread
tzulingk marked this conversation as resolved.
Resolved
"""Stat logger that drops every record. Installed on embedding workers,
where vLLM still calls the factory during engine init but the
chat-shaped publish path (KV cache usage, scheduler gauges) has no
meaning on a pooling engine."""

def __init__(
self,
vllm_config: Optional[VllmConfig] = None,
engine_index: int = 0,
) -> None:
# vLLM's ``StatLoggerBase`` declares ``__init__`` as abstract, so
# subclasses must provide one even when they hold no state.
# Without this, instantiation raises ``TypeError: Can't
# instantiate abstract class _NoopStatLogger without an
# implementation for abstract method '__init__'``. The
# ``(vllm_config, engine_index)`` signature mirrors what vLLM
# passes to concrete ``StatLoggerBase`` subclasses, so this
# logger remains a drop-in if vLLM ever wires the factory call
# to invoke the constructor directly.
del vllm_config, engine_index

def record(
self,
scheduler_stats: Optional[SchedulerStats],
iteration_stats: Optional[IterationStats],
mm_cache_stats: object = None,
engine_idx: int = 0,
*args: object,
**kwargs: object,
) -> None:
return

def log_engine_initialized(self) -> None:
pass


class StatLoggerFactory:
"""Factory for creating stat logger publishers. Required by vLLM."""

def __init__(
self,
endpoint: Endpoint,
component_gauges: Optional[LLMBackendMetrics] = None,
embedding_worker: bool = False,
Comment thread
tzulingk marked this conversation as resolved.
Resolved
) -> None:
self.endpoint = endpoint
self.component_gauges = component_gauges
self.embedding_worker = embedding_worker
self.created_logger: Optional[DynamoStatLoggerPublisher] = None

def create_stat_logger(self, dp_rank: int) -> StatLoggerBase:
# Embedding workers have no KV cache and no scheduler stats worth
# publishing -- short-circuit before constructing the chat-shaped
# WorkerMetricsPublisher and skipping the component_gauges check.
if self.embedding_worker:
return _NoopStatLogger()
# component_gauges must be set by setup_vllm_engine() before vLLM
# calls create_stat_logger() during engine initialization.
assert (
Expand Down
127 changes: 127 additions & 0 deletions components/src/dynamo/vllm/tests/test_vllm_publisher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Unit tests for the vLLM stat-logger factory.

These tests focus on the embedding-worker gating path: vLLM workers run
either a chat/decode engine or a pooling (embedding) engine, and the
chat-shaped Prometheus collectors are only meaningful on the former.
The factory is the single seam where vLLM calls into dynamo per dp_rank,
so it is also the seam where the embedding worker must short-circuit
the chat-shaped pipeline.
"""

from types import SimpleNamespace
from unittest.mock import Mock

import pytest

import dynamo.vllm.publisher as publisher_mod
from dynamo.vllm.publisher import (
DynamoStatLoggerPublisher,
StatLoggerFactory,
_NoopStatLogger,
)

pytestmark = [
pytest.mark.unit,
pytest.mark.vllm,
pytest.mark.core,
pytest.mark.gpu_0,
pytest.mark.pre_merge,
]
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved


def test_factory_returns_noop_logger_for_embedding_worker(monkeypatch):
"""``create_stat_logger`` returns a ``_NoopStatLogger`` on the
embedding path -- no ``DynamoStatLoggerPublisher`` /
``WorkerMetricsPublisher`` / NATS endpoint construction.

Why this matters: ``DynamoStatLoggerPublisher.__init__`` schedules a
``create_endpoint`` task on the runtime and registers chat-shaped
publish callbacks. On a pooling engine there is no kv_cache_usage to
publish (vLLM never emits ``SchedulerStats``) and the endpoint is
never queried -- so the factory must not construct it at all.
"""

def _explode(*_a, **_kw):
raise AssertionError(
"embedding-worker path must not construct DynamoStatLoggerPublisher"
)

monkeypatch.setattr(publisher_mod, "DynamoStatLoggerPublisher", _explode)

factory = StatLoggerFactory(
endpoint=SimpleNamespace(),
embedding_worker=True,
)

logger = factory.create_stat_logger(dp_rank=0)

assert isinstance(logger, _NoopStatLogger)
# Embedding factory never tracks a created chat logger, so the
# downstream ``init_publish`` / ``set_num_gpu_blocks_all`` calls in
# the chat path are safe no-ops if anyone ever wires them on the
# embedding branch by mistake.
assert factory.created_logger is None


def test_noop_stat_logger_record_is_safe_with_none_stats():
"""vLLM calls ``record`` every iteration even when the engine has
nothing useful to report. The chat-path publisher returns early on
``scheduler_stats is None``; the embedding noop must accept the same
shape (and the variadic mm/engine_idx args vLLM passes) without
raising."""
logger = _NoopStatLogger()

# Mirrors the call shape from vllm/v1/metrics/loggers.py.
logger.record(None, None)
logger.record(None, None, None, 0)
logger.record(
scheduler_stats=None,
iteration_stats=None,
mm_cache_stats=None,
engine_idx=0,
)
logger.log_engine_initialized()


def test_factory_embedding_flag_skips_component_gauges_assert():
"""On the chat path the factory asserts
``component_gauges is not None`` because ``setup_vllm_engine`` is
responsible for setting it before vLLM invokes the factory. The
embedding path skips that step entirely (no chat-shaped gauges to
register), so the factory must not blow up when it stays None."""
factory = StatLoggerFactory(
endpoint=SimpleNamespace(),
embedding_worker=True,
)
assert factory.component_gauges is None

# Would AssertionError on the chat path; must succeed here.
logger = factory.create_stat_logger(dp_rank=0)
assert isinstance(logger, _NoopStatLogger)


def test_factory_default_is_chat_path(monkeypatch):
"""Sibling check: the default (``embedding_worker=False``) still
constructs ``DynamoStatLoggerPublisher`` so the gating doesn't
accidentally swallow the chat path."""
constructed = []

def _fake_publisher(*args, **kwargs):
constructed.append(kwargs)
return Mock(spec=DynamoStatLoggerPublisher)

monkeypatch.setattr(publisher_mod, "DynamoStatLoggerPublisher", _fake_publisher)

endpoint = SimpleNamespace()
component_gauges = SimpleNamespace()
factory = StatLoggerFactory(endpoint=endpoint, component_gauges=component_gauges)

factory.create_stat_logger(dp_rank=3)

assert len(constructed) == 1
assert constructed[0]["endpoint"] is endpoint
assert constructed[0]["dp_rank"] == 3
assert constructed[0]["component_gauges"] is component_gauges
15 changes: 13 additions & 2 deletions components/src/dynamo/vllm/worker_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,10 @@
logger = logging.getLogger(__name__)

# (engine_client, vllm_config, default_sampling_params, prometheus_temp_dir, component_gauges)
EngineSetupResult = tuple[AsyncLLM, VllmConfig, Any, Any, LLMBackendMetrics]
# component_gauges is None on the embedding-worker path: pooling engines
# have no KV cache / scheduler gauges, so setup_vllm_engine() skips the
# LLMBackendMetrics registration there.
EngineSetupResult = tuple[AsyncLLM, VllmConfig, Any, Any, Optional[LLMBackendMetrics]]


async def _wait_and_load_benchmark(bench_cfg: dict, vllm_config: VllmConfig) -> dict:
Expand Down Expand Up @@ -254,7 +257,15 @@ async def _create_embedding_worker(
shutdown_endpoints[:] = [generate_endpoint]

fpm_worker_id = str(generate_endpoint.connection_id())
factory = StatLoggerFactory(endpoint=generate_endpoint)
# Embedding workers run on pooling engines: no KV cache, no
# scheduler stats, no decode loop. The factory still has to exist
# because vLLM unconditionally invokes it during AsyncLLM init,
# but it returns a no-op stat logger and setup_vllm_engine() skips
# the chat-shaped LLMBackendMetrics registration.
factory = StatLoggerFactory(
endpoint=generate_endpoint,
embedding_worker=True,
)
(
engine_client,
vllm_config,
Expand Down
Loading