feat(observability): Add embedding cache metrics

components/src/dynamo/common/utils/prometheus.py

@@ -38,15 +38,18 @@
 
 
 # Single source of truth for embedding cache metric names.
+EMBEDDING_CACHE_METRIC_PREFIX = f"{name_prefix.COMPONENT}_embedding_cache"
+
+
 class EmbeddingCacheMetrics(str, enum.Enum):
     """Prometheus metric names for the multimodal embedding cache."""
 
-    HITS_TOTAL = f"{name_prefix.COMPONENT}_embedding_cache_hits_total"
-    MISSES_TOTAL = f"{name_prefix.COMPONENT}_embedding_cache_misses_total"
-    EVICTIONS_TOTAL = f"{name_prefix.COMPONENT}_embedding_cache_evictions_total"
-    UTILIZATION = f"{name_prefix.COMPONENT}_embedding_cache_utilization"
-    CURRENT_BYTES = f"{name_prefix.COMPONENT}_embedding_cache_current_bytes"
-    ENTRIES = f"{name_prefix.COMPONENT}_embedding_cache_entries"
+    HITS_TOTAL = f"{EMBEDDING_CACHE_METRIC_PREFIX}_hits_total"
+    MISSES_TOTAL = f"{EMBEDDING_CACHE_METRIC_PREFIX}_misses_total"
+    EVICTIONS_TOTAL = f"{EMBEDDING_CACHE_METRIC_PREFIX}_evictions_total"
+    UTILIZATION = f"{EMBEDDING_CACHE_METRIC_PREFIX}_utilization"
+    CURRENT_BYTES = f"{EMBEDDING_CACHE_METRIC_PREFIX}_current_bytes"
+    ENTRIES = f"{EMBEDDING_CACHE_METRIC_PREFIX}_entries"
 
 
 def register_engine_metrics_callback(

components/src/dynamo/vllm/main.py

@@ -31,6 +31,7 @@
 )
 from dynamo.common.utils.graceful_shutdown import install_signal_handlers
 from dynamo.common.utils.prometheus import (
+    EMBEDDING_CACHE_METRIC_PREFIX,
     LLMBackendMetrics,
     register_engine_metrics_callback,
 )
@@ -213,6 +214,20 @@ def setup_metrics_collection(
         Additional labels can be provided via inject_labels parameter.
     """
     metrics_model_name = get_metrics_model_name(config)
+
+    # The DynamoMultimodalEmbeddingCacheConnector (scheduler side, EngineCore
+    # process) publishes its cache metrics through the multiprocess .db files.
+    # Forward that family only when the connector is configured — the
+    # encode-routing path exposes the same metric names in-process via
+    # register_embedding_cache_metrics instead.
+    engine_metric_prefixes = ["vllm:", "lmcache:"]
+    ec_config = getattr(config.engine_args, "ec_transfer_config", None)
+    if (
+        getattr(ec_config, "ec_connector", None)
+        == "DynamoMultimodalEmbeddingCacheConnector"
+    ):
+        engine_metric_prefixes.append(EMBEDDING_CACHE_METRIC_PREFIX)
+
     if config.engine_args.disable_log_stats is False:
         # Register the dedicated dynamo_component registry callback
         # IMPORTANT: We do NOT use MultiProcessCollector for DYNAMO_COMPONENT_REGISTRY
@@ -242,7 +257,7 @@ def setup_metrics_collection(
                 register_engine_metrics_callback(
                     endpoint=generate_endpoint,
                     registry=REGISTRY,
-                    metric_prefix_filters=["vllm:", "lmcache:"],
+                    metric_prefix_filters=engine_metric_prefixes,
                     namespace_name=config.namespace,
                     component_name=config.component,
                     endpoint_name=config.endpoint,
@@ -272,7 +287,7 @@ def setup_metrics_collection(
                 register_engine_metrics_callback(
                     endpoint=generate_endpoint,
                     registry=multiproc_registry,
-                    metric_prefix_filters=["vllm:", "lmcache:"],
+                    metric_prefix_filters=engine_metric_prefixes,
                     namespace_name=config.namespace,
                     component_name=config.component,
                     endpoint_name=config.endpoint,
@@ -288,7 +303,7 @@ def setup_metrics_collection(
             register_engine_metrics_callback(
                 endpoint=generate_endpoint,
                 registry=REGISTRY,
-                metric_prefix_filters=["vllm:", "lmcache:"],
+                metric_prefix_filters=engine_metric_prefixes,
                 namespace_name=config.namespace,
                 component_name=config.component,
                 endpoint_name=config.endpoint,
@@ -554,6 +569,7 @@ def setup_vllm_engine(
         capacity_gb=config.multimodal_embedding_cache_capacity_gb,
         namespace=config.namespace,
         component=config.component,
+        model_name=get_metrics_model_name(config),
     )
 
     # Taken from build_async_engine_client_from_engine_args()

components/src/dynamo/vllm/multimodal_utils/cache_config.py

@@ -15,6 +15,7 @@ def configure_multimodal_embedding_cache(
     capacity_gb: float,
     namespace: str,
     component: str,
+    model_name: str = "",
 ) -> None:
     """Configure vLLM's CPU embedding cache before engine creation.
 
@@ -40,6 +41,11 @@ def configure_multimodal_embedding_cache(
             ),
             ec_connector_extra_config={
                 "multimodal_embedding_cache_capacity_gb": capacity_gb,
+                # Prometheus label values for the connector's cache metrics —
+                # the scheduler-side connector has no other channel to learn
+                # its Dynamo identity.
+                "component": component,
+                "model_name": model_name,
             },
         ),
     )

