feat: add component-level NATS service metrics with auto-registration

- Add ComponentSystemStatusNatsMetrics struct for component service stats
- Implement automatic metrics registration in Namespace::component()
- Centralize NATS metric names in prometheus_names module
- Add filtering support for NATS metrics in tests
- Move ComponentSystemStatusNatsMetrics to service.rs module
- Update metric callbacks to use proper runtime prefix hierarchy

Author: keivenchang

lib/runtime/examples/Cargo.lock

@@ -30,7 +30,10 @@
 //! TODO: Top-level Overview of Endpoints/Functions
 
 use crate::{
-    config::HealthStatus, discovery::Lease, metrics::MetricsRegistry, service::ServiceSet,
+    config::HealthStatus,
+    discovery::Lease,
+    metrics::{prometheus_names, MetricsRegistry},
+    service::ServiceSet,
     transports::etcd::EtcdPath,
 };
 
@@ -247,6 +250,7 @@ impl Component {
         Ok(out)
     }
 
+    /// Scrape user defined stats (embedded in data field of ServiceInfo)
     pub async fn scrape_stats(&self, timeout: Duration) -> Result<ServiceSet> {
         let service_name = self.service_name();
         let service_client = self.drt().service_client();
@@ -255,6 +259,49 @@ impl Component {
             .await
     }
 
+    /// Register Prometheus metrics for this component's service stats
+    pub fn register_metrics(&self) -> Result<crate::service::ComponentSystemStatusNatsMetrics> {
+        let component_metrics =
+            crate::service::ComponentSystemStatusNatsMetrics::from_component(self)?;
+
+        // Create a callback that scrapes stats and updates metrics when called
+        let metrics_clone = component_metrics.clone();
+        let component_clone = self.clone();
+        let prefix = self.prefix();
+        let service_name = self.service_name();
+        let prefix_for_closure = prefix.clone();
+        self.drt().add_metrics_callback(&prefix, move |_runtime| {
+            println!(
+                "[DEBUG]CALLING  metrics callback for component: {}, prefix:{}",
+                service_name, prefix_for_closure
+            );
+            // Use tokio::runtime::Handle to run async code in the callback
+            let handle = tokio::runtime::Handle::try_current();
+            if let Ok(handle) = handle {
+                let metrics_ref = metrics_clone.clone();
+                let comp_ref = component_clone.clone();
+                handle.spawn(async move {
+                    let timeout = std::time::Duration::from_millis(500);
+                    match comp_ref.scrape_stats(timeout).await {
+                        Ok(service_set) => {
+                            metrics_ref.update_from_service_set(&service_set);
+                        }
+                        Err(err) => {
+                            tracing::warn!(
+                                "Failed to scrape stats for component '{}': {}",
+                                comp_ref.service_name(),
+                                err
+                            );
+                        }
+                    }
+                });
+            }
+            Ok("".to_string())
+        });
+
+        Ok(component_metrics)
+    }
+
     /// TODO
     ///
     /// This method will scrape the stats for all available services
@@ -488,11 +535,13 @@ impl Namespace {
 
     /// Create a [`Component`] in the namespace who's endpoints can be discovered with etcd
     pub fn component(&self, name: impl Into<String>) -> Result<Component> {
-        Ok(ComponentBuilder::from_runtime(self.runtime.clone())
+        let component = ComponentBuilder::from_runtime(self.runtime.clone())
             .name(name)
             .namespace(self.clone())
             .is_static(self.is_static)
-            .build()?)
+            .build()?;
+        component.register_metrics()?; // register a callback to scrape stats and update metrics
+        Ok(component)
     }
 
     /// Create a [`Namespace`] in the parent namespace

lib/runtime/src/component.rs

@@ -116,7 +116,7 @@ impl DistributedRuntime {
         // Register NATS client metrics after creation
         if let Err(e) = distributed_runtime
             .nats_client()
-            .add_metrics(&distributed_runtime)
+            .register_metrics(&distributed_runtime)
         {
             tracing::warn!("Failed to register NATS client metrics: {}", e);
         }

lib/runtime/src/distributed.rs

@@ -900,11 +900,23 @@ mod test_prefixes {
 #[cfg(test)]
 mod test_simple_metricsregistry_trait {
     use super::create_test_drt;
+    use super::prometheus_names::nats as nats_metrics;
     use super::*;
-            use super::prometheus_names::nats as nats_metrics;
     use prometheus::Counter;
     use std::sync::Arc;
 
+    /// Filters out all NATS metrics from Prometheus output for test comparisons.
+    fn filter_out_nats_metrics(input: &str) -> String {
+        input
+            .lines()
+            .filter(|line| {
+                !line.starts_with(&format!("dynamo_component_{}", nats_metrics::PREFIX))
+                    && !line.trim().is_empty()
+            })
+            .collect::<Vec<_>>()
+            .join("\n")
+    }
+
     #[test]
     fn test_prometheusfactory_using_metrics_registry_trait() {
         // Setup real DRT and registry using the test-friendly constructor
@@ -925,15 +937,17 @@ mod test_simple_metricsregistry_trait {
         let epsilon = 0.01;
         assert!((counter.get() - 123.456789).abs() < epsilon);
 
-        let endpoint_output = endpoint.prometheus_metrics_fmt().unwrap();
+        let endpoint_output_raw = endpoint.prometheus_metrics_fmt().unwrap();
         println!("Endpoint output:");
-        println!("{}", endpoint_output);
+        println!("{}", endpoint_output_raw);
+
+        // Filter out NATS service metrics for test comparison
+        let endpoint_output = filter_out_nats_metrics(&endpoint_output_raw);
 
         let expected_endpoint_output = format!(
             r#"# HELP dynamo_component_testcounter A test counter
 # TYPE dynamo_component_testcounter counter
-dynamo_component_testcounter{{dynamo_component="testcomponent",dynamo_endpoint="testendpoint",dynamo_namespace="testnamespace"}} 123.456789
-"#
+dynamo_component_testcounter{{dynamo_component="testcomponent",dynamo_endpoint="testendpoint",dynamo_namespace="testnamespace"}} 123.456789"#
         );
 
         assert_eq!(
@@ -953,18 +967,20 @@ dynamo_component_testcounter{{dynamo_component="testcomponent",dynamo_endpoint="
         assert_eq!(gauge.get(), 50000.0);
 
         // Test Prometheus format output for Component (gauge + histogram)
-        let component_output = component.prometheus_metrics_fmt().unwrap();
+        let component_output_raw = component.prometheus_metrics_fmt().unwrap();
         println!("Component output:");
-        println!("{}", component_output);
+        println!("{}", component_output_raw);
+
+        // Filter out NATS service metrics for test comparison
+        let component_output = filter_out_nats_metrics(&component_output_raw);
 
         let expected_component_output = format!(
             r#"# HELP dynamo_component_testcounter A test counter
 # TYPE dynamo_component_testcounter counter
 dynamo_component_testcounter{{dynamo_component="testcomponent",dynamo_endpoint="testendpoint",dynamo_namespace="testnamespace"}} 123.456789
 # HELP dynamo_component_testgauge A test gauge
 # TYPE dynamo_component_testgauge gauge
-dynamo_component_testgauge{{dynamo_component="testcomponent",dynamo_namespace="testnamespace"}} 50000
-"#
+dynamo_component_testgauge{{dynamo_component="testcomponent",dynamo_namespace="testnamespace"}} 50000"#
         );
 
         assert_eq!(
@@ -983,9 +999,12 @@ dynamo_component_testgauge{{dynamo_component="testcomponent",dynamo_namespace="t
         assert_eq!(intcounter.get(), 12345);
 
         // Test Prometheus format output for Namespace (int_counter + gauge + histogram)
-        let namespace_output = namespace.prometheus_metrics_fmt().unwrap();
+        let namespace_output_raw = namespace.prometheus_metrics_fmt().unwrap();
         println!("Namespace output:");
-        println!("{}", namespace_output);
+        println!("{}", namespace_output_raw);
+
+        // Filter out NATS service metrics for test comparison
+        let namespace_output = filter_out_nats_metrics(&namespace_output_raw);
 
         let expected_namespace_output = format!(
             r#"# HELP dynamo_component_testcounter A test counter
@@ -996,8 +1015,7 @@ dynamo_component_testcounter{{dynamo_component="testcomponent",dynamo_endpoint="
 dynamo_component_testgauge{{dynamo_component="testcomponent",dynamo_namespace="testnamespace"}} 50000
 # HELP dynamo_component_testintcounter A test int counter
 # TYPE dynamo_component_testintcounter counter
-dynamo_component_testintcounter{{dynamo_namespace="testnamespace"}} 12345
-"#
+dynamo_component_testintcounter{{dynamo_namespace="testnamespace"}} 12345"#
         );
 
         assert_eq!(
@@ -1063,16 +1081,7 @@ dynamo_component_testintcounter{{dynamo_namespace="testnamespace"}} 12345
         println!("DRT output:");
         println!("{}", drt_output);
 
-        // Helper function to filter out NATS client metrics
-        fn filter_out_nats_metrics(metrics: &str) -> String {
-            metrics
-                .lines()
-                .filter(|line| !line.contains(nats_metrics::PREFIX) && !line.trim().is_empty())
-                .collect::<Vec<_>>()
-                .join("\n")
-        }
-
-        // Filter out NATS client metrics for comparison
+        // Filter out all NATS metrics for comparison
         let filtered_drt_output = filter_out_nats_metrics(&drt_output);
 
         let expected_drt_output = format!(
@@ -1144,7 +1153,7 @@ dynamo_component_testintgaugevec{{instance="server2",service="api",status="inact
         // Check for specific NATS client metric names (without values)
         let nats_metric_names: Vec<&str> = nats_metrics
             .iter()
-            .filter(|line| line.starts_with(&format!("dynamo_component_{}_", nats_metrics::PREFIX)))
+            .filter(|line| line.starts_with(&format!("dynamo_component_{}", nats_metrics::PREFIX)))
             .map(|line| line.split('{').next().unwrap_or(line))
             .collect();
 

lib/runtime/src/metrics.rs

@@ -13,33 +13,53 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! NATS client Prometheus metrics constants
+//! Prometheus metric name constants
 //!
-//! This module provides centralized Prometheus metric name constants for NATS client metrics
+//! This module provides centralized Prometheus metric name constants for various components
 //! to ensure consistency and avoid duplication across the codebase.
 
-/// NATS client Prometheus metric names
+/// NATS Prometheus metric names
 pub mod nats {
     /// Prefix for all NATS client metrics
-    pub const PREFIX: &str = "nats_client";
+    pub const PREFIX: &str = "nats_";
 
+    /// DistributedRuntime metrics
     /// Total number of bytes received by NATS client
-    pub const IN_BYTES: &str = "nats_client_in_bytes";
+    pub const IN_BYTES: &str = "nats_in_bytes";
 
     /// Total number of bytes sent by NATS client
-    pub const OUT_BYTES: &str = "nats_client_out_bytes";
+    pub const OUT_BYTES: &str = "nats_out_bytes";
 
     /// Total number of messages received by NATS client
-    pub const IN_MESSAGES: &str = "nats_client_in_messages";
+    pub const IN_MESSAGES: &str = "nats_in_messages";
 
     /// Total number of messages sent by NATS client
-    pub const OUT_MESSAGES: &str = "nats_client_out_messages";
+    pub const OUT_MESSAGES: &str = "nats_out_messages";
 
     /// Total number of connections established by NATS client
-    pub const CONNECTS: &str = "nats_client_connects";
+    pub const CONNECTS: &str = "nats_connects";
 
     /// Current connection state of NATS client (0=disconnected, 1=connected, 2=reconnecting)
-    pub const CONNECTION_STATE: &str = "nats_client_connection_state";
+    pub const CONNECTION_STATE: &str = "nats_connection_state";
+
+    // Component metrics (ordered to match NatsStatsMetrics fields)
+    /// Average processing time in milliseconds (maps to: average_processing_time in nanoseconds)
+    pub const AVG_PROCESSING_MS: &str = "nats_avg_processing_time_ms";
+
+    /// Total errors across all endpoints (maps to: num_errors)
+    pub const TOTAL_ERRORS: &str = "nats_total_errors";
+
+    /// Total requests across all endpoints (maps to: num_requests)
+    pub const TOTAL_REQUESTS: &str = "nats_total_requests";
+
+    /// Total processing time in milliseconds (maps to: processing_time in nanoseconds)
+    pub const TOTAL_PROCESSING_MS: &str = "nats_total_processing_time_ms";
+
+    /// Number of active services (derived from ServiceSet.services)
+    pub const ACTIVE_SERVICES: &str = "nats_active_services";
+
+    /// Number of active endpoints (derived from ServiceInfo.endpoints)
+    pub const ACTIVE_ENDPOINTS: &str = "nats_active_endpoints";
 }
 
 /// All NATS client Prometheus metric names as an array for iteration/validation
@@ -52,3 +72,15 @@ pub const ALL_NATS_METRICS: &[&str] = &[
     nats::OUT_BYTES,
     nats::OUT_MESSAGES,
 ];
+
+/// All component service Prometheus metric names as an array for iteration/validation
+/// (ordered to match NatsStatsMetrics fields)
+#[allow(dead_code)]
+pub const ALL_COMPONENT_SERVICE_METRICS: &[&str] = &[
+    nats::AVG_PROCESSING_MS,   // maps to: average_processing_time (nanoseconds)
+    nats::TOTAL_ERRORS,        // maps to: num_errors
+    nats::TOTAL_REQUESTS,      // maps to: num_requests
+    nats::TOTAL_PROCESSING_MS, // maps to: processing_time (nanoseconds)
+    nats::ACTIVE_SERVICES,     // derived from ServiceSet.services
+    nats::ACTIVE_ENDPOINTS,    // derived from ServiceInfo.endpoints
+];

lib/runtime/src/metrics/prometheus_names.rs

@@ -253,6 +253,7 @@ async fn metrics_handler(state: Arc<MetricsServerState>) -> impl IntoResponse {
     // Update the uptime gauge with current value
     state.update_uptime_gauge();
 
+    // Empty prefix means this is the top-level metrics at the DistributedRuntime level
     state.drt().execute_metrics_callbacks("");
 
     // Get all metrics from DistributedRuntime (top-level)
@@ -345,7 +346,7 @@ mod tests {
         // Filter out NATS client metrics for comparison
         let filtered_response: String = response
             .lines()
-            .filter(|line| !line.contains("nats_client"))
+            .filter(|line| !line.contains(crate::metrics::prometheus_names::nats::PREFIX))
             .collect::<Vec<_>>()
             .join("\n");