chore(kv-router): untangle runtime from metrics

Signed-off-by: PeaBrane <yanrpei@gmail.com>

Author: PeaBrane

Cargo.lock

@@ -43,7 +43,7 @@ dynamo-config = { path = "lib/config", version = "1.0.0" }
 dynamo-tokens = { path = "lib/tokens", version = "1.0.0" }
 dynamo-memory = { path = "lib/memory", version = "1.0.0" }
 dynamo-mocker = { path = "lib/mocker", version = "1.0.0" }
-dynamo-kv-router = { path = "lib/kv-router", version = "1.0.0", features = ["metrics"] }
+dynamo-kv-router = { path = "lib/kv-router", version = "1.0.0", features = ["metrics", "runtime-protocols"] }
 dynamo-async-openai = { path = "lib/async-openai", version = "1.0.0", features = ["byot"] }
 dynamo-parsers = { path = "lib/parsers", version = "1.0.0" }
 

Cargo.toml

@@ -12,10 +12,11 @@ repository.workspace = true
 
 [features]
 default = []
-metrics = ["dep:dynamo-runtime"]
-bench = ["dep:clap", "dep:indicatif", "dep:serde_json", "dep:plotters"]
+metrics = ["dep:prometheus"]
+runtime-protocols = ["dep:dynamo-runtime"]
+bench = []
 standalone-indexer = ["dep:axum", "dep:bytes", "dep:zeromq", "dep:serde_json", "dep:reqwest"]
-indexer-runtime = ["metrics", "standalone-indexer"]
+indexer-runtime = ["metrics", "runtime-protocols", "standalone-indexer"]
 
 [dependencies]
 # repo
@@ -29,7 +30,7 @@ dashmap = { workspace = true }
 ordered-float = { workspace = true }
 derive_builder = { workspace = true }
 derive-getters = { workspace = true }
-prometheus = { workspace = true }
+prometheus = { workspace = true, optional = true }
 rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true, optional = true }
@@ -46,10 +47,6 @@ flume = "0.12.0"
 parking_lot = { workspace = true }
 rmp-serde = { workspace = true }
 
-# bench (optional)
-clap = { version = "4.5", features = ["derive"], optional = true }
-indicatif = { version = "0.18.0", optional = true }
-plotters = { version = "0.3", optional = true, default-features = false, features = ["svg_backend", "line_series", "point_series", "full_palette"] }
 rustc-hash = "2.1.1"
 
 # standalone-indexer (optional)
@@ -58,9 +55,6 @@ bytes = { workspace = true, optional = true }
 reqwest = { workspace = true, optional = true }
 zeromq = { version = "0.4.1", optional = true }
 
-[package.metadata.cargo-machete]
-ignored = ["clap", "indicatif", "plotters"]
-
 [dev-dependencies]
 rstest = "0.18.2"
 rstest_reuse = "0.7.0"

lib/kv-router/Cargo.toml

@@ -1,14 +1,16 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-#[cfg(feature = "metrics")]
-use std::sync::{Arc, OnceLock};
+#[cfg(feature = "runtime-protocols")]
+use std::sync::Arc;
+#[cfg(all(feature = "metrics", feature = "runtime-protocols"))]
+use std::sync::OnceLock;
 
+#[cfg(feature = "runtime-protocols")]
+use dynamo_runtime::component::Component;
+#[cfg(all(feature = "metrics", feature = "runtime-protocols"))]
+use dynamo_runtime::metrics::MetricsHierarchy;
 #[cfg(feature = "metrics")]
-use dynamo_runtime::{
-    component::Component,
-    metrics::{MetricsHierarchy, prometheus_names::kvrouter},
-};
 use prometheus::{IntCounterVec, Opts};
 
 use crate::protocols::{KvCacheEventData, KvCacheEventError};
@@ -17,9 +19,17 @@ use crate::protocols::{KvCacheEventData, KvCacheEventError};
 #[derive(Clone)]
 pub struct KvIndexerMetrics {
     /// Counter of events applied.
+    #[cfg(feature = "metrics")]
     pub kv_cache_events_applied: IntCounterVec,
 }
 
+#[cfg(not(feature = "metrics"))]
+impl Default for KvIndexerMetrics {
+    fn default() -> Self {
+        Self {}
+    }
+}
+
 /// Metric status labels.
 pub const METRIC_STATUS_OK: &str = "ok";
 pub const METRIC_STATUS_PARENT_NOT_FOUND: &str = "parent_block_not_found";
