microclaw_engine/internal/observability/
metrics.rs1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3use std::time::Duration;
4
5use opentelemetry::metrics::{Counter, MeterProvider, UpDownCounter};
6use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig};
7use opentelemetry_sdk::metrics::periodic_reader_with_async_runtime::PeriodicReader;
8use opentelemetry_sdk::metrics::SdkMeterProvider;
9use opentelemetry_sdk::runtime;
10use serde_yaml::Value;
11use tracing::{debug, info, warn};
12
13use crate::internal::observability::sdk::{
14 get_bool, get_trimmed, get_u64, parse_headers, OTelSdkContext,
15};
16
17#[derive(Debug, Clone)]
18pub struct OtlpMetricSnapshot {
19 pub timestamp_unix_nano: u64,
20 pub http_requests: i64,
21 pub llm_completions: i64,
22 pub llm_input_tokens: i64,
23 pub llm_output_tokens: i64,
24 pub tool_executions: i64,
25 pub mcp_calls: i64,
26 pub mcp_rate_limited_rejections: i64,
27 pub mcp_bulkhead_rejections: i64,
28 pub mcp_circuit_open_rejections: i64,
29 pub active_sessions: i64,
30}
31
32#[derive(Clone)]
33pub struct OtlpMetricExporter {
34 inner: Arc<OtlpMetricsInner>,
35}
36
37struct OtlpMetricsInner {
38 provider: SdkMeterProvider,
39 instruments: MetricInstruments,
40 previous: Mutex<Option<OtlpMetricSnapshot>>,
41}
42
43#[derive(Clone)]
44struct MetricInstruments {
45 http_requests: Counter<u64>,
46 llm_completions: Counter<u64>,
47 llm_input_tokens: Counter<u64>,
48 llm_output_tokens: Counter<u64>,
49 tool_executions: Counter<u64>,
50 mcp_calls: Counter<u64>,
51 mcp_rate_limited_rejections: Counter<u64>,
52 mcp_bulkhead_rejections: Counter<u64>,
53 mcp_circuit_open_rejections: Counter<u64>,
54 active_sessions: UpDownCounter<i64>,
55}
56
57impl OtlpMetricExporter {
58 pub fn from_observability(observability: Option<&Value>) -> Option<Arc<Self>> {
59 let map = observability?.as_mapping()?;
60 let enabled = get_bool(map, "otlp_enabled").unwrap_or(false);
61 if !enabled {
62 debug!("otlp metric exporter disabled by config");
63 return None;
64 }
65 let endpoint = get_trimmed(map, "otlp_endpoint")?.to_string();
66 let sdk = OTelSdkContext::from_observability(map);
67 let interval_secs = get_u64(map, "otlp_export_interval_seconds")
68 .map(|v| v.clamp(1, 300))
69 .unwrap_or(15);
70 let headers = parse_headers(map);
71
72 let mut exporter_builder = opentelemetry_otlp::MetricExporter::builder()
73 .with_http()
74 .with_protocol(Protocol::HttpBinary)
75 .with_http_client(reqwest::Client::new())
76 .with_endpoint(endpoint.clone());
77 if !headers.is_empty() {
78 exporter_builder =
79 exporter_builder.with_headers(headers.into_iter().collect::<HashMap<_, _>>());
80 }
81 let exporter = match exporter_builder.build() {
82 Ok(exporter) => exporter,
83 Err(err) => {
84 warn!(
85 endpoint = %endpoint,
86 "failed to build otlp metric exporter: {}",
87 err
88 );
89 return None;
90 }
91 };
92
93 let reader = PeriodicReader::builder(exporter, runtime::Tokio)
94 .with_interval(Duration::from_secs(interval_secs))
95 .build();
96 let provider = SdkMeterProvider::builder()
97 .with_resource(sdk.resource)
98 .with_reader(reader)
99 .build();
100 let meter = provider.meter("microclaw.observability.metrics");
101 let instruments = MetricInstruments {
102 http_requests: meter.u64_counter("microclaw_http_requests").build(),
103 llm_completions: meter.u64_counter("microclaw_llm_completions").build(),
104 llm_input_tokens: meter.u64_counter("microclaw_llm_input_tokens").build(),
105 llm_output_tokens: meter.u64_counter("microclaw_llm_output_tokens").build(),
106 tool_executions: meter.u64_counter("microclaw_tool_executions").build(),
107 mcp_calls: meter.u64_counter("microclaw_mcp_calls").build(),
108 mcp_rate_limited_rejections: meter
109 .u64_counter("microclaw_mcp_rate_limited_rejections")
110 .build(),
111 mcp_bulkhead_rejections: meter
112 .u64_counter("microclaw_mcp_bulkhead_rejections")
113 .build(),
114 mcp_circuit_open_rejections: meter
115 .u64_counter("microclaw_mcp_circuit_open_rejections")
116 .build(),
117 active_sessions: meter
118 .i64_up_down_counter("microclaw_active_sessions")
119 .build(),
120 };
121 info!(
122 endpoint = %endpoint,
123 export_interval_secs = interval_secs,
124 service_name = %sdk.service_name,
125 "otlp metric exporter initialized"
126 );
127
128 Some(Arc::new(Self {
129 inner: Arc::new(OtlpMetricsInner {
130 provider,
131 instruments,
132 previous: Mutex::new(None),
133 }),
134 }))
135 }
136
137 pub fn enqueue_metrics(&self, snapshot: OtlpMetricSnapshot) -> Result<(), String> {
138 let mut previous = self.inner.previous.lock().map_err(|_| {
139 warn!("failed to enqueue metrics snapshot: metrics state lock poisoned");
140 "metrics state lock poisoned".to_string()
141 })?;
142
143 let delta = if let Some(prev) = previous.as_ref() {
144 OtlpMetricSnapshot {
145 timestamp_unix_nano: snapshot.timestamp_unix_nano,
146 http_requests: (snapshot.http_requests - prev.http_requests).max(0),
147 llm_completions: (snapshot.llm_completions - prev.llm_completions).max(0),
148 llm_input_tokens: (snapshot.llm_input_tokens - prev.llm_input_tokens).max(0),
149 llm_output_tokens: (snapshot.llm_output_tokens - prev.llm_output_tokens).max(0),
150 tool_executions: (snapshot.tool_executions - prev.tool_executions).max(0),
151 mcp_calls: (snapshot.mcp_calls - prev.mcp_calls).max(0),
152 mcp_rate_limited_rejections: (snapshot.mcp_rate_limited_rejections
153 - prev.mcp_rate_limited_rejections)
154 .max(0),
155 mcp_bulkhead_rejections: (snapshot.mcp_bulkhead_rejections
156 - prev.mcp_bulkhead_rejections)
157 .max(0),
158 mcp_circuit_open_rejections: (snapshot.mcp_circuit_open_rejections
159 - prev.mcp_circuit_open_rejections)
160 .max(0),
161 active_sessions: snapshot.active_sessions - prev.active_sessions,
162 }
163 } else {
164 OtlpMetricSnapshot {
165 timestamp_unix_nano: snapshot.timestamp_unix_nano,
166 http_requests: snapshot.http_requests.max(0),
167 llm_completions: snapshot.llm_completions.max(0),
168 llm_input_tokens: snapshot.llm_input_tokens.max(0),
169 llm_output_tokens: snapshot.llm_output_tokens.max(0),
170 tool_executions: snapshot.tool_executions.max(0),
171 mcp_calls: snapshot.mcp_calls.max(0),
172 mcp_rate_limited_rejections: snapshot.mcp_rate_limited_rejections.max(0),
173 mcp_bulkhead_rejections: snapshot.mcp_bulkhead_rejections.max(0),
174 mcp_circuit_open_rejections: snapshot.mcp_circuit_open_rejections.max(0),
175 active_sessions: snapshot.active_sessions,
176 }
177 };
178
179 self.inner
180 .instruments
181 .http_requests
182 .add(delta.http_requests as u64, &[]);
183 self.inner
184 .instruments
185 .llm_completions
186 .add(delta.llm_completions as u64, &[]);
187 self.inner
188 .instruments
189 .llm_input_tokens
190 .add(delta.llm_input_tokens as u64, &[]);
191 self.inner
192 .instruments
193 .llm_output_tokens
194 .add(delta.llm_output_tokens as u64, &[]);
195 self.inner
196 .instruments
197 .tool_executions
198 .add(delta.tool_executions as u64, &[]);
199 self.inner
200 .instruments
201 .mcp_calls
202 .add(delta.mcp_calls as u64, &[]);
203 self.inner
204 .instruments
205 .mcp_rate_limited_rejections
206 .add(delta.mcp_rate_limited_rejections as u64, &[]);
207 self.inner
208 .instruments
209 .mcp_bulkhead_rejections
210 .add(delta.mcp_bulkhead_rejections as u64, &[]);
211 self.inner
212 .instruments
213 .mcp_circuit_open_rejections
214 .add(delta.mcp_circuit_open_rejections as u64, &[]);
215 self.inner
216 .instruments
217 .active_sessions
218 .add(delta.active_sessions, &[]);
219 debug!(
220 timestamp_unix_nano = snapshot.timestamp_unix_nano,
221 "otlp metric snapshot recorded"
222 );
223
224 *previous = Some(snapshot);
225 Ok(())
226 }
227}
228
229impl Drop for OtlpMetricsInner {
230 fn drop(&mut self) {
231 if let Err(err) = self.provider.force_flush() {
232 warn!("otlp metric exporter force_flush failed: {}", err);
233 } else {
234 debug!("otlp metric exporter force_flush succeeded");
235 }
236 if let Err(err) = self.provider.shutdown() {
237 warn!("otlp metric exporter shutdown failed: {}", err);
238 } else {
239 debug!("otlp metric exporter shutdown succeeded");
240 }
241 }
242}