components/src/dynamo/vllm/multimodal_utils/multimodal_embedding_cache_connector.py

@@ -55,6 +55,105 @@ class MultimodalEmbeddingCacheConnectorMetadata(ECConnectorMetadata):
     evicts: list[str] = field(default_factory=list)
 
 
+class _SchedulerCacheMetrics:
+    """Prometheus metrics for the scheduler-side logical CPU cache.
+
+    Emits the same EmbeddingCacheMetrics family as
+    register_embedding_cache_metrics (the worker-layer
+    MultimodalEmbeddingCacheManager used by encode-routing deployments), so
+    dashboards see one embedding-cache family regardless of which
+    implementation serves the model.
+
+    The scheduler-side connector runs in vLLM's EngineCore process, not the
+    process serving /metrics. Values reach the frontend through Dynamo's
+    multiprocess Prometheus setup: PROMETHEUS_MULTIPROC_DIR is set before the
+    engine starts (see setup_vllm_engine), so prometheus_client mmap-persists
+    values that setup_metrics_collection's MultiProcessCollector exposes. The
+    private CollectorRegistry keeps these metrics out of this process's global
+    REGISTRY so they are never exported twice when the engine core runs
+    in-process.
+
+    This filesystem transport only covers EngineCore processes that inherit
+    PROMETHEUS_MULTIPROC_DIR and share a filesystem with the metrics-serving
+    process — the default local-subprocess topology. Remote EngineCore actors
+    (Ray data-parallel placement) don't inherit the env var and write to their
+    own node's filesystem, so their cache metrics are never exported; covering
+    them would need EC-connector stats plumbed through EngineCoreOutputs, as
+    KV connectors do. The connector logs a warning for these topologies at
+    init instead of failing.
+    """
+
+    def __init__(
+        self, model_name: str, component_name: str, capacity_bytes: int, dp_rank: int
+    ) -> None:
+        # Deferred imports: prometheus_client must be imported after
+        # PROMETHEUS_MULTIPROC_DIR is set (inherited from the parent process).
+        from prometheus_client import CollectorRegistry, Counter, Gauge
+
+        from dynamo.common.utils.prometheus import EmbeddingCacheMetrics as ECM
+        from dynamo.prometheus_names import labels
+
+        self._capacity_bytes = capacity_bytes
+        self.registry = CollectorRegistry()
+        # dp_rank partitions series per data-parallel EngineCore (each rank has
+        # its own scheduler-side cache), so gauges never collide across ranks
+        # and "mostrecent" only arbitrates between a dead pre-restart pid and
+        # its live replacement within one rank. Same pattern as the kvstats
+        # gauges in LLMBackendMetrics.
+        labelnames = [labels.MODEL, labels.COMPONENT, labels.DP_RANK]
+        labelvalues = {
+            labels.MODEL: model_name,
+            labels.COMPONENT: component_name,
+            labels.DP_RANK: str(dp_rank),
+        }
+
+        def _counter(name: str, doc: str):
+            return Counter(name, doc, labelnames, registry=self.registry).labels(
+                **labelvalues
+            )
+
+        def _gauge(name: str, doc: str):
+            # "mostrecent" so MultiProcessCollector reports this process's
+            # latest snapshot instead of aggregating across dead pids.
+            return Gauge(
+                name,
+                doc,
+                labelnames,
+                registry=self.registry,
+                multiprocess_mode="mostrecent",
+            ).labels(**labelvalues)
+
+        self._hits = _counter(ECM.HITS_TOTAL, "Total embedding cache hits.")
+        self._misses = _counter(ECM.MISSES_TOTAL, "Total embedding cache misses.")
+        self._evictions = _counter(
+            ECM.EVICTIONS_TOTAL, "Total embedding cache evictions."
+        )
+        self._utilization = _gauge(
+            ECM.UTILIZATION, "Cache memory utilization ratio (0.0-1.0)."
+        )
+        self._current_bytes = _gauge(
+            ECM.CURRENT_BYTES, "Current cache memory usage in bytes."
+        )
+        self._entries = _gauge(ECM.ENTRIES, "Number of entries in the cache.")
+        self.update_usage(0, 0)
+
+    def record_hit(self) -> None:
+        self._hits.inc()
+
+    def record_miss(self) -> None:
+        self._misses.inc()
+
+    def record_evictions(self, count: int) -> None:
+        self._evictions.inc(count)
+
+    def update_usage(self, used_bytes: int, entries: int) -> None:
+        self._current_bytes.set(used_bytes)
+        self._entries.set(entries)
+        self._utilization.set(
+            used_bytes / self._capacity_bytes if self._capacity_bytes else 0.0
+        )
+
+
 class DynamoMultimodalEmbeddingCacheConnector(ECConnectorBase):
     """EC connector with scheduler-authoritative CPU embedding cache.
 
@@ -111,6 +210,37 @@ def __init__(self, vllm_config: "VllmConfig", role: ECConnectorRole) -> None:
         self._saves_this_step: set[str] = set()
         self._evicts_this_step: set[str] = set()
 
+        # Only the scheduler role does cache accounting, so only it emits
+        # metrics; worker-role instances would just publish idle zeros.
+        self._metrics: _SchedulerCacheMetrics | None = None
+        if role == ECConnectorRole.SCHEDULER:
+            extra_config = transfer_config.ec_connector_extra_config
+            self._metrics = _SchedulerCacheMetrics(
+                model_name=extra_config.get("model_name", ""),
+                component_name=extra_config.get("component", ""),
+                capacity_bytes=self._capacity_bytes,
+                dp_rank=getattr(vllm_config.parallel_config, "data_parallel_rank", 0),
+            )
+            # The mmap transport (see _SchedulerCacheMetrics) needs the env
+            # var inherited from the metrics-serving process; without it the
+            # values stay in this process's memory.
+            if "PROMETHEUS_MULTIPROC_DIR" not in os.environ:
+                logger.warning(
+                    "Embedding cache metrics: PROMETHEUS_MULTIPROC_DIR is not "
+                    "set; cache metrics stay local to this process and will "
+                    "not be exported (expected under offline LLM() usage or "
+                    "remote Ray DP actors)."
+                )
+            if (
+                getattr(vllm_config.parallel_config, "data_parallel_backend", "")
+                == "ray"
+            ):
+                logger.warning(
+                    "Embedding cache metrics: data_parallel_backend=ray — "
+                    "remote DP ranks do not share the multiprocess Prometheus "
+                    "directory and will not contribute cache metrics."
+                )
+
         # --- Worker-side: dumb CPU tensor store ---
         self._cpu_store: dict[str, torch.Tensor] = {}
 
@@ -172,24 +302,36 @@ def update_state_after_alloc(self, request: "Request", index: int) -> None:
         if mm_hash in self._cache_order:
             self._cache_order.move_to_end(mm_hash)
             self._loads_this_step.add(mm_hash)
+            if self._metrics is not None:
+                self._metrics.record_hit()
             return
 
+        if self._metrics is not None:
+            self._metrics.record_miss()
+
         if size_bytes > self._capacity_bytes:
             return
 
         self._saves_this_step.add(mm_hash)
 
+        num_evicted = 0
         while (
             self._num_used_bytes + size_bytes > self._capacity_bytes
             and self._cache_order
         ):
             evicted_hash, evicted_bytes = self._cache_order.popitem(last=False)
             self._num_used_bytes -= evicted_bytes
             self._evicts_this_step.add(evicted_hash)
+            num_evicted += 1
 
         self._cache_order[mm_hash] = size_bytes
         self._num_used_bytes += size_bytes
 
+        if self._metrics is not None:
+            if num_evicted:
+                self._metrics.record_evictions(num_evicted)
+            self._metrics.update_usage(self._num_used_bytes, len(self._cache_order))
+
     def build_connector_meta(
         self, scheduler_output: SchedulerOutput
     ) -> ECConnectorMetadata: