metrics.rs 74.9 KB
Newer Older
1
2
3
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

4
//! Metrics registry trait and implementation for Prometheus metrics
5
//!
6
7
//! This module provides a trait-based interface for creating and managing Prometheus metrics
//! with automatic label injection and hierarchical naming support.
8

9
10
pub mod prometheus_names;

11
use parking_lot::Mutex;
12
13
14
15
16
use std::collections::HashSet;
use std::sync::Arc;

use crate::component::ComponentBuilder;
use anyhow;
17
18
use once_cell::sync::Lazy;
use regex::Regex;
19
20
21
use std::any::Any;
use std::collections::HashMap;

22
23
// Import commonly used items to avoid verbose prefixes
use prometheus_names::{
24
25
    COMPONENT_NATS_METRICS, DRT_NATS_METRICS, build_component_metric_name, labels, name_prefix,
    nats_client, nats_service, sanitize_prometheus_label, sanitize_prometheus_name, work_handler,
26
27
28
29
};

// Pipeline imports for endpoint creation
use crate::pipeline::{
30
31
    AsyncEngine, AsyncEngineContextProvider, Error, ManyOut, ResponseStream, SingleIn, async_trait,
    network::Ingress,
32
33
34
35
36
37
};
use crate::protocols::annotated::Annotated;
use crate::stream;
use crate::stream::StreamExt;

// If set to true, then metrics will be labeled with the namespace, component, and endpoint labels.
38
// These labels are prefixed with "dynamo_" to avoid collisions with Kubernetes and other monitoring system labels.
39
40
41
42
43
pub const USE_AUTO_LABELS: bool = true;

// Prometheus imports
use prometheus::Encoder;

44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
/// Validate that a label slice has no duplicate keys.
/// Returns Ok(()) when all keys are unique; otherwise returns an error naming the duplicate key.
fn validate_no_duplicate_label_keys(labels: &[(&str, &str)]) -> anyhow::Result<()> {
    let mut seen_keys = std::collections::HashSet::new();
    for (key, _) in labels {
        if !seen_keys.insert(*key) {
            return Err(anyhow::anyhow!(
                "Duplicate label key '{}' found in labels",
                key
            ));
        }
    }
    Ok(())
}

59
60
61
/// ==============================
/// Prometheus section
/// ==============================
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
/// Trait that defines common behavior for Prometheus metric types
pub trait PrometheusMetric: prometheus::core::Collector + Clone + Send + Sync + 'static {
    /// Create a new metric with the given options
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error>
    where
        Self: Sized;

    /// Create a new metric with histogram options and custom buckets
    /// This is a default implementation that will panic for non-histogram metrics
    fn with_histogram_opts_and_buckets(
        _opts: prometheus::HistogramOpts,
        _buckets: Option<Vec<f64>>,
    ) -> Result<Self, prometheus::Error>
    where
        Self: Sized,
    {
        panic!("with_histogram_opts_and_buckets is not implemented for this metric type");
    }

    /// Create a new metric with counter options and label names (for CounterVec)
    /// This is a default implementation that will panic for non-countervec metrics
    fn with_opts_and_label_names(
        _opts: prometheus::Opts,
        _label_names: &[&str],
    ) -> Result<Self, prometheus::Error>
    where
        Self: Sized,
    {
        panic!("with_opts_and_label_names is not implemented for this metric type");
    }
}

// Implement the trait for Counter, IntCounter, and Gauge
impl PrometheusMetric for prometheus::Counter {
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        prometheus::Counter::with_opts(opts)
    }
}

impl PrometheusMetric for prometheus::IntCounter {
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        prometheus::IntCounter::with_opts(opts)
    }
}

impl PrometheusMetric for prometheus::Gauge {
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        prometheus::Gauge::with_opts(opts)
    }
}

impl PrometheusMetric for prometheus::IntGauge {
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        prometheus::IntGauge::with_opts(opts)
    }
}

119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
impl PrometheusMetric for prometheus::GaugeVec {
    fn with_opts(_opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        Err(prometheus::Error::Msg(
            "GaugeVec requires label names, use with_opts_and_label_names instead".to_string(),
        ))
    }

    fn with_opts_and_label_names(
        opts: prometheus::Opts,
        label_names: &[&str],
    ) -> Result<Self, prometheus::Error> {
        prometheus::GaugeVec::new(opts, label_names)
    }
}

134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
impl PrometheusMetric for prometheus::IntGaugeVec {
    fn with_opts(_opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        Err(prometheus::Error::Msg(
            "IntGaugeVec requires label names, use with_opts_and_label_names instead".to_string(),
        ))
    }

    fn with_opts_and_label_names(
        opts: prometheus::Opts,
        label_names: &[&str],
    ) -> Result<Self, prometheus::Error> {
        prometheus::IntGaugeVec::new(opts, label_names)
    }
}

149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
impl PrometheusMetric for prometheus::IntCounterVec {
    fn with_opts(_opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        Err(prometheus::Error::Msg(
            "IntCounterVec requires label names, use with_opts_and_label_names instead".to_string(),
        ))
    }

    fn with_opts_and_label_names(
        opts: prometheus::Opts,
        label_names: &[&str],
    ) -> Result<Self, prometheus::Error> {
        prometheus::IntCounterVec::new(opts, label_names)
    }
}

164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
// Implement the trait for Histogram
impl PrometheusMetric for prometheus::Histogram {
    fn with_opts(opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        // Convert Opts to HistogramOpts
        let histogram_opts = prometheus::HistogramOpts::new(opts.name, opts.help);
        prometheus::Histogram::with_opts(histogram_opts)
    }

    fn with_histogram_opts_and_buckets(
        mut opts: prometheus::HistogramOpts,
        buckets: Option<Vec<f64>>,
    ) -> Result<Self, prometheus::Error> {
        if let Some(custom_buckets) = buckets {
            opts = opts.buckets(custom_buckets);
        }
        prometheus::Histogram::with_opts(opts)
    }
}

// Implement the trait for CounterVec
impl PrometheusMetric for prometheus::CounterVec {
    fn with_opts(_opts: prometheus::Opts) -> Result<Self, prometheus::Error> {
        // This will panic - CounterVec needs label names
        panic!("CounterVec requires label names, use with_opts_and_label_names instead");
    }

    fn with_opts_and_label_names(
        opts: prometheus::Opts,
        label_names: &[&str],
    ) -> Result<Self, prometheus::Error> {
        prometheus::CounterVec::new(opts, label_names)
    }
}

198
199
200
201
202
203
/// ==============================
/// Metrics section
/// ==============================
/// Public helper function to create metrics - accessible for Python bindings
pub fn create_metric<T: PrometheusMetric, H: MetricsHierarchy + ?Sized>(
    hierarchy: &H,
204
205
206
207
208
    metric_name: &str,
    metric_desc: &str,
    labels: &[(&str, &str)],
    buckets: Option<Vec<f64>>,
    const_labels: Option<&[&str]>,
209
) -> anyhow::Result<T> {
210
    // Validate that user-provided labels don't have duplicate keys
211
    validate_no_duplicate_label_keys(labels)?;
212
    // Note: stored labels functionality has been removed
213

214
215
    let basename = hierarchy.basename();
    let parent_hierarchies = hierarchy.parent_hierarchies();
216

217
218
219
220
    // Build hierarchy path as vector of strings: parent names + [basename]
    let mut hierarchy_names: Vec<String> =
        parent_hierarchies.iter().map(|p| p.basename()).collect();
    hierarchy_names.push(basename.clone());
221

222
    let metric_name = build_component_metric_name(metric_name);
223

224
    // Build updated_labels: auto-labels first, then `labels` + stored labels
225
226
227
228
229
    let mut updated_labels: Vec<(String, String)> = Vec::new();

    if USE_AUTO_LABELS {
        // Validate that user-provided labels don't conflict with auto-generated labels
        for (key, _) in labels {
230
            if *key == labels::NAMESPACE || *key == labels::COMPONENT || *key == labels::ENDPOINT {
231
232
233
234
235
236
237
                return Err(anyhow::anyhow!(
                    "Label '{}' is automatically added by auto_label feature and cannot be manually set",
                    key
                ));
            }
        }

238
        // Add auto-generated labels with sanitized values
239
240
        if hierarchy_names.len() > 1 {
            let namespace = &hierarchy_names[1];
241
            if !namespace.is_empty() {
242
                let valid_namespace = sanitize_prometheus_label(namespace)?;
243
                if !valid_namespace.is_empty() {
244
                    updated_labels.push((labels::NAMESPACE.to_string(), valid_namespace));
245
246
                }
            }
247
        }
248
249
        if hierarchy_names.len() > 2 {
            let component = &hierarchy_names[2];
250
            if !component.is_empty() {
251
                let valid_component = sanitize_prometheus_label(component)?;
252
                if !valid_component.is_empty() {
253
                    updated_labels.push((labels::COMPONENT.to_string(), valid_component));
254
                }
255
256
            }
        }
257
258
        if hierarchy_names.len() > 3 {
            let endpoint = &hierarchy_names[3];
259
            if !endpoint.is_empty() {
260
                let valid_endpoint = sanitize_prometheus_label(endpoint)?;
261
                if !valid_endpoint.is_empty() {
262
                    updated_labels.push((labels::ENDPOINT.to_string(), valid_endpoint));
263
                }
264
265
266
267
268
269
270
271
272
273
            }
        }
    }

    // Add user labels
    updated_labels.extend(
        labels
            .iter()
            .map(|(k, v)| ((*k).to_string(), (*v).to_string())),
    );
274
    // Note: stored labels functionality has been removed
275
276

    // Handle different metric types
277
    let prometheus_metric = if std::any::TypeId::of::<T>()
278
        == std::any::TypeId::of::<prometheus::CounterVec>()
279
    {
280
281
282
283
284
285
286
287
288
289
290
291
292
293
        // Special handling for CounterVec with label names
        // const_labels parameter is required for CounterVec
        if buckets.is_some() {
            return Err(anyhow::anyhow!(
                "buckets parameter is not valid for CounterVec"
            ));
        }
        let mut opts = prometheus::Opts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        let label_names = const_labels
            .ok_or_else(|| anyhow::anyhow!("CounterVec requires const_labels parameter"))?;
        T::with_opts_and_label_names(opts, label_names)?
294
295
296
    } else if std::any::TypeId::of::<T>() == std::any::TypeId::of::<prometheus::GaugeVec>() {
        // Special handling for GaugeVec with label names
        // const_labels parameter is required for GaugeVec
297
298
        if buckets.is_some() {
            return Err(anyhow::anyhow!(
299
                "buckets parameter is not valid for GaugeVec"
300
301
302
303
304
305
306
            ));
        }
        let mut opts = prometheus::Opts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        let label_names = const_labels
307
            .ok_or_else(|| anyhow::anyhow!("GaugeVec requires const_labels parameter"))?;
308
        T::with_opts_and_label_names(opts, label_names)?
309
310
311
312
313
314
315
316
317
318
319
320
321
    } else if std::any::TypeId::of::<T>() == std::any::TypeId::of::<prometheus::Histogram>() {
        // Special handling for Histogram with custom buckets
        // buckets parameter is valid for Histogram, const_labels is not used
        if const_labels.is_some() {
            return Err(anyhow::anyhow!(
                "const_labels parameter is not valid for Histogram"
            ));
        }
        let mut opts = prometheus::HistogramOpts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        T::with_histogram_opts_and_buckets(opts, buckets)?
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
    } else if std::any::TypeId::of::<T>() == std::any::TypeId::of::<prometheus::IntCounterVec>() {
        // Special handling for IntCounterVec with label names
        // const_labels parameter is required for IntCounterVec
        if buckets.is_some() {
            return Err(anyhow::anyhow!(
                "buckets parameter is not valid for IntCounterVec"
            ));
        }
        let mut opts = prometheus::Opts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        let label_names = const_labels
            .ok_or_else(|| anyhow::anyhow!("IntCounterVec requires const_labels parameter"))?;
        T::with_opts_and_label_names(opts, label_names)?
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
    } else if std::any::TypeId::of::<T>() == std::any::TypeId::of::<prometheus::IntGaugeVec>() {
        // Special handling for IntGaugeVec with label names
        // const_labels parameter is required for IntGaugeVec
        if buckets.is_some() {
            return Err(anyhow::anyhow!(
                "buckets parameter is not valid for IntGaugeVec"
            ));
        }
        let mut opts = prometheus::Opts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        let label_names = const_labels
            .ok_or_else(|| anyhow::anyhow!("IntGaugeVec requires const_labels parameter"))?;
        T::with_opts_and_label_names(opts, label_names)?
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
    } else {
        // Standard handling for Counter, IntCounter, Gauge, IntGauge
        // buckets and const_labels parameters are not valid for these types
        if buckets.is_some() {
            return Err(anyhow::anyhow!(
                "buckets parameter is not valid for Counter, IntCounter, Gauge, or IntGauge"
            ));
        }
        if const_labels.is_some() {
            return Err(anyhow::anyhow!(
                "const_labels parameter is not valid for Counter, IntCounter, Gauge, or IntGauge"
            ));
        }
        let mut opts = prometheus::Opts::new(&metric_name, metric_desc);
        for (key, value) in &updated_labels {
            opts = opts.const_label(key.clone(), value.clone());
        }
        T::with_opts(opts)?
    };

372
373
374
    // Register the metric at all hierarchy levels (parents + self)
    // First register at all parent levels
    for parent in parent_hierarchies {
375
        let collector: Box<dyn prometheus::core::Collector> = Box::new(prometheus_metric.clone());
376
        parent.get_metrics_registry().add_metric(collector)?;
377
378
    }

379
380
381
382
    // Then register at this level
    let collector: Box<dyn prometheus::core::Collector> = Box::new(prometheus_metric.clone());
    hierarchy.get_metrics_registry().add_metric(collector)?;

383
    Ok(prometheus_metric)
384
385
}

386
387
388
389
390
/// Wrapper struct that provides access to metrics functionality
/// This struct is accessed via the `.metrics()` method on DistributedRuntime, Namespace, Component, and Endpoint
pub struct Metrics<H: MetricsHierarchy> {
    hierarchy: H,
}
391

392
393
394
impl<H: MetricsHierarchy> Metrics<H> {
    pub fn new(hierarchy: H) -> Self {
        Self { hierarchy }
395
396
397
398
399
400
    }

    // TODO: Add support for additional Prometheus metric types:
    // - Counter: ✅ IMPLEMENTED - create_counter()
    // - CounterVec: ✅ IMPLEMENTED - create_countervec()
    // - Gauge: ✅ IMPLEMENTED - create_gauge()
401
    // - GaugeVec: ✅ IMPLEMENTED - create_gaugevec()
402
    // - GaugeHistogram: create_gauge_histogram() - for gauge histograms
403
404
405
    // - Histogram: ✅ IMPLEMENTED - create_histogram()
    // - HistogramVec with custom buckets: create_histogram_with_buckets()
    // - Info: create_info() - for info metrics with labels
406
407
408
409
    // - IntCounter: ✅ IMPLEMENTED - create_intcounter()
    // - IntCounterVec: ✅ IMPLEMENTED - create_intcountervec()
    // - IntGauge: ✅ IMPLEMENTED - create_intgauge()
    // - IntGaugeVec: ✅ IMPLEMENTED - create_intgaugevec()
410
    // - Stateset: create_stateset() - for state-based metrics
411
412
413
    // - Summary: create_summary() - for quantiles and sum/count metrics
    // - SummaryVec: create_summary_vec() - for labeled summaries
    // - Untyped: create_untyped() - for untyped metrics
414
415
416
    //
    // NOTE: The order of create_* methods below is mirrored in lib/bindings/python/rust/lib.rs::Metrics
    // Keep them synchronized when adding new metric types
417
418

    /// Create a Counter metric
419
    pub fn create_counter(
420
421
422
423
        &self,
        name: &str,
        description: &str,
        labels: &[(&str, &str)],
424
    ) -> anyhow::Result<prometheus::Counter> {
425
        create_metric(&self.hierarchy, name, description, labels, None, None)
426
427
    }

428
    /// Create a CounterVec metric with label names (for dynamic labels)
429
    pub fn create_countervec(
430
431
432
        &self,
        name: &str,
        description: &str,
433
434
        const_labels: &[&str],
        const_label_values: &[(&str, &str)],
435
    ) -> anyhow::Result<prometheus::CounterVec> {
436
        create_metric(
437
            &self.hierarchy,
438
439
440
441
442
443
            name,
            description,
            const_label_values,
            None,
            Some(const_labels),
        )
444
445
    }

446
    /// Create a Gauge metric
447
    pub fn create_gauge(
448
449
450
451
        &self,
        name: &str,
        description: &str,
        labels: &[(&str, &str)],
452
    ) -> anyhow::Result<prometheus::Gauge> {
453
        create_metric(&self.hierarchy, name, description, labels, None, None)
454
455
    }

456
    /// Create a GaugeVec metric with label names (for dynamic labels)
457
    pub fn create_gaugevec(
458
459
460
461
462
463
464
        &self,
        name: &str,
        description: &str,
        const_labels: &[&str],
        const_label_values: &[(&str, &str)],
    ) -> anyhow::Result<prometheus::GaugeVec> {
        create_metric(
465
            &self.hierarchy,
466
467
468
469
470
471
472
473
            name,
            description,
            const_label_values,
            None,
            Some(const_labels),
        )
    }

474
    /// Create a Histogram metric with custom buckets
475
    pub fn create_histogram(
476
477
478
479
480
        &self,
        name: &str,
        description: &str,
        labels: &[(&str, &str)],
        buckets: Option<Vec<f64>>,
481
    ) -> anyhow::Result<prometheus::Histogram> {
482
        create_metric(&self.hierarchy, name, description, labels, buckets, None)
483
484
    }

485
    /// Create an IntCounter metric
486
    pub fn create_intcounter(
487
488
489
490
        &self,
        name: &str,
        description: &str,
        labels: &[(&str, &str)],
491
    ) -> anyhow::Result<prometheus::IntCounter> {
492
        create_metric(&self.hierarchy, name, description, labels, None, None)
493
494
495
    }