@@ -32,9 +42,10 @@ pub const METRIC_EVENT_REMOVED: &str = "removed";
 pub const METRIC_EVENT_CLEARED: &str = "cleared";
 
 /// Metric name for KV cache events applied counter.
+const KV_CACHE_EVENTS_APPLIED_SUFFIX: &str = "kv_cache_events_applied";
 const KV_CACHE_EVENTS_APPLIED_NAME: &str = "dynamo_kvrouter_kv_cache_events_applied";
 
-#[cfg(feature = "metrics")]
+#[cfg(all(feature = "metrics", feature = "runtime-protocols"))]
 static KV_INDEXER_METRICS: OnceLock<Arc<KvIndexerMetrics>> = OnceLock::new();
 
 impl KvIndexerMetrics {
@@ -47,26 +58,40 @@ impl KvIndexerMetrics {
 
     /// Creates a new KvIndexerMetrics from a Component, memoizing the result in
     /// KV_INDEXER_METRICS to avoid duplicate registration issues.
-    #[cfg(feature = "metrics")]
+    #[cfg(feature = "runtime-protocols")]
     pub fn from_component(component: &Component) -> Arc<Self> {
-        KV_INDEXER_METRICS.get_or_init(|| {
-            match component.metrics().create_intcountervec(
-                kvrouter::KV_CACHE_EVENTS_APPLIED,
-                "Total number of KV cache events applied to index",
-                &["event_type", "status"],
-                &[],
-            ) {
-                Ok(kv_cache_events_applied) => Arc::new(Self::new(kv_cache_events_applied)),
-                Err(e) => {
-                    tracing::warn!("Failed to create kv indexer metrics from component: {}. Using unregistered metrics as fallback.", e);
-                    Arc::new(Self::new_unregistered())
-                }
-            }
-        }).clone()
+        #[cfg(feature = "metrics")]
+        {
+            KV_INDEXER_METRICS
+                .get_or_init(|| {
+                    match component.metrics().create_intcountervec(
+                        KV_CACHE_EVENTS_APPLIED_SUFFIX,
+                        "Total number of KV cache events applied to index",
+                        &["event_type", "status"],
+                        &[],
+                    ) {
+                        Ok(kv_cache_events_applied) => {
+                            Arc::new(Self::new(kv_cache_events_applied))
+                        }
+                        Err(e) => {
+                            tracing::warn!("Failed to create kv indexer metrics from component: {}. Using unregistered metrics as fallback.", e);
+                            Arc::new(Self::new_unregistered())
+                        }
+                    }
+                })
+                .clone()
+        }
+
+        #[cfg(not(feature = "metrics"))]
+        {
+            let _ = component;
+            Arc::new(Self::new_unregistered())
+        }
     }
 
     /// Creates a new KvIndexerMetrics which is not registered with a MetricsRegistry.
     /// This may be used for tests or as a fallback for when a MetricsRegistry is not available / has errored.
+    #[cfg(feature = "metrics")]
     pub fn new_unregistered() -> Self {
         Self {
             kv_cache_events_applied: IntCounterVec::new(
@@ -80,6 +105,12 @@ impl KvIndexerMetrics {
         }
     }
 
+    /// Creates a no-op metrics instance when Prometheus support is disabled.
+    #[cfg(not(feature = "metrics"))]
+    pub fn new_unregistered() -> Self {
+        Self::default()
+    }
+
     pub fn get_event_type(event_data: &KvCacheEventData) -> &'static str {
         match event_data {
             KvCacheEventData::Stored(_) => METRIC_EVENT_STORED,
@@ -93,22 +124,27 @@ impl KvIndexerMetrics {
         event_type: &'static str,
         result: Result<(), KvCacheEventError>,
     ) {
-        match result {
-            Ok(_) => {
-                self.kv_cache_events_applied
-                    .with_label_values(&[event_type, METRIC_STATUS_OK])
-                    .inc_by(1);
-            }
-            Err(e) => {
-                let error_label = match e {
-                    KvCacheEventError::ParentBlockNotFound => METRIC_STATUS_PARENT_NOT_FOUND,
-                    KvCacheEventError::BlockNotFound => METRIC_STATUS_BLOCK_NOT_FOUND,
-                    KvCacheEventError::InvalidBlockSequence => METRIC_STATUS_INVALID_BLOCK,
-                };
-                self.kv_cache_events_applied
-                    .with_label_values(&[event_type, error_label])
-                    .inc_by(1);
+        #[cfg(feature = "metrics")]
+        {
+            match result {
+                Ok(_) => {
+                    self.kv_cache_events_applied
+                        .with_label_values(&[event_type, METRIC_STATUS_OK])
+                        .inc_by(1);
+                }
+                Err(e) => {
+                    let error_label = match e {
+                        KvCacheEventError::ParentBlockNotFound => METRIC_STATUS_PARENT_NOT_FOUND,
+                        KvCacheEventError::BlockNotFound => METRIC_STATUS_BLOCK_NOT_FOUND,
+                        KvCacheEventError::InvalidBlockSequence => METRIC_STATUS_INVALID_BLOCK,
+                    };
+                    self.kv_cache_events_applied
+                        .with_label_values(&[event_type, error_label])
+                        .inc_by(1);
+                }
             }
         }
+        #[cfg(not(feature = "metrics"))]
+        let _ = (self, event_type, result);
     }
 }

lib/kv-router/src/indexer/metrics.rs

@@ -4,10 +4,6 @@
 #[cfg(feature = "bench")]
 use std::time::Instant;
 
-#[cfg(feature = "metrics")]
-use dynamo_runtime::error::DynamoError;
-#[cfg(feature = "metrics")]
-pub use dynamo_runtime::protocols::maybe_error::MaybeError;
 use serde::{Deserialize, Serialize};
 use tokio::sync::{mpsc, oneshot};
 
@@ -16,7 +12,6 @@ use dynamo_tokens::SequenceHash;
 
 /// Trait for types that may represent an error response.
 /// Used for RPC-style responses that can indicate success or failure.
-#[cfg(not(feature = "metrics"))]
 pub trait MaybeError {
     /// Construct an instance from an error.
     fn from_err(err: impl std::error::Error + 'static) -> Self;
@@ -75,15 +70,30 @@ pub enum WorkerKvQueryResponse {
     Error(String),
 }
 
-#[cfg(feature = "metrics")]
 impl MaybeError for WorkerKvQueryResponse {
     fn from_err(err: impl std::error::Error + 'static) -> Self {
         WorkerKvQueryResponse::Error(err.to_string())
     }
 
-    fn err(&self) -> Option<DynamoError> {
+    fn err(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
+        match self {
+            WorkerKvQueryResponse::Error(msg) => Some(Box::new(std::io::Error::other(msg.clone()))),
+            _ => None,
+        }
+    }
+}
+
+#[cfg(feature = "runtime-protocols")]
+impl dynamo_runtime::protocols::maybe_error::MaybeError for WorkerKvQueryResponse {
+    fn from_err(err: impl std::error::Error + 'static) -> Self {
+        WorkerKvQueryResponse::Error(err.to_string())
+    }
+
+    fn err(&self) -> Option<dynamo_runtime::error::DynamoError> {
         match self {
-            WorkerKvQueryResponse::Error(msg) => Some(DynamoError::msg(msg.clone())),
+            WorkerKvQueryResponse::Error(msg) => {
+                Some(dynamo_runtime::error::DynamoError::msg(msg.clone()))
+            }
             _ => None,
         }
     }
@@ -147,29 +157,30 @@ pub enum IndexerQueryResponse {
     Error(String),
 }
 
-#[cfg(feature = "metrics")]
 impl MaybeError for IndexerQueryResponse {
     fn from_err(err: impl std::error::Error + 'static) -> Self {
         IndexerQueryResponse::Error(err.to_string())
     }
 
-    fn err(&self) -> Option<DynamoError> {
+    fn err(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
         match self {
-            IndexerQueryResponse::Error(msg) => Some(DynamoError::msg(msg.clone())),
+            IndexerQueryResponse::Error(msg) => Some(Box::new(std::io::Error::other(msg.clone()))),
             _ => None,
         }
     }
 }
 
-#[cfg(not(feature = "metrics"))]
-impl MaybeError for IndexerQueryResponse {
+#[cfg(feature = "runtime-protocols")]
+impl dynamo_runtime::protocols::maybe_error::MaybeError for IndexerQueryResponse {
     fn from_err(err: impl std::error::Error + 'static) -> Self {
         IndexerQueryResponse::Error(err.to_string())
     }
 
-    fn err(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
+    fn err(&self) -> Option<dynamo_runtime::error::DynamoError> {
         match self {
-            IndexerQueryResponse::Error(msg) => Some(Box::new(std::io::Error::other(msg.clone()))),
+            IndexerQueryResponse::Error(msg) => {
+                Some(dynamo_runtime::error::DynamoError::msg(msg.clone()))
+            }
             _ => None,
         }
     }