Ensure that exporting metrics doesn't interfere with logging metrics

Signed-off-by: Shivangi Kumar <[email protected]>

Author: Shivangi Kumar

mountpoint-s3-fs/src/metrics.rs

@@ -3,7 +3,8 @@
 //! This module hooks up the [metrics](https://docs.rs/metrics) facade to a metrics sink that
 //! currently just emits them to a tracing log entry.
 
-use crate::metrics_otel::{OtlpConfig, OtlpMetricsExporter};
+pub use crate::metrics_otel::OtlpConfig;
+use crate::metrics_otel::OtlpMetricsExporter;
 use opentelemetry::KeyValue;
 
 use std::thread::{self, JoinHandle};
@@ -17,6 +18,7 @@ use crate::sync::Arc;
 use crate::sync::mpsc::{RecvTimeoutError, Sender, channel};
 
 mod data;
+pub use data::MetricValue;
 use data::*;
 
 mod tracing_span;
@@ -160,38 +162,22 @@ impl MetricsSink {
         for mut entry in self.metrics.iter_mut() {
             let (key, metric) = entry.pair_mut();
 
-            // If OTLP export is enabled, also send metrics to OpenTelemetry
+            // Get both the value and string representation of the metric (this also resets the metric)
+            let Some((value, metric_str)) = metric.value_and_fmt_and_reset() else {
+                continue;
+            };
+
+            // If OTLP export is enabled, send metrics to OpenTelemetry
             if let Some(exporter) = &self.otlp_exporter {
                 // Convert labels to OpenTelemetry KeyValue pairs
                 let attributes: Vec<KeyValue> = key
                     .labels()
                     .map(|label| KeyValue::new(label.key().to_string(), label.value().to_string()))
                     .collect();
 
-                // Record the metric based on its type
-                match metric {
-                    Metric::Counter(counter) => {
-                        if let Some((value, _)) = counter.load_and_reset() {
-                            exporter.record_counter(key, value, &attributes);
-                        }
-                    }
-                    Metric::Gauge(gauge) => {
-                        if let Some(value) = gauge.load_if_changed() {
-                            exporter.record_gauge(key, value, &attributes);
-                        }
-                    }
-                    Metric::Histogram(histogram) => {
-                        histogram.run_and_reset(|h| {
-                            let value = h.mean();
-                            exporter.record_histogram(key, value, &attributes);
-                        });
-                    }
-                }
+                // Record the metric using its value
+                exporter.record_metric(key, &value, &attributes);
             }
-
-            let Some(metric) = metric.fmt_and_reset() else {
-                continue;
-            };
             let labels = if key.labels().len() == 0 {
                 String::new()
             } else {
@@ -203,7 +189,7 @@ impl MetricsSink {
                         .join(",")
                 )
             };
-            metrics.push(format!("{}{}: {}", key.name(), labels, metric));
+            metrics.push(format!("{}{}: {}", key.name(), labels, metric_str));
         }
 
         metrics.sort();
@@ -497,7 +483,7 @@ mod tests {
         let error = result.unwrap_err().to_string();
         assert!(
             error.contains("Invalid OTLP endpoint configuration"),
-            "Error message should indicate invalid configuration: {error}"
+            "Error message should indicate invalid configuration: {error}",
         );
 
         // Test with no OTLP config (should succeed)

mountpoint-s3-fs/src/metrics/data.rs

@@ -1,6 +1,14 @@
 use crate::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use crate::sync::{Arc, Mutex};
 
+/// Represents the value of a metric
+#[derive(Debug, Clone)]
+pub enum MetricValue {
+    Counter(u64),
+    Gauge(f64),
+    Histogram(f64),
+}
+
 /// A single metric
 #[derive(Debug)]
 pub enum Metric {
@@ -43,34 +51,43 @@ impl Metric {
         metrics::Histogram::from_arc(inner.clone())
     }
 
-    /// Generate a string representation of this metric, or None if the metric has had no values
+    /// Generate both the value and string representation of this metric, or None if the metric has had no values
     /// emitted since the last call to this function.
-    pub fn fmt_and_reset(&self) -> Option<String> {
+    pub fn value_and_fmt_and_reset(&self) -> Option<(MetricValue, String)> {
         match self {
             Metric::Counter(inner) => {
                 let (sum, n) = inner.load_and_reset()?;
-                if n == 1 {
-                    Some(format!("{sum}"))
+                let fmt = if n == 1 {
+                    format!("{sum}")
                 } else {
-                    Some(format!("{sum} (n={n})"))
-                }
+                    format!("{sum} (n={n})")
+                };
+                Some((MetricValue::Counter(sum), fmt))
             }
             // Gauges can't reset because they can be incremented/decremented
-            Metric::Gauge(inner) => inner.load_if_changed().map(|value| format!("{value}")),
-            Metric::Histogram(histogram) => histogram.run_and_reset(|histogram| {
-                format!(
-                    "n={}: min={} p10={} p50={} avg={:.2} p90={} p99={} p99.9={} max={}",
-                    histogram.len(),
-                    histogram.min(),
-                    histogram.value_at_quantile(0.1),
-                    histogram.value_at_quantile(0.5),
-                    histogram.mean(),
-                    histogram.value_at_quantile(0.9),
-                    histogram.value_at_quantile(0.99),
-                    histogram.value_at_quantile(0.999),
-                    histogram.max(),
-                )
-            }),
+            Metric::Gauge(inner) => {
+                let value = inner.load_if_changed()?;
+                let fmt = format!("{value}");
+                Some((MetricValue::Gauge(value), fmt))
+            }
+            Metric::Histogram(histogram) => {
+                // run_and_reset already returns an Option, so we map it to our return type
+                histogram.run_and_reset(|histogram| {
+                    let fmt = format!(
+                        "n={}: min={} p10={} p50={} avg={:.2} p90={} p99={} p99.9={} max={}",
+                        histogram.len(),
+                        histogram.min(),
+                        histogram.value_at_quantile(0.1),
+                        histogram.value_at_quantile(0.5),
+                        histogram.mean(),
+                        histogram.value_at_quantile(0.9),
+                        histogram.value_at_quantile(0.99),
+                        histogram.value_at_quantile(0.999),
+                        histogram.max(),
+                    );
+                    (MetricValue::Histogram(histogram.max() as f64), fmt)
+                })
+            }
         }
     }
 }

mountpoint-s3-fs/src/metrics_otel.rs

@@ -3,7 +3,10 @@ use opentelemetry::global;
 use opentelemetry_otlp::{Protocol, WithExportConfig};
 use std::time::Duration;
 
+use crate::metrics::MetricValue;
 use metrics::Key;
+use std::collections::HashMap;
+use std::sync::Mutex;
 
 /// Configuration for OpenTelemetry metrics export
 #[derive(Debug, Clone)]
@@ -19,7 +22,7 @@ impl OtlpConfig {
     pub fn new(endpoint: &str) -> Self {
         Self {
             endpoint: endpoint.to_string(),
-            interval_secs: 5, // Default to 5 seconds
+            interval_secs: 60, // Default to 60 seconds to align with the meter provider default interval
         }
     }
 
@@ -33,17 +36,31 @@ impl OtlpConfig {
 #[derive(Debug)]
 pub struct OtlpMetricsExporter {
     meter: opentelemetry::metrics::Meter,
+    counters: Mutex<HashMap<String, opentelemetry::metrics::Counter<u64>>>,
+    gauges: Mutex<HashMap<String, opentelemetry::metrics::Gauge<f64>>>,
+    histograms: Mutex<HashMap<String, opentelemetry::metrics::Histogram<f64>>>,
 }
 
 impl OtlpMetricsExporter {
     /// Create a new OtlpMetricsExporter with the specified configuration
     /// Returns a Result containing the new exporter or an error if initialisation failed
     pub fn new(config: &OtlpConfig) -> Result<Self, Box<dyn std::error::Error>> {
+        // Ensure endpoint ends with /v1/metrics
+        let endpoint_url = if !config.endpoint.ends_with("/v1/metrics") {
+            if config.endpoint.ends_with('/') {
+                format!("{}v1/metrics", config.endpoint)
+            } else {
+                format!("{}/v1/metrics", config.endpoint)
+            }
+        } else {
+            config.endpoint.to_string()
+        };
+
         // Initialise OTLP exporter using HTTP binary protocol with the specified endpoint
         let exporter = opentelemetry_otlp::MetricExporter::builder()
             .with_http()
             .with_protocol(Protocol::HttpBinary)
-            .with_endpoint(&config.endpoint)
+            .with_endpoint(&endpoint_url)
             .build()?;
 
         // Create a meter provider with the OTLP Metric Exporter that will collect and export metrics at regular intervals
@@ -62,29 +79,55 @@ impl OtlpMetricsExporter {
         // The meter will be used to create specific metric instruments (counters, gauges, histograms) and record values to them
         let meter = global::meter("mountpoint-s3");
 
-        Ok(Self { meter })
+        Ok(Self {
+            meter,
+            counters: Mutex::new(HashMap::new()),
+            gauges: Mutex::new(HashMap::new()),
+            histograms: Mutex::new(HashMap::new()),
+        })
     }
 
     /// Record a counter metric in OTel format
     pub fn record_counter(&self, key: &Key, value: u64, attributes: &[KeyValue]) {
-        let counter = self.meter.u64_counter(key.name().to_string()).build();
-
+        let name = format!("mountpoint.{}", key.name());
+        let mut counters = self.counters.lock().unwrap();
+        let counter = counters
+            .entry(name.clone())
+            .or_insert_with(|| self.meter.u64_counter(name).build());
         counter.add(value, attributes);
     }
 
     /// Record a gauge metric in OTel format
     pub fn record_gauge(&self, key: &Key, value: f64, attributes: &[KeyValue]) {
-        let gauge = self.meter.f64_gauge(key.name().to_string()).build();
-
+        let name = format!("mountpoint.{}", key.name());
+        let mut gauges = self.gauges.lock().unwrap();
+        let gauge = gauges
+            .entry(name.clone())
+            .or_insert_with(|| self.meter.f64_gauge(name).build());
         gauge.record(value, attributes);
     }
 
     /// Record a histogram metric in OTel format
     pub fn record_histogram(&self, key: &Key, value: f64, attributes: &[KeyValue]) {
-        let histogram = self.meter.f64_histogram(key.name().to_string()).build();
-
+        let name = format!("mountpoint.{}", key.name());
+        let mut histograms = self.histograms.lock().unwrap();
+        let histogram = histograms
+            .entry(name.clone())
+            .or_insert_with(|| self.meter.f64_histogram(name).build());
         histogram.record(value, attributes);
     }
+
+    /// Record a metric using its MetricValue
+    pub fn record_metric(&self, key: &Key, value: &MetricValue, attributes: &[KeyValue]) {
+        match value {
+            MetricValue::Counter(count) => self.record_counter(key, *count, attributes),
+            MetricValue::Gauge(val) => self.record_gauge(key, *val, attributes),
+            MetricValue::Histogram(_mean) => {
+                // Do nothing for histograms for now
+                // Will be implemented later
+            }
+        }
+    }
 }
 
 #[cfg(test)]