    /// Create an IntCounterVec metric with label names (for dynamic labels)
496
    pub fn create_intcountervec(
497
498
499
500
501
        &self,
        name: &str,
        description: &str,
        const_labels: &[&str],
        const_label_values: &[(&str, &str)],
502
    ) -> anyhow::Result<prometheus::IntCounterVec> {
503
        create_metric(
504
            &self.hierarchy,
505
506
507
508
509
510
511
512
513
            name,
            description,
            const_label_values,
            None,
            Some(const_labels),
        )
    }

    /// Create an IntGauge metric
514
    pub fn create_intgauge(
515
516
517
518
        &self,
        name: &str,
        description: &str,
        labels: &[(&str, &str)],
519
    ) -> anyhow::Result<prometheus::IntGauge> {
520
        create_metric(&self.hierarchy, name, description, labels, None, None)
521
522
523
    }

    /// Create an IntGaugeVec metric with label names (for dynamic labels)
524
    pub fn create_intgaugevec(
525
526
527
528
529
        &self,
        name: &str,
        description: &str,
        const_labels: &[&str],
        const_label_values: &[(&str, &str)],
530
    ) -> anyhow::Result<prometheus::IntGaugeVec> {
531
        create_metric(
532
            &self.hierarchy,
533
534
535
536
537
538
539
540
541
            name,
            description,
            const_label_values,
            None,
            Some(const_labels),
        )
    }

    /// Get metrics in Prometheus text format
542
    pub fn prometheus_expfmt(&self) -> anyhow::Result<String> {
543
        // Execute callbacks first to ensure any new metrics are added to the registry
544
        let callback_results = self
545
546
547
            .hierarchy
            .get_metrics_registry()
            .execute_update_callbacks();
548
549
550
551
552
553
554
555

        // Log any callback errors but continue
        for result in callback_results {
            if let Err(e) = result {
                tracing::error!("Error executing metrics callback: {}", e);
            }
        }

556
557
558
559
560
        // Get the Prometheus registry for this hierarchy
        let prometheus_registry = self
            .hierarchy
            .get_metrics_registry()
            .get_prometheus_registry();
561
562

        // Encode metrics from the registry
563
564
565
566
        let metric_families = prometheus_registry.gather();
        let encoder = prometheus::TextEncoder::new();
        let mut buffer = Vec::new();
        encoder.encode(&metric_families, &mut buffer)?;
567
568
        let mut result = String::from_utf8(buffer)?;

569
570
571
572
573
        // Execute and append exposition text callback results
        let expfmt = self
            .hierarchy
            .get_metrics_registry()
            .execute_expfmt_callbacks();
574
575
576
577
578
579
580
581
        if !expfmt.is_empty() {
            if !result.ends_with('\n') {
                result.push('\n');
            }
            result.push_str(&expfmt);
        }

        Ok(result)
582
583
584
    }
}

585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
/// This trait should be implemented by all metric registries, including Prometheus, Envy, OpenTelemetry, and others.
/// It offers a unified interface for creating and managing metrics, organizing sub-registries, and
/// generating output in Prometheus text format.
use crate::traits::DistributedRuntimeProvider;

pub trait MetricsHierarchy: Send + Sync {
    // ========================================================================
    // Required methods - must be implemented by all types
    // ========================================================================

    /// Get the name of this hierarchy (without any hierarchy prefix)
    fn basename(&self) -> String;

    /// Get the parent hierarchies as actual objects (not strings)
    /// Returns a vector of hierarchy references, ordered from root to immediate parent.
    /// For example, an Endpoint would return [DRT, Namespace, Component].
    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy>;

    /// Get a reference to this hierarchy's metrics registry
    fn get_metrics_registry(&self) -> &MetricsRegistry;

    // ========================================================================
    // Provided methods - have default implementations
    // ========================================================================

    /// Access the metrics interface for this hierarchy
    /// This is a provided method that works for any type implementing MetricsHierarchy
    fn metrics(&self) -> Metrics<&Self>
    where
        Self: Sized,
    {
        Metrics::new(self)
    }
}

// Blanket implementation for references to types that implement MetricsHierarchy
impl<T: MetricsHierarchy + ?Sized> MetricsHierarchy for &T {
    fn basename(&self) -> String {
        (**self).basename()
    }

    fn parent_hierarchies(&self) -> Vec<&dyn MetricsHierarchy> {
        (**self).parent_hierarchies()
    }

    fn get_metrics_registry(&self) -> &MetricsRegistry {
        (**self).get_metrics_registry()
    }
}

/// Type alias for runtime callback functions to reduce complexity
///
/// This type represents an Arc-wrapped callback function that can be:
/// - Shared efficiently across multiple threads and contexts
/// - Cloned without duplicating the underlying closure
/// - Used in generic contexts requiring 'static lifetime
///
/// The Arc wrapper is included in the type to make sharing explicit.
pub type PrometheusUpdateCallback = Arc<dyn Fn() -> anyhow::Result<()> + Send + Sync + 'static>;

/// Type alias for exposition text callback functions that return Prometheus text
pub type PrometheusExpositionFormatCallback =
    Arc<dyn Fn() -> anyhow::Result<String> + Send + Sync + 'static>;

649
650
651
652
653
/// Structure to hold Prometheus registries and associated callbacks for a given hierarchy.
///
/// All fields are Arc-wrapped, so cloning shares state. This ensures metrics registered
/// on cloned instances (e.g., cloned Client/Endpoint) are visible to the original.
#[derive(Clone)]
654
pub struct MetricsRegistry {
655
656
657
    /// The Prometheus registry for this hierarchy.
    /// Arc-wrapped so clones share the same registry (metrics registered on clones are visible everywhere).
    pub prometheus_registry: Arc<std::sync::RwLock<prometheus::Registry>>,
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694

    /// Update callbacks invoked before metrics are scraped.
    /// Wrapped in Arc to preserve callbacks across clones (prevents callback loss when MetricsRegistry is cloned).
    pub prometheus_update_callbacks: Arc<std::sync::RwLock<Vec<PrometheusUpdateCallback>>>,

    /// Callbacks that return Prometheus exposition text appended to metrics output.
    /// Wrapped in Arc to preserve callbacks across clones (e.g., vLLM callbacks registered at Endpoint remain accessible at DRT).
    pub prometheus_expfmt_callbacks:
        Arc<std::sync::RwLock<Vec<PrometheusExpositionFormatCallback>>>,
}

impl std::fmt::Debug for MetricsRegistry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MetricsRegistry")
            .field("prometheus_registry", &"<RwLock<Registry>>")
            .field(
                "prometheus_update_callbacks",
                &format!(
                    "<RwLock<Vec<Callback>>> with {} callbacks",
                    self.prometheus_update_callbacks.read().unwrap().len()
                ),
            )
            .field(
                "prometheus_expfmt_callbacks",
                &format!(
                    "<RwLock<Vec<Callback>>> with {} callbacks",
                    self.prometheus_expfmt_callbacks.read().unwrap().len()
                ),
            )
            .finish()
    }
}

impl MetricsRegistry {
    /// Create a new metrics registry with an empty Prometheus registry and callback lists
    pub fn new() -> Self {
        Self {
695
            prometheus_registry: Arc::new(std::sync::RwLock::new(prometheus::Registry::new())),
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
            prometheus_update_callbacks: Arc::new(std::sync::RwLock::new(Vec::new())),
            prometheus_expfmt_callbacks: Arc::new(std::sync::RwLock::new(Vec::new())),
        }
    }

    /// Add a callback function that receives a reference to any MetricsHierarchy
    pub fn add_update_callback(&self, callback: PrometheusUpdateCallback) {
        self.prometheus_update_callbacks
            .write()
            .unwrap()
            .push(callback);
    }

    /// Add an exposition text callback that returns Prometheus text
    pub fn add_expfmt_callback(&self, callback: PrometheusExpositionFormatCallback) {
        self.prometheus_expfmt_callbacks
            .write()
            .unwrap()
            .push(callback);
    }

    /// Execute all update callbacks and return their results
    pub fn execute_update_callbacks(&self) -> Vec<anyhow::Result<()>> {
        self.prometheus_update_callbacks
            .read()
            .unwrap()
            .iter()
            .map(|callback| callback())
            .collect()
    }

    /// Execute all exposition text callbacks and return their concatenated text
    pub fn execute_expfmt_callbacks(&self) -> String {
        let callbacks = self.prometheus_expfmt_callbacks.read().unwrap();
        let mut result = String::new();
        for callback in callbacks.iter() {
            match callback() {
                Ok(text) => {
                    if !text.is_empty() {
                        if !result.is_empty() && !result.ends_with('\n') {
                            result.push('\n');
                        }
                        result.push_str(&text);
                    }
                }
                Err(e) => {
                    tracing::error!("Error executing exposition text callback: {}", e);
                }
            }
        }
        result
    }

    /// Add a Prometheus metric collector to this registry
    pub fn add_metric(
        &self,
        collector: Box<dyn prometheus::core::Collector>,
    ) -> anyhow::Result<()> {
        self.prometheus_registry
            .write()
            .unwrap()
            .register(collector)
            .map_err(|e| anyhow::anyhow!("Failed to register metric: {}", e))
    }

    /// Get a read guard to the Prometheus registry for scraping
    pub fn get_prometheus_registry(&self) -> std::sync::RwLockReadGuard<'_, prometheus::Registry> {
        self.prometheus_registry.read().unwrap()
    }

    /// Returns true if a metric with the given name already exists in the Prometheus registry
    pub fn has_metric_named(&self, metric_name: &str) -> bool {
        self.prometheus_registry
            .read()
            .unwrap()
            .gather()
            .iter()
            .any(|mf| mf.name() == metric_name)
    }
}

