feat(metrics): add NATS client metrics to prometheus_metrics_fmt (#2292)

Co-authored-by: Keiven Chang <keivenchang@users.noreply.github.com>

Author: keivenchang

components/metrics/src/main.rs

@@ -31,7 +31,6 @@ use dynamo_llm::kv_router::scheduler::KVHitRateEvent;
 use dynamo_llm::kv_router::KV_HIT_RATE_SUBJECT;
 use dynamo_runtime::{
     error, logging,
-    metrics::MetricsRegistry,
     traits::events::{EventPublisher, EventSubscriber},
     utils::{Duration, Instant},
     DistributedRuntime, ErrorContext, Result, Runtime, Worker,
@@ -137,14 +136,7 @@ async fn app(runtime: Runtime) -> Result<()> {
         .await
         .context("Unable to create unique instance of Count; possibly one already exists")?;
 
-    let target_component = {
-        let c = namespace.component(&config.component_name)?;
-        if let Some(ref model) = config.model_name {
-            c.add_labels(&[("model", model.as_str())])?
-        } else {
-            c
-        }
-    };
+    let target_component = namespace.component(&config.component_name)?;
     let target_endpoint = target_component.endpoint(&config.endpoint_name);
 
     let service_path = target_endpoint.path();

lib/bindings/python/rust/lib.rs

@@ -485,21 +485,6 @@ impl Component {
             Ok(())
         })
     }
-
-    /// Add constant labels to this component (for metrics). Returns a new Component with labels.
-    /// labels: list of (key, value) tuples.
-    fn add_labels(&self, labels: Vec<(String, String)>) -> PyResult<Component> {
-        use rs::metrics::MetricsRegistry as _;
-        let pairs: Vec<(&str, &str)> = labels
-            .iter()
-            .map(|(k, v)| (k.as_str(), v.as_str()))
-            .collect();
-        let inner = self.inner.clone().add_labels(&pairs).map_err(to_pyerr)?;
-        Ok(Component {
-            inner,
-            event_loop: self.event_loop.clone(),
-        })
-    }
 }
 
 #[pymethods]

lib/llm/src/discovery/watcher.rs

