metrics.rs 53.5 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
    build_component_metric_name, labels, name_prefix, 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
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
mod test_helpers {
    use super::prometheus_names::name_prefix;
    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<_>>()
    }

    /// 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| {
805
            line.starts_with(&format!("{}_", name_prefix::COMPONENT))
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
                && !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))
    }
857
858
}

859
#[cfg(test)]
860
mod test_metricsregistry_units {
861
862
863
    use super::*;

    #[test]
864
865
866
    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");
867
        assert_eq!(result, "dynamo_component_requests");
868

869
        let result = build_component_metric_name("counter");
870
        assert_eq!(result, "dynamo_component_counter");
871
872
    }

873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
    #[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);
902

903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
        // 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!");
    }

926
    #[test]
927
    fn test_metrics_registry_entry_callbacks() {
928
        use crate::MetricsRegistry;
929
        use std::sync::atomic::{AtomicUsize, Ordering};
930

931
932
        // Test 1: Basic callback execution with counter increments
        {
933
            let registry = MetricsRegistry::new();
934
935
936
937
938
            let counter = Arc::new(AtomicUsize::new(0));

            // Add callbacks with different increment values
            for increment in [1, 10, 100] {
                let counter_clone = counter.clone();
939
                registry.add_update_callback(Arc::new(move || {
940
941
942
943
                    counter_clone.fetch_add(increment, Ordering::SeqCst);
                    Ok(())
                }));
            }
944

945
946
            // Verify counter starts at 0
            assert_eq!(counter.load(Ordering::SeqCst), 0);
947

948
            // First execution
949
            let results = registry.execute_update_callbacks();
950
951
952
            assert_eq!(results.len(), 3);
            assert!(results.iter().all(|r| r.is_ok()));
            assert_eq!(counter.load(Ordering::SeqCst), 111); // 1 + 10 + 100
953

954
            // Second execution - callbacks should be reusable
955
            let results = registry.execute_update_callbacks();
956
957
            assert_eq!(results.len(), 3);
            assert_eq!(counter.load(Ordering::SeqCst), 222); // 111 + 111
958

959
960
961
            // Test cloning - cloned entry shares callbacks (callbacks are Arc-wrapped)
            let cloned = registry.clone();
            assert_eq!(cloned.execute_update_callbacks().len(), 3);
962
            assert_eq!(counter.load(Ordering::SeqCst), 333); // 222 + 111
963
964
965
966

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

969
970
        // Test 2: Mixed success and error callbacks
        {
971
            let registry = MetricsRegistry::new();
972
973
974
975
            let counter = Arc::new(AtomicUsize::new(0));

            // Successful callback
            let counter_clone = counter.clone();
976
            registry.add_update_callback(Arc::new(move || {
977
978
979
980
981
                counter_clone.fetch_add(1, Ordering::SeqCst);
                Ok(())
            }));

            // Error callback
982
            registry.add_update_callback(Arc::new(|| Err(anyhow::anyhow!("Simulated error"))));
983
984
985

            // Another successful callback
            let counter_clone = counter.clone();
986
            registry.add_update_callback(Arc::new(move || {
987
988
989
990
991
                counter_clone.fetch_add(10, Ordering::SeqCst);
                Ok(())
            }));

            // Execute and verify mixed results
992
            let results = registry.execute_update_callbacks();
993
994
995
996
997
998
999
1000
1001
1002
            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"
            );
1003

1004
1005
            // Verify successful callbacks still executed
            assert_eq!(counter.load(Ordering::SeqCst), 11); // 1 + 10
1006

1007
            // Execute again - errors should be consistent
1008
            let results = registry.execute_update_callbacks();
1009
1010
1011
            assert!(results[1].is_err());
            assert_eq!(counter.load(Ordering::SeqCst), 22); // 11 + 11
        }
1012

1013
1014
        // Test 3: Empty registry
        {
1015
1016
            let registry = MetricsRegistry::new();
            let results = registry.execute_update_callbacks();
1017
1018
1019
1020
            assert_eq!(results.len(), 0);
        }
    }
}
1021

1022
1023
1024
1025
#[cfg(feature = "integration")]
#[cfg(test)]
mod test_metricsregistry_prefixes {
    use super::*;
1026
    use crate::distributed::distributed_test_utils::create_test_drt_async;
1027
1028
    use prometheus::core::Collector;

1029
1030
    #[tokio::test]
    async fn test_hierarchical_prefixes_and_parent_hierarchies() {
1031
        let drt = create_test_drt_async().await;
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042

        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);
1043
1044
        assert_eq!(drt.parent_hierarchies().len(), 0);
        // DRT hierarchy is just its basename (empty string)
1045
1046
1047

        // Namespace
        assert_eq!(namespace.basename(), NAMESPACE_NAME);
1048
1049
1050
        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
1051
1052
1053

        // Component
        assert_eq!(component.basename(), COMPONENT_NAME);
1054
1055
1056
1057
        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
1058

1059
1060
        // Endpoint
        assert_eq!(endpoint.basename(), ENDPOINT_NAME);
1061
1062
1063
1064
1065
        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
1066

1067
        // Relationships
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
        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())
        );