impl Default for MetricsRegistry {
    fn default() -> Self {
        Self::new()
    }
}

783
#[cfg(test)]
784
785
mod test_helpers {
    use super::prometheus_names::name_prefix;
786
    use super::prometheus_names::{nats_client, nats_service};
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
    use super::*;

    /// Base function to filter Prometheus output lines based on a predicate.
    /// Returns lines that match the predicate, converted to String.
    fn filter_prometheus_lines<F>(input: &str, mut predicate: F) -> Vec<String>
    where
        F: FnMut(&str) -> bool,
    {
        input
            .lines()
            .filter(|line| predicate(line))
            .map(|line| line.to_string())
            .collect::<Vec<_>>()
    }

    /// Filters out all NATS metrics from Prometheus output for test comparisons.
    pub fn remove_nats_lines(input: &str) -> Vec<String> {
        filter_prometheus_lines(input, |line| {
            !line.contains(&format!(
806
                "{}_{}",
807
                name_prefix::COMPONENT,
808
809
                nats_client::PREFIX
            )) && !line.contains(&format!(
810
                "{}_{}",
811
812
                name_prefix::COMPONENT,
                nats_service::PREFIX
813
814
815
816
817
818
819
820
            )) && !line.trim().is_empty()
        })
    }

    /// Filters to only include NATS metrics from Prometheus output for test comparisons.
    pub fn extract_nats_lines(input: &str) -> Vec<String> {
        filter_prometheus_lines(input, |line| {
            line.contains(&format!(
821
                "{}_{}",
822
                name_prefix::COMPONENT,
823
824
                nats_client::PREFIX
            )) || line.contains(&format!(
825
                "{}_{}",
826
827
                name_prefix::COMPONENT,
                nats_service::PREFIX
828
829
830
831
832
833
834
835
            ))
        })
    }

    /// Extracts all component metrics (excluding help text and type definitions).
    /// Returns only the actual metric lines with values.
    pub fn extract_metrics(input: &str) -> Vec<String> {
        filter_prometheus_lines(input, |line| {
836
            line.starts_with(&format!("{}_", name_prefix::COMPONENT))
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
                && !line.starts_with("#")
                && !line.trim().is_empty()
        })
    }

    /// Parses a Prometheus metric line and extracts the name, labels, and value.
    /// Used instead of fetching metrics directly to test end-to-end results, not intermediate state.
    ///
    /// # Example
    /// ```
    /// let line = "http_requests_total{method=\"GET\"} 1234";
    /// let (name, labels, value) = parse_prometheus_metric(line).unwrap();
    /// assert_eq!(name, "http_requests_total");
    /// assert_eq!(labels.get("method"), Some(&"GET".to_string()));
    /// assert_eq!(value, 1234.0);
    /// ```
    pub fn parse_prometheus_metric(
        line: &str,
    ) -> Option<(String, std::collections::HashMap<String, String>, f64)> {
        if line.trim().is_empty() || line.starts_with('#') {
            return None;
        }

        let parts: Vec<&str> = line.split_whitespace().collect();
        if parts.len() < 2 {
            return None;
        }

        let metric_part = parts[0];
        let value: f64 = parts[1].parse().ok()?;

        let (name, labels) = if metric_part.contains('{') {
            let brace_start = metric_part.find('{').unwrap();
            let brace_end = metric_part.rfind('}').unwrap_or(metric_part.len());
            let name = &metric_part[..brace_start];
            let labels_str = &metric_part[brace_start + 1..brace_end];

            let mut labels = std::collections::HashMap::new();
            for pair in labels_str.split(',') {
                if let Some((k, v)) = pair.split_once('=') {
                    let v = v.trim_matches('"');
                    labels.insert(k.trim().to_string(), v.to_string());
                }
            }
            (name.to_string(), labels)
        } else {
            (metric_part.to_string(), std::collections::HashMap::new())
        };

        Some((name, labels, value))
    }
888
889
}

890
#[cfg(test)]
891
mod test_metricsregistry_units {
892
893
894
    use super::*;

    #[test]
895
896
897
    fn test_build_component_metric_name_with_prefix() {
        // Test that build_component_metric_name correctly prepends the dynamo_component prefix
        let result = build_component_metric_name("requests");
898
        assert_eq!(result, "dynamo_component_requests");
899

900
        let result = build_component_metric_name("counter");
901
        assert_eq!(result, "dynamo_component_counter");
902
903
    }

904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
    #[test]
    fn test_parse_prometheus_metric() {
        use super::test_helpers::parse_prometheus_metric;
        use std::collections::HashMap;

        // Test parsing a metric with labels
        let line = "http_requests_total{method=\"GET\",status=\"200\"} 1234";
        let parsed = parse_prometheus_metric(line);
        assert!(parsed.is_some());

        let (name, labels, value) = parsed.unwrap();
        assert_eq!(name, "http_requests_total");

        let mut expected_labels = HashMap::new();
        expected_labels.insert("method".to_string(), "GET".to_string());
        expected_labels.insert("status".to_string(), "200".to_string());
        assert_eq!(labels, expected_labels);

        assert_eq!(value, 1234.0);

        // Test parsing a metric without labels
        let line = "cpu_usage 98.5";
        let parsed = parse_prometheus_metric(line);
        assert!(parsed.is_some());

        let (name, labels, value) = parsed.unwrap();
        assert_eq!(name, "cpu_usage");
        assert!(labels.is_empty());
        assert_eq!(value, 98.5);
933

934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
        // Test parsing a metric with float value
        let line = "response_time{service=\"api\"} 0.123";
        let parsed = parse_prometheus_metric(line);
        assert!(parsed.is_some());

        let (name, labels, value) = parsed.unwrap();
        assert_eq!(name, "response_time");

        let mut expected_labels = HashMap::new();
        expected_labels.insert("service".to_string(), "api".to_string());
        assert_eq!(labels, expected_labels);

        assert_eq!(value, 0.123);

        // Test parsing invalid lines
        assert!(parse_prometheus_metric("").is_none()); // Empty line
        assert!(parse_prometheus_metric("# HELP metric description").is_none()); // Help text
        assert!(parse_prometheus_metric("# TYPE metric counter").is_none()); // Type definition
        assert!(parse_prometheus_metric("metric_name").is_none()); // No value

        println!("✓ Prometheus metric parsing works correctly!");
    }

957
    #[test]
958
    fn test_metrics_registry_entry_callbacks() {
959
        use crate::MetricsRegistry;
960
        use std::sync::atomic::{AtomicUsize, Ordering};
961

962
963
        // Test 1: Basic callback execution with counter increments
        {
964
            let registry = MetricsRegistry::new();
965
966
967
968
969
            let counter = Arc::new(AtomicUsize::new(0));

            // Add callbacks with different increment values
            for increment in [1, 10, 100] {
                let counter_clone = counter.clone();
970
                registry.add_update_callback(Arc::new(move || {
971
972
973
974
                    counter_clone.fetch_add(increment, Ordering::SeqCst);
                    Ok(())
                }));
            }
975

976
977
            // Verify counter starts at 0
            assert_eq!(counter.load(Ordering::SeqCst), 0);
978

979
            // First execution
980
            let results = registry.execute_update_callbacks();
981
982
983
            assert_eq!(results.len(), 3);
            assert!(results.iter().all(|r| r.is_ok()));
            assert_eq!(counter.load(Ordering::SeqCst), 111); // 1 + 10 + 100
984

985
            // Second execution - callbacks should be reusable
986
            let results = registry.execute_update_callbacks();
987
988
            assert_eq!(results.len(), 3);
            assert_eq!(counter.load(Ordering::SeqCst), 222); // 111 + 111
989

990
991
992
            // Test cloning - cloned entry shares callbacks (callbacks are Arc-wrapped)
            let cloned = registry.clone();
            assert_eq!(cloned.execute_update_callbacks().len(), 3);
993
            assert_eq!(counter.load(Ordering::SeqCst), 333); // 222 + 111
994
995
996
997

            // Original still has callbacks and shares the same Arc
            registry.execute_update_callbacks();
            assert_eq!(counter.load(Ordering::SeqCst), 444); // 333 + 111
998
        }
999

1000
1001
        // Test 2: Mixed success and error callbacks
        {
1002
            let registry = MetricsRegistry::new();
1003
1004
1005
1006
            let counter = Arc::new(AtomicUsize::new(0));

            // Successful callback
            let counter_clone = counter.clone();
1007
            registry.add_update_callback(Arc::new(move || {
1008
1009
1010
1011
1012
                counter_clone.fetch_add(1, Ordering::SeqCst);
                Ok(())
            }));

            // Error callback
1013
            registry.add_update_callback(Arc::new(|| Err(anyhow::anyhow!("Simulated error"))));
1014
1015
1016

            // Another successful callback
            let counter_clone = counter.clone();
1017
            registry.add_update_callback(Arc::new(move || {
1018
1019
1020
1021
1022
                counter_clone.fetch_add(10, Ordering::SeqCst);
                Ok(())
            }));

            // Execute and verify mixed results
1023
            let results = registry.execute_update_callbacks();
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
            assert_eq!(results.len(), 3);
            assert!(results[0].is_ok());
            assert!(results[1].is_err());
            assert!(results[2].is_ok());

            // Verify error message
            assert_eq!(
                results[1].as_ref().unwrap_err().to_string(),
                "Simulated error"
            );
1034

1035
1036
            // Verify successful callbacks still executed
            assert_eq!(counter.load(Ordering::SeqCst), 11); // 1 + 10
1037

1038
            // Execute again - errors should be consistent
1039
            let results = registry.execute_update_callbacks();
1040
1041
1042
            assert!(results[1].is_err());
            assert_eq!(counter.load(Ordering::SeqCst), 22); // 11 + 11
        }
1043

1044
1045
        // Test 3: Empty registry
        {
1046
1047
            let registry = MetricsRegistry::new();
            let results = registry.execute_update_callbacks();
1048
1049
1050
1051
            assert_eq!(results.len(), 0);
        }
    }
}
1052