@@ -7,7 +7,6 @@ use anyhow::Context as _;
 use tokio::sync::{mpsc::Receiver, Notify};
 
 use dynamo_runtime::{
-    metrics::MetricsRegistry,
     pipeline::{
         network::egress::push_router::PushRouter, ManyOut, Operator, RouterMode, SegmentSource,
         ServiceBackend, SingleIn, Source,
@@ -170,8 +169,7 @@ impl ModelWatcher {
         let component = self
             .drt
             .namespace(&endpoint_id.namespace)?
-            .component(&endpoint_id.component)
-            .and_then(|c| c.add_labels(&[("model", &model_entry.name)]))?;
+            .component(&endpoint_id.component)?;
         let client = component.endpoint(&endpoint_id.name).client().await?;
 
         let Some(etcd_client) = self.drt.etcd_client() else {

lib/llm/src/entrypoint/input/common.rs

@@ -27,7 +27,6 @@ use dynamo_runtime::{
     component::Client,
     distributed::DistributedConfig,
     engine::{AsyncEngineStream, Data},
-    metrics::MetricsRegistry,
     pipeline::{
         Context, ManyOut, Operator, PushRouter, RouterMode, SegmentSource, ServiceBackend,
         ServiceEngine, ServiceFrontend, SingleIn, Source,
@@ -111,8 +110,7 @@ pub async fn prepare_engine(
             let endpoint_id = local_model.endpoint_id();
             let component = distributed_runtime
                 .namespace(&endpoint_id.namespace)?
-                .component(&endpoint_id.component)
-                .and_then(|c| c.add_labels(&[("model", card.slug().to_string().as_str())]))?;
+                .component(&endpoint_id.component)?;
 
             let client = component.endpoint(&endpoint_id.name).client().await?;
 

lib/llm/src/entrypoint/input/endpoint.rs

@@ -17,7 +17,6 @@ use crate::{
 };
 
 use dynamo_runtime::engine::AsyncEngineStream;
-use dynamo_runtime::metrics::MetricsRegistry;
 use dynamo_runtime::pipeline::{
     network::Ingress, Context, ManyOut, Operator, SegmentSource, ServiceBackend, SingleIn, Source,
 };
@@ -33,25 +32,9 @@ pub async fn run(
     let cancel_token = distributed_runtime.primary_token().clone();
     let endpoint_id: EndpointId = path.parse()?;
 
-    let model_name = match &engine_config {
-        EngineConfig::StaticFull { model, .. } | EngineConfig::StaticCore { model, .. } => {
-            Some(model.service_name().to_string())
-        }
-        EngineConfig::StaticRemote(model) | EngineConfig::Dynamic(model) => {
-            Some(model.service_name().to_string())
-        }
-    };
-
     let component = distributed_runtime
         .namespace(&endpoint_id.namespace)?
-        .component(&endpoint_id.component)
-        .and_then(|c| {
-            if let Some(ref name) = model_name {
-                c.add_labels(&[("model", name.as_str())])
-            } else {
-                Ok(c)
-            }
-        })?;
+        .component(&endpoint_id.component)?;
     let endpoint = component
         .service_builder()
         .create()

lib/runtime/examples/hello_world/src/bin/client.rs

@@ -14,8 +14,8 @@
 // limitations under the License.
 
 use dynamo_runtime::{
-    logging, metrics::MetricsRegistry, pipeline::PushRouter, protocols::annotated::Annotated,
-    stream::StreamExt, DistributedRuntime, Result, Runtime, Worker,
+    logging, pipeline::PushRouter, protocols::annotated::Annotated, stream::StreamExt,
+    DistributedRuntime, Result, Runtime, Worker,
 };
 use hello_world::DEFAULT_NAMESPACE;
 
@@ -31,7 +31,6 @@ async fn app(runtime: Runtime) -> Result<()> {
     let client = distributed
         .namespace(DEFAULT_NAMESPACE)?
         .component("backend")?
-        .add_labels(&[("model", "hello_world_model")])?
         .endpoint("generate")
         .client()
         .await?;

lib/runtime/examples/hello_world/src/bin/server.rs

@@ -15,7 +15,6 @@
 
 use dynamo_runtime::{
     logging,
-    metrics::MetricsRegistry,
     pipeline::{
         async_trait, network::Ingress, AsyncEngine, AsyncEngineContextProvider, Error, ManyOut,
         ResponseStream, SingleIn,
@@ -70,7 +69,6 @@ async fn backend(runtime: DistributedRuntime) -> Result<()> {
     runtime
         .namespace(DEFAULT_NAMESPACE)?
         .component("backend")?
-        .add_labels(&[("model", "hello_world_model")])?
         .service_builder()
         .create()
         .await?

lib/runtime/examples/service_metrics/src/bin/service_client.rs

@@ -17,8 +17,8 @@ use futures::StreamExt;
 use service_metrics::DEFAULT_NAMESPACE;
 
 use dynamo_runtime::{
-    logging, metrics::MetricsRegistry, pipeline::PushRouter, protocols::annotated::Annotated,
-    utils::Duration, DistributedRuntime, Result, Runtime, Worker,
+    logging, pipeline::PushRouter, protocols::annotated::Annotated, utils::Duration,
+    DistributedRuntime, Result, Runtime, Worker,
 };
 
 fn main() -> Result<()> {
@@ -31,9 +31,7 @@ async fn app(runtime: Runtime) -> Result<()> {
     let distributed = DistributedRuntime::from_settings(runtime.clone()).await?;
 
     let namespace = distributed.namespace(DEFAULT_NAMESPACE)?;
-    let component = namespace
-        .component("backend")?
-        .add_labels(&[("model", "service_metrics_model")])?;
+    let component = namespace.component("backend")?;
 
     let client = component.endpoint("generate").client().await?;
 

lib/runtime/examples/service_metrics/src/bin/service_server.rs

@@ -17,7 +17,6 @@ use service_metrics::{MyStats, DEFAULT_NAMESPACE};
 
 use dynamo_runtime::{
     logging,
-    metrics::MetricsRegistry,
     pipeline::{
         async_trait, network::Ingress, AsyncEngine, AsyncEngineContextProvider, Error, ManyOut,
         ResponseStream, SingleIn,
@@ -72,7 +71,6 @@ async fn backend(runtime: DistributedRuntime) -> Result<()> {
     runtime
         .namespace(DEFAULT_NAMESPACE)?
         .component("backend")?
-        .add_labels(&[("model", "service_metrics_model")])?
         .service_builder()
         .create()
         .await?

lib/runtime/examples/system_metrics/src/lib.rs

@@ -91,7 +91,6 @@ pub async fn backend(drt: DistributedRuntime, endpoint_name: Option<&str>) -> Re
     let endpoint = drt
         .namespace(DEFAULT_NAMESPACE)?
         .component(DEFAULT_COMPONENT)?
-        .add_labels(&[("model", DEFAULT_MODEL_NAME)])?
         .service_builder()
         .create()
         .await?