1086

1087
        // Depth
1088
1089
1090
1091
        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);
1092

1093
1094
1095
        // 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.
1096
        let invalid_namespace = drt.namespace("@@123").unwrap();
1097
1098
1099
1100
        let result =
            invalid_namespace
                .metrics()
                .create_counter("test_counter", "A test counter", &[]);
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
        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");
1111
        }
1112

1113
1114
        // Valid namespace works
        let valid_namespace = drt.namespace("ns567").unwrap();
1115
1116
        assert!(
            valid_namespace
1117
                .metrics()
1118
1119
1120
                .create_counter("test_counter", "A test counter", &[])
                .is_ok()
        );
1121
    }
1122

1123
1124
    #[tokio::test]
    async fn test_recursive_namespace() {
1125
        // Create a distributed runtime for testing
1126
        let drt = create_test_drt_async().await;
1127

1128
1129
1130
1131
        // 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();
1132

1133
1134
        // Create a component in the deepest namespace
        let component = ns3.component("test-component").unwrap();
1135

1136
1137
        // Verify the hierarchy structure
        assert_eq!(ns1.basename(), "ns1");
1138
1139
1140
        assert_eq!(ns1.parent_hierarchies().len(), 1);
        assert_eq!(ns1.parent_hierarchies()[0].basename(), "");
        // ns1 hierarchy is just its basename since parent is empty
1141
1142

        assert_eq!(ns2.basename(), "ns2");
1143
1144
1145
1146
        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
1147

1148
        assert_eq!(ns3.basename(), "ns3");
1149
1150
1151
1152
1153
        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
1154

1155
        assert_eq!(component.basename(), "test-component");
1156
1157
1158
1159
1160
1161
        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
1162

1163
        println!("✓ Chained namespace test passed - all prefixes correct");
1164
1165
1166
1167
1168
    }
}