1053
1054
1055
1056
#[cfg(feature = "integration")]
#[cfg(test)]
mod test_metricsregistry_prefixes {
    use super::*;
1057
    use crate::distributed::distributed_test_utils::create_test_drt_async;
1058
1059
    use prometheus::core::Collector;

1060
1061
    #[tokio::test]
    async fn test_hierarchical_prefixes_and_parent_hierarchies() {
1062
        let drt = create_test_drt_async().await;
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073

        const DRT_NAME: &str = "";
        const NAMESPACE_NAME: &str = "ns901";
        const COMPONENT_NAME: &str = "comp901";
        const ENDPOINT_NAME: &str = "ep901";
        let namespace = drt.namespace(NAMESPACE_NAME).unwrap();
        let component = namespace.component(COMPONENT_NAME).unwrap();
        let endpoint = component.endpoint(ENDPOINT_NAME);

        // DRT
        assert_eq!(drt.basename(), DRT_NAME);
1074
1075
        assert_eq!(drt.parent_hierarchies().len(), 0);
        // DRT hierarchy is just its basename (empty string)
1076
1077
1078

        // Namespace
        assert_eq!(namespace.basename(), NAMESPACE_NAME);
1079
1080
1081
        assert_eq!(namespace.parent_hierarchies().len(), 1);
        assert_eq!(namespace.parent_hierarchies()[0].basename(), DRT_NAME);
        // Namespace hierarchy is just its basename since parent is empty
1082
1083
1084

        // Component
        assert_eq!(component.basename(), COMPONENT_NAME);
1085
1086
1087
1088
        assert_eq!(component.parent_hierarchies().len(), 2);
        assert_eq!(component.parent_hierarchies()[0].basename(), DRT_NAME);
        assert_eq!(component.parent_hierarchies()[1].basename(), NAMESPACE_NAME);
        // Component hierarchy structure is validated by the individual assertions above
1089

1090
1091
        // Endpoint
        assert_eq!(endpoint.basename(), ENDPOINT_NAME);
1092
1093
1094
1095
1096
        assert_eq!(endpoint.parent_hierarchies().len(), 3);
        assert_eq!(endpoint.parent_hierarchies()[0].basename(), DRT_NAME);
        assert_eq!(endpoint.parent_hierarchies()[1].basename(), NAMESPACE_NAME);
        assert_eq!(endpoint.parent_hierarchies()[2].basename(), COMPONENT_NAME);
        // Endpoint hierarchy structure is validated by the individual assertions above
1097

1098
        // Relationships
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
        assert!(
            namespace
                .parent_hierarchies()
                .iter()
                .any(|h| h.basename() == drt.basename())
        );
        assert!(
            component
                .parent_hierarchies()
                .iter()
                .any(|h| h.basename() == namespace.basename())
        );
        assert!(
            endpoint
                .parent_hierarchies()
                .iter()
                .any(|h| h.basename() == component.basename())
        );
1117

1118
        // Depth
1119
1120
1121
1122
        assert_eq!(drt.parent_hierarchies().len(), 0);
        assert_eq!(namespace.parent_hierarchies().len(), 1);
        assert_eq!(component.parent_hierarchies().len(), 2);
        assert_eq!(endpoint.parent_hierarchies().len(), 3);
1123

1124
1125
1126
        // Invalid namespace behavior - sanitizes to "_123" and succeeds
        // @ryanolson intended to enable validation (see TODO comment in component.rs) but didn't turn it on,
        // so invalid characters are sanitized in MetricsRegistry rather than rejected.
1127
        let invalid_namespace = drt.namespace("@@123").unwrap();
1128
1129
1130
1131
        let result =
            invalid_namespace
                .metrics()
                .create_counter("test_counter", "A test counter", &[]);
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
        assert!(result.is_ok());
        if let Ok(counter) = &result {
            // Verify the namespace was sanitized to "_123" in the label
            let desc = counter.desc();
            let namespace_label = desc[0]
                .const_label_pairs
                .iter()
                .find(|l| l.name() == "dynamo_namespace")
                .expect("Should have dynamo_namespace label");
            assert_eq!(namespace_label.value(), "_123");
1142
        }
1143

1144
1145
        // Valid namespace works
        let valid_namespace = drt.namespace("ns567").unwrap();
1146
1147
        assert!(
            valid_namespace
1148
                .metrics()
1149
1150
1151
                .create_counter("test_counter", "A test counter", &[])
                .is_ok()
        );
1152
    }
1153

1154
1155
    #[tokio::test]
    async fn test_recursive_namespace() {
1156
        // Create a distributed runtime for testing
1157
        let drt = create_test_drt_async().await;
1158

1159
1160
1161
1162
        // Create a deeply chained namespace: ns1.ns2.ns3
        let ns1 = drt.namespace("ns1").unwrap();
        let ns2 = ns1.namespace("ns2").unwrap();
        let ns3 = ns2.namespace("ns3").unwrap();
1163

1164
1165
        // Create a component in the deepest namespace
        let component = ns3.component("test-component").unwrap();
1166

1167
1168
        // Verify the hierarchy structure
        assert_eq!(ns1.basename(), "ns1");
1169
1170
1171
        assert_eq!(ns1.parent_hierarchies().len(), 1);
        assert_eq!(ns1.parent_hierarchies()[0].basename(), "");
        // ns1 hierarchy is just its basename since parent is empty
1172
1173

        assert_eq!(ns2.basename(), "ns2");
1174
1175
1176
1177
        assert_eq!(ns2.parent_hierarchies().len(), 2);
        assert_eq!(ns2.parent_hierarchies()[0].basename(), "");
        assert_eq!(ns2.parent_hierarchies()[1].basename(), "ns1");
        // ns2 hierarchy structure validated by parent assertions above
1178

1179
        assert_eq!(ns3.basename(), "ns3");
1180
1181
1182
1183
1184
        assert_eq!(ns3.parent_hierarchies().len(), 3);
        assert_eq!(ns3.parent_hierarchies()[0].basename(), "");
        assert_eq!(ns3.parent_hierarchies()[1].basename(), "ns1");
        assert_eq!(ns3.parent_hierarchies()[2].basename(), "ns2");
        // ns3 hierarchy structure validated by parent assertions above
1185

1186
        assert_eq!(component.basename(), "test-component");
1187
1188
1189
1190
1191
1192
        assert_eq!(component.parent_hierarchies().len(), 4);
        assert_eq!(component.parent_hierarchies()[0].basename(), "");
        assert_eq!(component.parent_hierarchies()[1].basename(), "ns1");
        assert_eq!(component.parent_hierarchies()[2].basename(), "ns2");
        assert_eq!(component.parent_hierarchies()[3].basename(), "ns3");
        // component hierarchy structure validated by parent assertions above
1193

1194
        println!("✓ Chained namespace test passed - all prefixes correct");
1195
1196
1197
1198
1199
    }
}

