llm_optimizer_processor/telemetry/
metrics.rs

1//! OTLP metrics export for OpenTelemetry.
2//!
3//! This module provides metrics export to OTLP collectors with support for
4//! different temporality modes and aggregation strategies.
5
6use crate::telemetry::resource::ResourceBuilder;
7use opentelemetry::{global, metrics::MeterProvider as _, KeyValue};
8use opentelemetry_otlp::WithExportConfig;
9use opentelemetry_sdk::{
10    metrics::{
11        reader::{DefaultAggregationSelector, DefaultTemporalitySelector, TemporalitySelector},
12        PeriodicReader, SdkMeterProvider,
13    },
14    runtime,
15    Resource,
16};
17use serde::{Deserialize, Serialize};
18use std::collections::HashMap;
19use std::time::Duration;
20
21/// Aggregation temporality for metrics.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(rename_all = "lowercase")]
24pub enum Temporality {
25    /// Delta temporality (export deltas between collections).
26    Delta,
27    /// Cumulative temporality (export cumulative values).
28    Cumulative,
29}
30
31impl Default for Temporality {
32    fn default() -> Self {
33        Self::Cumulative
34    }
35}
36
37/// Metric export interval configuration.
38#[derive(Debug, Clone, Serialize, Deserialize)]
39pub struct ExportConfig {
40    /// Export interval (how often to export metrics).
41    #[serde(with = "humantime_serde", default = "default_export_interval")]
42    pub export_interval: Duration,
43
44    /// Export timeout (maximum time to wait for export).
45    #[serde(with = "humantime_serde", default = "default_export_timeout")]
46    pub export_timeout: Duration,
47}
48
49fn default_export_interval() -> Duration {
50    Duration::from_secs(60)
51}
52
53fn default_export_timeout() -> Duration {
54    Duration::from_secs(30)
55}
56
57impl Default for ExportConfig {
58    fn default() -> Self {
59        Self {
60            export_interval: default_export_interval(),
61            export_timeout: default_export_timeout(),
62        }
63    }
64}
65
66/// Metrics configuration.
67#[derive(Debug, Clone, Serialize, Deserialize)]
68pub struct MetricsConfig {
69    /// Whether metrics export is enabled.
70    #[serde(default = "default_enabled")]
71    pub enabled: bool,
72
73    /// Service name.
74    pub service_name: String,
75
76    /// Service version.
77    #[serde(default = "default_service_version")]
78    pub service_version: String,
79
80    /// OTLP endpoint (e.g., "http://localhost:4317").
81    #[serde(default = "default_otlp_endpoint")]
82    pub otlp_endpoint: String,
83
84    /// Metric temporality (delta or cumulative).
85    #[serde(default)]
86    pub temporality: Temporality,
87
88    /// Export configuration.
89    #[serde(default)]
90    pub export_config: ExportConfig,
91
92    /// Additional headers for OTLP export.
93    #[serde(default)]
94    pub headers: HashMap<String, String>,
95
96    /// Deployment environment.
97    #[serde(default)]
98    pub environment: Option<String>,
99}
100
101fn default_enabled() -> bool {
102    true
103}
104
105fn default_service_version() -> String {
106    env!("CARGO_PKG_VERSION").to_string()
107}
108
109fn default_otlp_endpoint() -> String {
110    "http://localhost:4317".to_string()
111}
112
113impl Default for MetricsConfig {
114    fn default() -> Self {
115        Self {
116            enabled: default_enabled(),
117            service_name: "llm-optimizer-processor".to_string(),
118            service_version: default_service_version(),
119            otlp_endpoint: default_otlp_endpoint(),
120            temporality: Temporality::default(),
121            export_config: ExportConfig::default(),
122            headers: HashMap::new(),
123            environment: None,
124        }
125    }
126}
127
128/// Initializes the global meter provider with OTLP exporter.
129///
130/// # Errors
131///
132/// Returns an error if the meter provider cannot be initialized.
133pub fn init_meter_provider(
134    config: MetricsConfig,
135) -> Result<SdkMeterProvider, Box<dyn std::error::Error>> {
136    if !config.enabled {
137        tracing::info!("OpenTelemetry metrics export is disabled");
138        return Err("metrics disabled".into());
139    }
140
141    // Build resource
142    let resource = ResourceBuilder::new()
143        .with_service_name(config.service_name.clone())
144        .with_service_version(config.service_version.clone())
145        .with_attribute("service.component", "stream-processor");
146
147    let resource = if let Some(env) = config.environment.clone() {
148        resource.with_deployment_environment(env)
149    } else {
150        resource
151    };
152
153    let resource = resource.build();
154
155    // Create OTLP exporter with gRPC
156    let exporter_builder = opentelemetry_otlp::new_exporter()
157        .tonic()
158        .with_endpoint(&config.otlp_endpoint);
159
160    // Add custom headers if present
161    let exporter_builder = if !config.headers.is_empty() {
162        let mut metadata = tonic::metadata::MetadataMap::new();
163        for (key, value) in &config.headers {
164            if let (Ok(k), Ok(v)) = (
165                key.parse::<tonic::metadata::MetadataKey<_>>(),
166                value.parse::<tonic::metadata::MetadataValue<_>>(),
167            ) {
168                metadata.insert(k, v);
169            }
170        }
171        exporter_builder.with_metadata(metadata)
172    } else {
173        exporter_builder
174    };
175
176    let exporter = opentelemetry_otlp::new_pipeline()
177        .metrics(runtime::Tokio)
178        .with_exporter(exporter_builder)
179        .with_resource(resource)
180        .with_period(config.export_config.export_interval)
181        .with_timeout(config.export_config.export_timeout)
182        .build()?;
183
184    Ok(exporter)
185}
186
187/// Shuts down the global meter provider and flushes all pending metrics.
188pub fn shutdown_meter_provider() -> Result<(), Box<dyn std::error::Error>> {
189    Ok(())
190}
191
192/// Helper struct for recording metrics with common attributes.
193#[derive(Debug, Clone)]
194pub struct MetricsRecorder {
195    common_attributes: Vec<KeyValue>,
196}
197
198impl MetricsRecorder {
199    /// Creates a new metrics recorder.
200    pub fn new() -> Self {
201        Self {
202            common_attributes: Vec::new(),
203        }
204    }
205
206    /// Adds a common attribute that will be included in all metrics.
207    pub fn with_attribute(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
208        self.common_attributes
209            .push(KeyValue::new(key.into(), value.into()));
210        self
211    }
212
213    /// Adds multiple common attributes.
214    pub fn with_attributes(mut self, attributes: Vec<KeyValue>) -> Self {
215        self.common_attributes.extend(attributes);
216        self
217    }
218
219    /// Records a counter increment.
220    pub fn record_counter(
221        &self,
222        meter_name: &str,
223        counter_name: &str,
224        value: u64,
225        attributes: Vec<KeyValue>,
226    ) {
227        let meter = global::meter(meter_name.to_string());
228        let counter = meter.u64_counter(counter_name.to_string()).init();
229
230        let mut all_attrs = self.common_attributes.clone();
231        all_attrs.extend(attributes);
232
233        counter.add(value, &all_attrs);
234    }
235
236    /// Records a histogram value.
237    pub fn record_histogram(
238        &self,
239        meter_name: &str,
240        histogram_name: &str,
241        value: f64,
242        attributes: Vec<KeyValue>,
243    ) {
244        let meter = global::meter(meter_name.to_string());
245        let histogram = meter.f64_histogram(histogram_name.to_string()).init();
246
247        let mut all_attrs = self.common_attributes.clone();
248        all_attrs.extend(attributes);
249
250        histogram.record(value, &all_attrs);
251    }
252
253    /// Records a gauge value (using UpDownCounter).
254    pub fn record_gauge(
255        &self,
256        meter_name: &str,
257        gauge_name: &str,
258        value: i64,
259        attributes: Vec<KeyValue>,
260    ) {
261        let meter = global::meter(meter_name.to_string());
262        let gauge = meter.i64_up_down_counter(gauge_name.to_string()).init();
263
264        let mut all_attrs = self.common_attributes.clone();
265        all_attrs.extend(attributes);
266
267        gauge.add(value, &all_attrs);
268    }
269}
270
271impl Default for MetricsRecorder {
272    fn default() -> Self {
273        Self::new()
274    }
275}
276
277/// Processor-specific metric names.
278pub mod metric_names {
279    pub const EVENTS_PROCESSED: &str = "processor.events.processed";
280    pub const EVENTS_DROPPED: &str = "processor.events.dropped";
281    pub const PROCESSING_DURATION: &str = "processor.processing.duration";
282    pub const WINDOW_TRIGGER_COUNT: &str = "processor.window.triggers";
283    pub const AGGREGATION_DURATION: &str = "processor.aggregation.duration";
284    pub const STATE_OPERATION_DURATION: &str = "processor.state.operation.duration";
285    pub const DEDUPLICATION_COUNT: &str = "processor.deduplication.count";
286    pub const WATERMARK_LAG: &str = "processor.watermark.lag";
287}
288
289/// Common metric attribute keys.
290pub mod metric_attributes {
291    pub const EVENT_TYPE: &str = "event.type";
292    pub const WINDOW_TYPE: &str = "window.type";
293    pub const AGGREGATION_TYPE: &str = "aggregation.type";
294    pub const STATE_BACKEND: &str = "state.backend";
295    pub const OPERATION_TYPE: &str = "operation.type";
296    pub const ERROR_TYPE: &str = "error.type";
297    pub const PARTITION_ID: &str = "partition.id";
298}
299
300#[cfg(test)]
301mod tests {
302    use super::*;
303
304    #[test]
305    fn test_temporality_default() {
306        let temporality = Temporality::default();
307        assert_eq!(temporality, Temporality::Cumulative);
308    }
309
310    #[test]
311    fn test_export_config_default() {
312        let config = ExportConfig::default();
313        assert_eq!(config.export_interval, Duration::from_secs(60));
314        assert_eq!(config.export_timeout, Duration::from_secs(30));
315    }
316
317    #[test]
318    fn test_metrics_config_default() {
319        let config = MetricsConfig::default();
320        assert!(config.enabled);
321        assert_eq!(config.service_name, "llm-optimizer-processor");
322        assert_eq!(config.otlp_endpoint, "http://localhost:4317");
323        assert_eq!(config.temporality, Temporality::Cumulative);
324    }
325
326    #[test]
327    fn test_metrics_recorder_new() {
328        let recorder = MetricsRecorder::new();
329        assert!(recorder.common_attributes.is_empty());
330    }
331
332    #[test]
333    fn test_metrics_recorder_with_attributes() {
334        let recorder = MetricsRecorder::new()
335            .with_attribute("service", "test")
336            .with_attribute("env", "dev");
337
338        assert_eq!(recorder.common_attributes.len(), 2);
339    }
340
341    #[test]
342    fn test_temporality_serialization() {
343        let delta = Temporality::Delta;
344        let cumulative = Temporality::Cumulative;
345
346        let delta_json = serde_json::to_string(&delta).unwrap();
347        let cumulative_json = serde_json::to_string(&cumulative).unwrap();
348
349        assert_eq!(delta_json, "\"delta\"");
350        assert_eq!(cumulative_json, "\"cumulative\"");
351    }
352}
llm_optimizer_processor/telemetry/metrics.rs

llm_optimizer_processor/telemetry/
metrics.rs