#[cfg(feature = "integration")]
#[cfg(test)]
1169
1170
mod test_metricsregistry_prometheus_fmt_outputs {
    use super::prometheus_names::name_prefix;
1171
    use super::*;
1172
    use crate::distributed::distributed_test_utils::create_test_drt_async;
1173
1174
1175
    use prometheus::Counter;
    use std::sync::Arc;

1176
1177
    #[tokio::test]
    async fn test_prometheusfactory_using_metrics_registry_trait() {
1178
        // Setup real DRT and registry using the test-friendly constructor
1179
        let drt = create_test_drt_async().await;
1180

1181
        // Use a simple constant namespace name
1182
        let namespace_name = "ns345";
1183

1184
        let namespace = drt.namespace(namespace_name).unwrap();
1185
1186
        let component = namespace.component("comp345").unwrap();
        let endpoint = component.endpoint("ep345");
1187
1188
1189

        // Test Counter creation
        let counter = endpoint
1190
            .metrics()
1191
            .create_counter("testcounter", "A test counter", &[])
1192
1193
1194
1195
1196
            .unwrap();
        counter.inc_by(123.456789);
        let epsilon = 0.01;
        assert!((counter.get() - 123.456789).abs() < epsilon);

1197
        let endpoint_output_raw = endpoint.metrics().prometheus_expfmt().unwrap();
1198
        println!("Endpoint output:");
1199
1200
        println!("{}", endpoint_output_raw);

1201
        let expected_endpoint_output = r#"# HELP dynamo_component_testcounter A test counter
1202
# TYPE dynamo_component_testcounter counter
1203
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789"#.to_string();
1204
1205

        assert_eq!(
1206
            endpoint_output_raw, expected_endpoint_output,
1207
1208
            "\n=== ENDPOINT COMPARISON FAILED ===\n\
             Actual:\n{}\n\
1209
             Expected:\n{}\n\
1210
             ==============================",
1211
            endpoint_output_raw, expected_endpoint_output
1212
1213
1214
1215
        );

        // Test Gauge creation
        let gauge = component
1216
            .metrics()
1217
            .create_gauge("testgauge", "A test gauge", &[])
1218
1219
1220
1221
1222
            .unwrap();
        gauge.set(50000.0);
        assert_eq!(gauge.get(), 50000.0);

        // Test Prometheus format output for Component (gauge + histogram)
1223
        let component_output_raw = component.metrics().prometheus_expfmt().unwrap();
1224
        println!("Component output:");
1225
1226
        println!("{}", component_output_raw);

1227
        let expected_component_output = r#"# HELP dynamo_component_testcounter A test counter
1228
# TYPE dynamo_component_testcounter counter
1229
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1230
1231
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1232
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000"#.to_string();
1233
1234

        assert_eq!(
1235
            component_output_raw, expected_component_output,
1236
1237
            "\n=== COMPONENT COMPARISON FAILED ===\n\
             Actual:\n{}\n\
1238
             Expected:\n{}\n\
1239
             ==============================",
1240
            component_output_raw, expected_component_output
1241
1242
1243
        );

        let intcounter = namespace
1244
            .metrics()
1245
            .create_intcounter("testintcounter", "A test int counter", &[])
1246
1247
1248
1249
1250
            .unwrap();
        intcounter.inc_by(12345);
        assert_eq!(intcounter.get(), 12345);

        // Test Prometheus format output for Namespace (int_counter + gauge + histogram)
1251
        let namespace_output_raw = namespace.metrics().prometheus_expfmt().unwrap();
1252
        println!("Namespace output:");
1253
1254
        println!("{}", namespace_output_raw);

1255
        let expected_namespace_output = r#"# HELP dynamo_component_testcounter A test counter
1256
# TYPE dynamo_component_testcounter counter
1257
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1258
1259
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1260
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000
1261
1262
# HELP dynamo_component_testintcounter A test int counter
# TYPE dynamo_component_testintcounter counter
1263
dynamo_component_testintcounter{dynamo_namespace="ns345"} 12345"#.to_string();
1264
1265

        assert_eq!(
1266
            namespace_output_raw, expected_namespace_output,
1267
1268
            "\n=== NAMESPACE COMPARISON FAILED ===\n\
             Actual:\n{}\n\
1269
             Expected:\n{}\n\
1270
             ==============================",
1271
            namespace_output_raw, expected_namespace_output
1272
1273
1274
        );

        // Test IntGauge creation
1275
        let intgauge = namespace
1276
            .metrics()
1277
            .create_intgauge("testintgauge", "A test int gauge", &[])
1278
1279
1280
1281
1282
            .unwrap();
        intgauge.set(42);
        assert_eq!(intgauge.get(), 42);

        // Test IntGaugeVec creation
1283
        let intgaugevec = namespace
1284
            .metrics()
1285
            .create_intgaugevec(
1286
                "testintgaugevec",
1287
                "A test int gauge vector",
1288
                &["instance", "status"],
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
                &[("service", "api")],
            )
            .unwrap();
        intgaugevec
            .with_label_values(&["server1", "active"])
            .set(10);
        intgaugevec
            .with_label_values(&["server2", "inactive"])
            .set(0);

1299
1300
        // Test CounterVec creation
        let countervec = endpoint
1301
            .metrics()
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
            .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
1314
            .metrics()
1315
1316
1317
1318
1319
1320
1321
            .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)
1322
        let drt_output_raw = drt.metrics().prometheus_expfmt().unwrap();
1323
        println!("DRT output:");
1324
1325
        println!("{}", drt_output_raw);

1326
        let expected_drt_output = r#"# HELP dynamo_component_testcounter A test counter
1327
# TYPE dynamo_component_testcounter counter
1328
dynamo_component_testcounter{dynamo_component="comp345",dynamo_endpoint="ep345",dynamo_namespace="ns345"} 123.456789
1329
1330
# HELP dynamo_component_testcountervec A test counter vector
# TYPE dynamo_component_testcountervec counter
1331
1332
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
1333
1334
# HELP dynamo_component_testgauge A test gauge
# TYPE dynamo_component_testgauge gauge
1335
dynamo_component_testgauge{dynamo_component="comp345",dynamo_namespace="ns345"} 50000
1336
1337
# HELP dynamo_component_testhistogram A test histogram
# TYPE dynamo_component_testhistogram histogram
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
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
1352
1353
# HELP dynamo_component_testintcounter A test int counter
# TYPE dynamo_component_testintcounter counter
1354
dynamo_component_testintcounter{dynamo_namespace="ns345"} 12345
1355
1356
# HELP dynamo_component_testintgauge A test int gauge
# TYPE dynamo_component_testintgauge gauge
1357
dynamo_component_testintgauge{dynamo_namespace="ns345"} 42
1358
1359
# HELP dynamo_component_testintgaugevec A test int gauge vector
# TYPE dynamo_component_testintgaugevec gauge
1360
dynamo_component_testintgaugevec{dynamo_namespace="ns345",instance="server1",service="api",status="active"} 10
1361
1362
1363
1364
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();
1365
1366

        assert_eq!(
1367
            drt_output_raw, expected_drt_output,
1368
1369
            "\n=== DRT COMPARISON FAILED ===\n\
             Expected:\n{}\n\
1370
             Actual (filtered):\n{}\n\
1371
             ==============================",
1372
            expected_drt_output, drt_output_raw
1373
1374
1375
1376
        );

        println!("✓ All Prometheus format outputs verified successfully!");
    }
1377
1378
1379

    #[test]
    fn test_refactored_filter_functions() {
1380
        // Test data with component metrics
1381
1382
1383
1384
1385
1386
1387
        let test_input = r#"# HELP dynamo_component_requests Total requests
# TYPE dynamo_component_requests counter
dynamo_component_requests 42
# 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
1388
dynamo_component_errors_total 5"#;
1389
1390
1391

        // Test extract_metrics (only actual metric lines, excluding help/type)
        let metrics_only = super::test_helpers::extract_metrics(test_input);
1392
        assert_eq!(metrics_only.len(), 4); // 4 actual metric lines (excluding help/type)
1393
1394
1395
1396
1397
        assert!(
            metrics_only
                .iter()
                .all(|line| line.starts_with("dynamo_component") && !line.starts_with("#"))
        );
1398
1399
1400
1401

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