#[cfg(feature = "integration")]
#[cfg(test)]
1200
1201
1202
mod test_metricsregistry_prometheus_fmt_outputs {
    use super::prometheus_names::name_prefix;
    use super::prometheus_names::{COMPONENT_NATS_METRICS, DRT_NATS_METRICS};
1203
    use super::prometheus_names::{nats_client, nats_service};
1204
    use super::*;
1205
    use crate::distributed::distributed_test_utils::create_test_drt_async;
1206
1207
1208
    use prometheus::Counter;
    use std::sync::Arc;

1209
1210
    #[tokio::test]
    async fn test_prometheusfactory_using_metrics_registry_trait() {
1211
        // Setup real DRT and registry using the test-friendly constructor
1212
        let drt = create_test_drt_async().await;
1213

1214
        // Use a simple constant namespace name
1215
        let namespace_name = "ns345";
1216

1217
        let namespace = drt.namespace(namespace_name).unwrap();
1218
1219
        let component = namespace.component("comp345").unwrap();
        let endpoint = component.endpoint("ep345");
1220
1221
1222

        // Test Counter creation
        let counter = endpoint
1223
            .metrics()
1224
            .create_counter("testcounter", "A test counter", &[])
1225
1226
1227
1228
1229
            .unwrap();
        counter.inc_by(123.456789);
        let epsilon = 0.01;
        assert!((counter.get() - 123.456789).abs() < epsilon);

1230
        let endpoint_output_raw = endpoint.metrics().prometheus_expfmt().unwrap();
1231
        println!("Endpoint output:");
1232
1233
1234
1235
1236
        println!("{}", endpoint_output_raw);

        // Filter out NATS service metrics for test comparison
        let endpoint_output =
            super::test_helpers::remove_nats_lines(&endpoint_output_raw).join("\n");
1237

1238
        let expected_endpoint_output = r#"# HELP dynamo_component_testcounter A test counter
1239
# TYPE dynamo_component_testcounter counter
1240
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789"#.to_string();
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252

        assert_eq!(
            endpoint_output, expected_endpoint_output,
            "\n=== ENDPOINT COMPARISON FAILED ===\n\
             Expected:\n{}\n\
             Actual:\n{}\n\
             ==============================",
            expected_endpoint_output, endpoint_output
        );

        // Test Gauge creation
        let gauge = component
1253
            .metrics()
1254
            .create_gauge("testgauge", "A test gauge", &[])
1255
1256
1257
1258
1259
            .unwrap();
        gauge.set(50000.0);
        assert_eq!(gauge.get(), 50000.0);

        // Test Prometheus format output for Component (gauge + histogram)
1260
        let component_output_raw = component.metrics().prometheus_expfmt().unwrap();
1261
        println!("Component output:");
1262
1263
1264
1265
1266
        println!("{}", component_output_raw);

        // Filter out NATS service metrics for test comparison
        let component_output =
            super::test_helpers::remove_nats_lines(&component_output_raw).join("\n");
1267

1268
        let expected_component_output = r#"# HELP dynamo_component_testcounter A test counter
1269
# TYPE dynamo_component_testcounter counter
1270
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1271
1272
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1273
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000"#.to_string();
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284

        assert_eq!(
            component_output, expected_component_output,
            "\n=== COMPONENT COMPARISON FAILED ===\n\
             Expected:\n{}\n\
             Actual:\n{}\n\
             ==============================",
            expected_component_output, component_output
        );

        let intcounter = namespace
1285
            .metrics()
1286
            .create_intcounter("testintcounter", "A test int counter", &[])
1287
1288
1289
1290
1291
            .unwrap();
        intcounter.inc_by(12345);
        assert_eq!(intcounter.get(), 12345);

        // Test Prometheus format output for Namespace (int_counter + gauge + histogram)
1292
        let namespace_output_raw = namespace.metrics().prometheus_expfmt().unwrap();
1293
        println!("Namespace output:");
1294
1295
1296
1297
1298
        println!("{}", namespace_output_raw);

        // Filter out NATS service metrics for test comparison
        let namespace_output =
            super::test_helpers::remove_nats_lines(&namespace_output_raw).join("\n");
1299

1300
        let expected_namespace_output = r#"# HELP dynamo_component_testcounter A test counter
1301
# TYPE dynamo_component_testcounter counter
1302
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1303
1304
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1305
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000
1306
1307
# HELP dynamo_component_testintcounter A test int counter
# TYPE dynamo_component_testintcounter counter
1308
dynamo_component_testintcounter{dynamo_namespace="ns345"} 12345"#.to_string();
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319

        assert_eq!(
            namespace_output, expected_namespace_output,
            "\n=== NAMESPACE COMPARISON FAILED ===\n\
             Expected:\n{}\n\
             Actual:\n{}\n\
             ==============================",
            expected_namespace_output, namespace_output
        );

        // Test IntGauge creation
1320
        let intgauge = namespace
1321
            .metrics()
1322
            .create_intgauge("testintgauge", "A test int gauge", &[])
1323
1324
1325
1326
1327
            .unwrap();
        intgauge.set(42);
        assert_eq!(intgauge.get(), 42);

        // Test IntGaugeVec creation
1328
        let intgaugevec = namespace
1329
            .metrics()
1330
            .create_intgaugevec(
1331
                "testintgaugevec",
1332
                "A test int gauge vector",
1333
                &["instance", "status"],
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
                &[("service", "api")],
            )
            .unwrap();
        intgaugevec
            .with_label_values(&["server1", "active"])
            .set(10);
        intgaugevec
            .with_label_values(&["server2", "inactive"])
            .set(0);

1344
1345
        // Test CounterVec creation
        let countervec = endpoint
1346
            .metrics()
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
            .create_countervec(
                "testcountervec",
                "A test counter vector",
                &["method", "status"],
                &[("service", "api")],
            )
            .unwrap();
        countervec.with_label_values(&["GET", "200"]).inc_by(10.0);
        countervec.with_label_values(&["POST", "201"]).inc_by(5.0);

        // Test Histogram creation
        let histogram = component
1359
            .metrics()
1360
1361
1362
1363
1364
1365
1366
            .create_histogram("testhistogram", "A test histogram", &[], None)
            .unwrap();
        histogram.observe(1.0);
        histogram.observe(2.5);
        histogram.observe(4.0);

        // Test Prometheus format output for DRT (all metrics combined)
1367
        let drt_output_raw = drt.metrics().prometheus_expfmt().unwrap();
1368
        println!("DRT output:");
1369
1370
1371
1372
1373
        println!("{}", drt_output_raw);

        // Filter out all NATS metrics for comparison
        let filtered_drt_output =
            super::test_helpers::remove_nats_lines(&drt_output_raw).join("\n");
1374

1375
        let expected_drt_output = r#"# HELP dynamo_component_testcounter A test counter
1376
# TYPE dynamo_component_testcounter counter
1377
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1378
1379
# HELP dynamo_component_testcountervec A test counter vector
# TYPE dynamo_component_testcountervec counter
1380
1381
dynamo_component_testcountervec{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345",method="GET",service="api",status="200"} 10
dynamo_component_testcountervec{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345",method="POST",service="api",status="201"} 5
1382
1383
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1384
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000
1385
1386
# HELP dynamo_component_testhistogram A test histogram
# TYPE dynamo_component_testhistogram histogram
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.005"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.01"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.025"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.05"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.1"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.25"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="0.5"} 0
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="1"} 1
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="2.5"} 2
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="5"} 3
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="10"} 3
dynamo_component_testhistogram_bucket{dynamo_component="comp345",dynamo_namespace="ns345",le="+Inf"} 3
dynamo_component_testhistogram_sum{dynamo_component="comp345",dynamo_namespace="ns345"} 7.5
dynamo_component_testhistogram_count{dynamo_component="comp345",dynamo_namespace="ns345"} 3
1401
1402
# HELP dynamo_component_testintcounter A test int counter
# TYPE dynamo_component_testintcounter counter
1403
dynamo_component_testintcounter{dynamo_namespace="ns345"} 12345
1404
1405
# HELP dynamo_component_testintgauge A test int gauge
# TYPE dynamo_component_testintgauge gauge
1406
dynamo_component_testintgauge{dynamo_namespace="ns345"} 42
1407
1408
# HELP dynamo_component_testintgaugevec A test int gauge vector
# TYPE dynamo_component_testintgaugevec gauge
1409
dynamo_component_testintgaugevec{dynamo_namespace="ns345",instance="server1",service="api",status="active"} 10
1410
1411
1412
1413
dynamo_component_testintgaugevec{dynamo_namespace="ns345",instance="server2",service="api",status="inactive"} 0
# HELP dynamo_component_uptime_seconds Total uptime of the DistributedRuntime in seconds
# TYPE dynamo_component_uptime_seconds gauge
dynamo_component_uptime_seconds 0"#.to_string();
1414
1415

        assert_eq!(
1416
            filtered_drt_output, expected_drt_output,
1417
1418
            "\n=== DRT COMPARISON FAILED ===\n\
             Expected:\n{}\n\
1419
             Actual (filtered):\n{}\n\
1420
             ==============================",
1421
            expected_drt_output, filtered_drt_output
1422
1423
1424
1425
        );

        println!("✓ All Prometheus format outputs verified successfully!");
    }
1426
1427
1428
1429
1430
1431
1432

    #[test]
    fn test_refactored_filter_functions() {
        // Test data with mixed content
        let test_input = r#"# HELP dynamo_component_requests Total requests
# TYPE dynamo_component_requests counter
dynamo_component_requests 42
1433
1434
1435
# HELP dynamo_component_nats_client_connection_state Connection state
# TYPE dynamo_component_nats_client_connection_state gauge
dynamo_component_nats_client_connection_state 1
1436
1437
1438
1439
# HELP dynamo_component_latency Response latency
# TYPE dynamo_component_latency histogram
dynamo_component_latency_bucket{le="0.1"} 10
dynamo_component_latency_bucket{le="0.5"} 25
1440
1441
dynamo_component_nats_service_requests_total 100
dynamo_component_nats_service_errors_total 5"#;
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455

        // Test remove_nats_lines (excludes NATS lines but keeps help/type)
        let filtered_out = super::test_helpers::remove_nats_lines(test_input);
        assert_eq!(filtered_out.len(), 7); // 7 non-NATS lines
        assert!(!filtered_out.iter().any(|line| line.contains("nats")));

        // Test extract_nats_lines (includes all NATS lines including help/type)
        let filtered_only = super::test_helpers::extract_nats_lines(test_input);
        assert_eq!(filtered_only.len(), 5); // 5 NATS lines
        assert!(filtered_only.iter().all(|line| line.contains("nats")));

        // Test extract_metrics (only actual metric lines, excluding help/type)
        let metrics_only = super::test_helpers::extract_metrics(test_input);
        assert_eq!(metrics_only.len(), 6); // 6 actual metric lines (excluding help/type)
1456
1457
1458
1459
1460
        assert!(
            metrics_only
                .iter()
                .all(|line| line.starts_with("dynamo_component") && !line.starts_with("#"))
        );
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470

        println!("✓ All refactored filter functions work correctly!");
    }
}

