rskit_observability/
metrics.rs

1use std::time::Duration;
2
3use opentelemetry::metrics::{Counter, Gauge, Histogram, UpDownCounter};
4use serde::Deserialize;
5
6use rskit_errors::{AppError, AppResult, ErrorCode};
7
8use crate::tracer::{OtlpProtocol, SERVICE_NAME};
9
10/// Configuration for the OpenTelemetry metrics pipeline.
11#[derive(Debug, Clone, Deserialize)]
12pub struct MetricsConfig {
13    /// Logical service name attached to every metric.
14    pub service_name: String,
15    /// How often metrics are exported to the collector.
16    pub export_interval: Duration,
17    /// OTLP endpoint. When `None`, metrics are collected but not exported.
18    pub otlp_endpoint: Option<String>,
19}
20
21/// Handle to the metrics pipeline — use it to create instruments.
22pub struct MetricsHandle {
23    meter: opentelemetry::metrics::Meter,
24    _provider: opentelemetry_sdk::metrics::SdkMeterProvider,
25}
26
27impl MetricsHandle {
28    /// Create a monotonic `Counter<u64>`.
29    pub fn counter(&self, name: impl Into<String>, description: impl Into<String>) -> Counter<u64> {
30        self.meter
31            .u64_counter(name.into())
32            .with_description(description.into())
33            .build()
34    }
35
36    /// Create a `Histogram<f64>` for latency distributions etc.
37    pub fn histogram(
38        &self,
39        name: impl Into<String>,
40        description: impl Into<String>,
41    ) -> Histogram<f64> {
42        self.meter
43            .f64_histogram(name.into())
44            .with_description(description.into())
45            .build()
46    }
47
48    /// Create a `Gauge<f64>` for point-in-time values.
49    pub fn gauge(&self, name: impl Into<String>, description: impl Into<String>) -> Gauge<f64> {
50        self.meter
51            .f64_gauge(name.into())
52            .with_description(description.into())
53            .build()
54    }
55
56    /// Create an `UpDownCounter<i64>` for values that go up and down.
57    pub fn up_down_counter(
58        &self,
59        name: impl Into<String>,
60        description: impl Into<String>,
61    ) -> UpDownCounter<i64> {
62        self.meter
63            .i64_up_down_counter(name.into())
64            .with_description(description.into())
65            .build()
66    }
67}
68
69/// Initialise an OpenTelemetry metrics pipeline with optional OTLP export.
70pub fn init_metrics(cfg: &MetricsConfig) -> AppResult<MetricsHandle> {
71    init_metrics_with_protocol(cfg, OtlpProtocol::Grpc)
72}
73
74/// Initialise an OpenTelemetry metrics pipeline with an explicit OTLP protocol.
75pub fn init_metrics_with_protocol(
76    cfg: &MetricsConfig,
77    protocol: OtlpProtocol,
78) -> AppResult<MetricsHandle> {
79    use opentelemetry::metrics::MeterProvider as _;
80    use opentelemetry::{InstrumentationScope, KeyValue};
81    use opentelemetry_sdk::Resource;
82    use opentelemetry_sdk::metrics::SdkMeterProvider;
83
84    let resource = Resource::builder_empty()
85        .with_attributes([KeyValue::new(SERVICE_NAME, cfg.service_name.clone())])
86        .build();
87
88    let builder = SdkMeterProvider::builder().with_resource(resource);
89
90    if let Some(endpoint) = &cfg.otlp_endpoint {
91        #[cfg(not(feature = "otlp"))]
92        {
93            let _ = (endpoint, protocol);
94            Err(AppError::new(
95                ErrorCode::InvalidInput,
96                "OTLP metric exporter requires the `otlp` feature",
97            ))?;
98        }
99
100        #[cfg(feature = "otlp")]
101        {
102            use opentelemetry_otlp::{MetricExporter, WithExportConfig};
103            use opentelemetry_sdk::metrics::PeriodicReader;
104
105            let exporter = match protocol {
106                OtlpProtocol::Grpc => MetricExporter::builder()
107                    .with_tonic()
108                    .with_endpoint(endpoint)
109                    .build(),
110                OtlpProtocol::HttpBinary => MetricExporter::builder()
111                    .with_http()
112                    .with_protocol(opentelemetry_otlp::Protocol::HttpBinary)
113                    .with_endpoint(endpoint)
114                    .build(),
115            }
116            .map_err(|e| {
117                AppError::new(ErrorCode::Internal, format!("OTLP metric exporter: {e}"))
118            })?;
119
120            let reader = PeriodicReader::builder(exporter)
121                .with_interval(cfg.export_interval)
122                .build();
123
124            let builder = builder.with_reader(reader);
125            let provider = builder.build();
126            let scope = InstrumentationScope::builder(cfg.service_name.clone()).build();
127            let meter = provider.meter_with_scope(scope);
128
129            return Ok(MetricsHandle {
130                meter,
131                _provider: provider,
132            });
133        }
134    }
135
136    let provider = builder.build();
137    let scope = InstrumentationScope::builder(cfg.service_name.clone()).build();
138    let meter = provider.meter_with_scope(scope);
139
140    Ok(MetricsHandle {
141        meter,
142        _provider: provider,
143    })
144}
145
146#[cfg(test)]
147mod tests {
148    use std::time::Duration;
149
150    use super::*;
151
152    fn config(endpoint: Option<&str>) -> MetricsConfig {
153        MetricsConfig {
154            service_name: "metrics-test".to_string(),
155            export_interval: Duration::from_millis(50),
156            otlp_endpoint: endpoint.map(str::to_string),
157        }
158    }
159
160    #[test]
161    fn no_export_metrics_pipeline_creates_all_instrument_kinds() {
162        let metrics = init_metrics(&config(None)).expect("build metrics provider");
163
164        metrics.counter("requests", "request count").add(1, &[]);
165        metrics
166            .histogram("latency", "request latency")
167            .record(12.5, &[]);
168        metrics.gauge("load", "current load").record(0.75, &[]);
169        metrics
170            .up_down_counter("inflight", "inflight requests")
171            .add(1, &[]);
172    }
173
174    #[cfg(not(feature = "otlp"))]
175    #[test]
176    fn otlp_endpoint_requires_feature() {
177        let error = match init_metrics_with_protocol(
178            &config(Some("http://127.0.0.1:4317")),
179            OtlpProtocol::Grpc,
180        ) {
181            Ok(_) => panic!("otlp disabled should reject metric exporter configuration"),
182            Err(error) => error,
183        };
184
185        assert_eq!(error.code(), ErrorCode::InvalidInput);
186        assert!(error.message().contains("otlp"));
187    }
188
189    #[cfg(feature = "otlp")]
190    #[tokio::test]
191    async fn otlp_metrics_pipeline_builds_for_supported_protocols() {
192        for protocol in [OtlpProtocol::Grpc, OtlpProtocol::HttpBinary] {
193            let metrics =
194                init_metrics_with_protocol(&config(Some("http://127.0.0.1:4317")), protocol)
195                    .expect("build otlp metrics provider");
196            metrics.counter("requests", "request count").add(1, &[]);
197        }
198    }
199}
rskit_observability/metrics.rs

rskit_observability/
metrics.rs