#[cfg(feature = "integration")]
#[cfg(test)]
mod test_metricsregistry_nats {
    use super::prometheus_names::name_prefix;
    use super::prometheus_names::{COMPONENT_NATS_METRICS, DRT_NATS_METRICS};
1471
    use super::prometheus_names::{nats_client, nats_service};
1472
    use super::*;
1473
    use crate::distributed::distributed_test_utils::create_test_drt_async;
1474
1475
    use crate::pipeline::PushRouter;
    use crate::{DistributedRuntime, Runtime};
1476
    use tokio::time::{Duration, sleep};
1477
    #[ignore = "Deprecated - NATS related code to be deleted soon"]
1478
1479
    #[tokio::test]
    async fn test_drt_nats_metrics() {
1480
        // Setup real DRT and registry using the test-friendly constructor
1481
        let drt = create_test_drt_async().await;
1482
1483

        // Get DRT output which should include NATS client metrics
1484
        let drt_output = drt.metrics().prometheus_expfmt().unwrap();
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
        println!("DRT output with NATS metrics:");
        println!("{}", drt_output);

        // Additional checks for NATS client metrics (without checking specific values)
        let drt_nats_metrics = super::test_helpers::extract_nats_lines(&drt_output);

        // Check that NATS client metrics are present
        assert!(
            !drt_nats_metrics.is_empty(),
            "NATS client metrics should be present"
        );

        // Check for specific NATS client metric names (without values)
1498
1499
1500
1501
        // Extract only the metric lines from the already-filtered NATS metrics
        let drt_nats_metric_lines =
            super::test_helpers::extract_metrics(&drt_nats_metrics.join("\n"));
        let actual_drt_nats_metrics_sorted: Vec<&str> = drt_nats_metric_lines
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
            .iter()
            .map(|line| {
                let without_labels = line.split('{').next().unwrap_or(line);
                // Remove the value part (everything after the last space)
                without_labels.split(' ').next().unwrap_or(without_labels)
            })
            .collect();

        let expect_drt_nats_metrics_sorted = {
            let mut temp = DRT_NATS_METRICS
                .iter()
1513
                .map(|metric| build_component_metric_name(metric))
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
                .collect::<Vec<_>>();
            temp.sort();
            temp
        };

        // Print both lists for comparison
        println!(
            "actual_drt_nats_metrics_sorted: {:?}",
            actual_drt_nats_metrics_sorted
        );
        println!(
            "expect_drt_nats_metrics_sorted: {:?}",
            expect_drt_nats_metrics_sorted
        );

        // Compare the sorted lists
        assert_eq!(
1531
            actual_drt_nats_metrics_sorted, expect_drt_nats_metrics_sorted,
1532
1533
1534
1535
1536
1537
            "DRT_NATS_METRICS with prefix and expected_nats_metrics should be identical when sorted"
        );

        println!("✓ DistributedRuntime NATS metrics integration test passed!");
    }

1538
    #[ignore = "Deprecated - NATS related code to be deleted soon"]
1539
1540
    #[tokio::test]
    async fn test_nats_metric_names() {
1541
1542
1543
1544
        // This test only tests the existence of the NATS metrics. It does not check
        // the values of the metrics.

        // Setup real DRT and registry using the test-friendly constructor
1545
        let drt = create_test_drt_async().await;
1546

1547
        // Create a namespace and component from the DRT
1548
        let namespace = drt.namespace("ns789").unwrap();
1549
        let component = namespace.component("comp789").unwrap();
1550

1551
        // Get component output which should include NATS client metrics
1552
        // Additional checks for NATS client metrics (without checking specific values)
1553
1554
1555
        let component_nats_metrics = super::test_helpers::extract_nats_lines(
            &component.metrics().prometheus_expfmt().unwrap(),
        );
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
        println!(
            "Component NATS metrics count: {}",
            component_nats_metrics.len()
        );

        // Check that NATS client metrics are present
        assert!(
            !component_nats_metrics.is_empty(),
            "NATS client metrics should be present"
        );

        // Check for specific NATS client metric names (without values)
        let component_metrics =
1569
            super::test_helpers::extract_metrics(&component.metrics().prometheus_expfmt().unwrap());
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
        let actual_component_nats_metrics_sorted: Vec<&str> = component_metrics
            .iter()
            .map(|line| {
                let without_labels = line.split('{').next().unwrap_or(line);
                // Remove the value part (everything after the last space)
                without_labels.split(' ').next().unwrap_or(without_labels)
            })
            .collect();

        let expect_component_nats_metrics_sorted = {
            let mut temp = COMPONENT_NATS_METRICS
                .iter()
1582
                .map(|metric| build_component_metric_name(metric))
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
                .collect::<Vec<_>>();
            temp.sort();
            temp
        };

        // Print both lists for comparison
        println!(
            "actual_component_nats_metrics_sorted: {:?}",
            actual_component_nats_metrics_sorted
        );
        println!(
            "expect_component_nats_metrics_sorted: {:?}",
            expect_component_nats_metrics_sorted
        );

        // Compare the sorted lists
        assert_eq!(
1600
            actual_component_nats_metrics_sorted, expect_component_nats_metrics_sorted,
1601
1602
1603
            "COMPONENT_NATS_METRICS with prefix and expected_nats_metrics should be identical when sorted"
        );

1604
        // Get both DRT and component output and filter for NATS metrics only
1605
        let drt_output = drt.metrics().prometheus_expfmt().unwrap();
1606
1607
1608
        let drt_nats_lines = super::test_helpers::extract_nats_lines(&drt_output);
        let drt_and_component_nats_metrics =
            super::test_helpers::extract_metrics(&drt_nats_lines.join("\n"));
1609
        println!(
1610
1611
            "DRT and component NATS metrics count: {}",
            drt_and_component_nats_metrics.len()
1612
1613
1614
1615
        );

        // Check that the NATS metrics are present in the component output
        assert_eq!(
1616
            drt_and_component_nats_metrics.len(),
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
            DRT_NATS_METRICS.len() + COMPONENT_NATS_METRICS.len(),
            "DRT at this point should have both the DRT and component NATS metrics"
        );

        // Check that the NATS metrics are present in the component output
        println!("✓ Component NATS metrics integration test passed!");
    }

    /// Tests NATS metrics values before and after endpoint activity with large message processing.
    /// Creates endpoint, sends test messages + 10k byte message, validates metrics (NATS + work handler)
    /// at initial state and post-activity state. Ensures byte thresholds, message counts, and processing
    /// times are within expected ranges. Tests end-to-end client-server communication and metrics collection.
1629
    #[ignore = "Deprecated - NATS related code to be deleted soon"]
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
    #[tokio::test]
    async fn test_nats_metrics_values() -> anyhow::Result<()> {
        struct MessageHandler {}
        impl MessageHandler {
            fn new() -> std::sync::Arc<Self> {
                std::sync::Arc::new(Self {})
            }
        }

        #[async_trait]
        impl AsyncEngine<SingleIn<String>, ManyOut<Annotated<String>>, Error> for MessageHandler {
            async fn generate(
                &self,
                input: SingleIn<String>,
            ) -> Result<ManyOut<Annotated<String>>, Error> {
                let (data, ctx) = input.into_parts();
1646
                let response = data.to_string();
1647
1648
1649
1650
1651
1652
1653
1654
1655
                let stream = stream::iter(vec![Annotated::from_data(response)]);
                Ok(ResponseStream::new(Box::pin(stream), ctx.context()))
            }
        }

        println!("\n=== Initializing DistributedRuntime ===");
        let runtime = Runtime::from_current()?;
        let drt = DistributedRuntime::from_settings(runtime.clone()).await?;
        let namespace = drt.namespace("ns123").unwrap();
1656
        let component = namespace.component("comp123").unwrap();
1657
1658
1659
        let ingress = Ingress::for_engine(MessageHandler::new()).unwrap();

        let _backend_handle = tokio::spawn(async move {
1660
1661
1662
1663
            let endpoint = component
                .endpoint("echo")
                .endpoint_builder()
                .handler(ingress);
1664
1665
1666
1667
1668
1669
            endpoint.start().await.unwrap();
        });

        sleep(Duration::from_millis(500)).await;
        println!("✓ Launched endpoint service in background successfully");

1670
        let drt_output = drt.metrics().prometheus_expfmt().unwrap();
1671
1672
        let parsed_metrics: Vec<_> = drt_output
            .lines()
1673
            .filter_map(super::test_helpers::parse_prometheus_metric)
1674
1675
1676
1677
1678
1679
1680
1681
1682
            .collect();

        println!("=== Initial DRT metrics output ===");
        println!("{}", drt_output);

        println!("\n=== Checking Initial Metric Values ===");

        let initial_expected_metric_values = [
            // DRT NATS metrics (ordered to match DRT_NATS_METRICS)
1683
            (
1684
1685
1686
1687
                build_component_metric_name(nats_client::CONNECTION_STATE),
                1.0,
                1.0,
            ), // Should be connected
1688
1689
1690
1691
1692
            (
                build_component_metric_name(nats_client::CURRENT_CONNECTIONS),
                1.0,
                1.0,
            ), // Should have 1 connection
1693
1694
            (
                build_component_metric_name(nats_client::IN_TOTAL_BYTES),
1695
1696
1697
                800.0,
                4000.0,
            ), // Wide range around observed value of 1888
1698
            (
1699
1700
1701
1702
1703
1704
                build_component_metric_name(nats_client::IN_MESSAGES),
                0.0,
                5.0,
            ), // Wide range around 2
            (
                build_component_metric_name(nats_client::OUT_OVERHEAD_BYTES),
1705
1706
1707
                1500.0,
                5000.0,
            ), // Wide range around observed value of 2752
1708
1709
1710
1711
1712
            (
                build_component_metric_name(nats_client::OUT_MESSAGES),
                0.0,
                5.0,
            ), // Wide range around 2
1713
            // Component NATS metrics (ordered to match COMPONENT_NATS_METRICS)
1714
            (
1715
                build_component_metric_name(nats_service::PROCESSING_MS_AVG),
1716
1717
1718
1719
                0.0,
                0.0,
            ), // No processing yet
            (
1720
                build_component_metric_name(nats_service::ERRORS_TOTAL),
1721
1722
1723
1724
                0.0,
                0.0,
            ), // No errors yet
            (
1725
                build_component_metric_name(nats_service::REQUESTS_TOTAL),
1726
1727
1728
1729
                0.0,
                0.0,
            ), // No requests yet
            (
1730
                build_component_metric_name(nats_service::PROCESSING_MS_TOTAL),
1731
1732
1733
                0.0,
                0.0,
            ), // No processing yet
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
            (
                build_component_metric_name(nats_service::ACTIVE_SERVICES),
                0.0,
                2.0,
            ), // Service may not be fully active yet
            (
                build_component_metric_name(nats_service::ACTIVE_ENDPOINTS),
                0.0,
                2.0,
            ), // Endpoint may not be fully active yet
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
        ];

        for (metric_name, min_value, max_value) in &initial_expected_metric_values {
            let actual_value = parsed_metrics
                .iter()
                .find(|(name, _, _)| name == metric_name)
                .map(|(_, _, value)| *value)
                .unwrap_or_else(|| panic!("Could not find expected metric: {}", metric_name));

            assert!(
                actual_value >= *min_value && actual_value <= *max_value,
                "Initial metric {} should be between {} and {}, but got {}",
                metric_name,
                min_value,
                max_value,
                actual_value
            );
        }

        println!("\n=== Client Runtime to hit the endpoint ===");
        let client_runtime = Runtime::from_current()?;
        let client_distributed = DistributedRuntime::from_settings(client_runtime.clone()).await?;
        let namespace = client_distributed.namespace("ns123")?;
        let component = namespace.component("comp123")?;
        let client = component.endpoint("echo").client().await?;

        client.wait_for_instances().await?;
        println!("✓ Connected to endpoint, waiting for instances...");

        let router =
            PushRouter::<String, Annotated<String>>::from_client(client, Default::default())
                .await?;

        for i in 0..10 {
            let msg = i.to_string().repeat(2000); // 2k bytes message
            let mut stream = router.random(msg.clone().into()).await?;
            while let Some(resp) = stream.next().await {
                // Check if response matches the original message
                if let Some(data) = &resp.data {
                    let is_same = data == &msg;
                    println!(
                        "Response {}: {} bytes, matches original: {}",
                        i,
                        data.len(),
                        is_same
                    );
                }
            }
        }
        println!("✓ Sent messages and received responses successfully");

1795
1796
1797
1798
        println!("\n=== Waiting 500ms for metrics to update ===");
        sleep(Duration::from_millis(500)).await;
        println!("✓ Wait complete, getting final metrics...");

1799
        let final_drt_output = drt.metrics().prometheus_expfmt().unwrap();
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
        println!("\n=== Final Prometheus DRT output ===");
        println!("{}", final_drt_output);

        let final_drt_nats_output = super::test_helpers::extract_nats_lines(&final_drt_output);
        println!("\n=== Filtered NATS metrics from final DRT output ===");
        for line in &final_drt_nats_output {
            println!("{}", line);
        }

        let final_parsed_metrics: Vec<_> = super::test_helpers::extract_metrics(&final_drt_output)
            .iter()
1811
            .filter_map(|line| super::test_helpers::parse_prometheus_metric(line.as_str()))
1812
1813
1814
            .collect();

        let post_expected_metric_values = [
1815
            // DRT NATS metrics
1816
            (
1817
1818
1819
1820
                build_component_metric_name(nats_client::CONNECTION_STATE),
                1.0,
                1.0,
            ), // Connected
1821
1822
1823
1824
1825
            (
                build_component_metric_name(nats_client::CURRENT_CONNECTIONS),
                1.0,
                1.0,
            ), // 1 connection
1826
1827
            (
                build_component_metric_name(nats_client::IN_TOTAL_BYTES),
1828
1829
1830
1831
                20000.0,
                32000.0,
            ), // Wide range around 26117
            (
1832
1833
1834
1835
1836
1837
                build_component_metric_name(nats_client::IN_MESSAGES),
                8.0,
                20.0,
            ), // Wide range around 16
            (
                build_component_metric_name(nats_client::OUT_OVERHEAD_BYTES),
1838
1839
1840
                2500.0,
                8000.0,
            ), // Wide range around 5524
1841
1842
1843
1844
1845
            (
                build_component_metric_name(nats_client::OUT_MESSAGES),
                8.0,
                20.0,
            ), // Wide range around 16
1846
            // Component NATS metrics
1847
            (
1848
                build_component_metric_name(nats_service::PROCESSING_MS_AVG),
1849
1850
1851
1852
                0.0,
                1.0,
            ), // Low processing time
            (
1853
                build_component_metric_name(nats_service::ERRORS_TOTAL),
1854
1855
1856
1857
                0.0,
                0.0,
            ), // No errors
            (
1858
                build_component_metric_name(nats_service::REQUESTS_TOTAL),
1859
                0.0,
1860
1861
                10.0,
            ), // NATS service stats requests (may differ from work handler count)
1862
            (
1863
                build_component_metric_name(nats_service::PROCESSING_MS_TOTAL),
1864
1865
1866
                0.0,
                5.0,
            ), // Low total processing time
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
            (
                build_component_metric_name(nats_service::ACTIVE_SERVICES),
                0.0,
                2.0,
            ), // Service may not be fully active
            (
                build_component_metric_name(nats_service::ACTIVE_ENDPOINTS),
                0.0,
                2.0,
            ), // Endpoint may not be fully active
1877
            // Work handler metrics
1878
            (
1879
1880
1881
1882
1883
1884
                build_component_metric_name(work_handler::REQUESTS_TOTAL),
                10.0,
                10.0,
            ), // 10 messages
            (
                build_component_metric_name(work_handler::REQUEST_BYTES_TOTAL),
1885
1886
                21000.0,
                26000.0,
1887
            ), // ~75-125% of 23520
1888
            (
1889
                build_component_metric_name(work_handler::RESPONSE_BYTES_TOTAL),
1890
1891
                18000.0,
                23000.0,
1892
            ), // ~75-125% of 20660
1893
1894
1895
1896
1897
            (
                build_component_metric_name(work_handler::INFLIGHT_REQUESTS),
                0.0,
                1.0,
            ), // 0 or very low
1898
            // Histograms have _{count,sum} suffixes
1899
1900
1901
            (
                format!(
                    "{}_count",
1902
                    build_component_metric_name(work_handler::REQUEST_DURATION_SECONDS)
1903
1904
1905
                ),
                10.0,
                10.0,
1906
            ), // 10 messages
1907
1908
1909
            (
                format!(
                    "{}_sum",
1910
                    build_component_metric_name(work_handler::REQUEST_DURATION_SECONDS)
1911
                ),
1912
1913
1914
                0.0001,
                1.0,
            ), // Processing time sum (wide range)
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
        ];

        println!("\n=== Checking Post-Activity All Metrics (NATS + Work Handler) ===");
        for (metric_name, min_value, max_value) in &post_expected_metric_values {
            let actual_value = final_parsed_metrics
                .iter()
                .find(|(name, _, _)| name == metric_name)
                .map(|(_, _, value)| *value)
                .unwrap_or_else(|| {
                    panic!(
                        "Could not find expected post-activity metric: {}",
                        metric_name
                    )
                });

            assert!(
                actual_value >= *min_value && actual_value <= *max_value,
                "Post-activity metric {} should be between {} and {}, but got {}",
                metric_name,
                min_value,
                max_value,
                actual_value
            );
            println!(
                "✓ {}: {} (range: {} to {})",
                metric_name, actual_value, min_value, max_value
            );
        }

        println!("✓ All NATS and component metrics parsed successfully!");
        println!("✓ Byte metrics verified to be >= 100 bytes!");
        println!("✓ Post-activity metrics verified with higher thresholds!");
        println!("✓ Work handler metrics reflect increased activity!");

        Ok(())
    }
1951
}