1use crate::{
3 DbMetricsRecorder, DbQueryMetric, MetricDescriptor, MetricError, MetricResult, MetricsProvider,
4 MetricsRecorder, ProductCounter, ProductGauge, ProductHistogram,
5};
6use std::{future::Future, pin::Pin, sync::Arc, time::Instant};
7use tokio_util::sync::CancellationToken;
8
9struct InfrastructureMetricSet {
10 http_requests_total: ProductCounter,
11 http_request_duration_seconds: ProductHistogram,
12 db_queries_total: ProductCounter,
13 db_query_duration_seconds: ProductHistogram,
14 auth_events_total: ProductCounter,
15 application_events_total: ProductCounter,
16 config_reloads_total: ProductCounter,
17 config_reload_duration_seconds: ProductHistogram,
18 config_reload_changed_keys: ProductHistogram,
19 config_mutations_total: ProductCounter,
20 config_mutation_changed_keys: ProductHistogram,
21 background_tasks_total: ProductCounter,
22 background_tasks_pending: ProductGauge,
23 background_task_retries_total: ProductCounter,
24 external_operations_total: ProductCounter,
25 external_operation_duration_seconds: ProductHistogram,
26 #[cfg(feature = "runtime-health")]
27 health_report_status: ProductGauge,
28 #[cfg(feature = "runtime-health")]
29 health_report_duration_seconds: ProductHistogram,
30 #[cfg(feature = "runtime-health")]
31 health_component_status: ProductGauge,
32 #[cfg(feature = "runtime-health")]
33 health_component_duration_seconds: ProductHistogram,
34 process_memory_rss_bytes: ProductGauge,
35 process_cpu_milliseconds_total: ProductGauge,
36 uptime_seconds: ProductGauge,
37 #[cfg(feature = "allocator-metrics")]
38 process_heap_memory_mib: ProductGauge,
39 #[cfg(feature = "otlp")]
40 exporter_successes: ProductGauge,
41 #[cfg(feature = "otlp")]
42 exporter_failures: ProductGauge,
43 #[cfg(feature = "otlp")]
44 exporter_last_success: ProductGauge,
45}
46impl InfrastructureMetricSet {
47 #[expect(
48 clippy::too_many_lines,
49 reason = "the shared schema and atomic typed registration are reviewed together"
50 )]
51 fn new(provider: &MetricsProvider) -> MetricResult<Self> {
52 let descriptors = [
53 MetricDescriptor::counter("", "http_requests_total", "Total HTTP requests", &["method", "route", "status"]).with_unit("1"),
54 MetricDescriptor::histogram_with_buckets("", "http_request_duration_seconds", "HTTP request duration in seconds", &["method", "route", "status"], &[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0]).with_unit("s"),
55 MetricDescriptor::counter("", "db_queries_total", "Total database queries observed through the shared database metrics adapter", &["backend", "kind", "status"]).with_unit("1"),
56 MetricDescriptor::histogram_with_buckets("", "db_query_duration_seconds", "Database query duration in seconds", &["backend", "kind", "status"], &[
57 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
58 ]).with_unit("s"),
59 MetricDescriptor::counter("", "auth_events_total", "Total authentication events", &["action", "status", "reason"]).with_unit("1"),
60 MetricDescriptor::counter("", "application_events_total", "Total low-cardinality application events", &["category", "event", "status"]).with_unit("1"),
61 MetricDescriptor::counter("", "config_reloads_total", "Total runtime config reload attempts", &["source", "decision", "status"]).with_unit("1"),
62 MetricDescriptor::histogram_with_buckets("", "config_reload_duration_seconds", "Runtime config reload duration in seconds", &["source", "decision", "status"], &[
63 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
64 ]).with_unit("s"),
65 MetricDescriptor::histogram_with_buckets("", "config_reload_changed_keys", "Number of changed keys observed by runtime config reload attempts", &["source", "decision", "status"], &[0.0, 1.0, 2.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0]).with_unit("1"),
66 MetricDescriptor::counter("", "config_mutations_total", "Total runtime config mutation attempts", &["source", "operation", "status"]).with_unit("1"),
67 MetricDescriptor::histogram_with_buckets("", "config_mutation_changed_keys", "Number of changed keys in runtime config mutation attempts", &["source", "operation", "status"], &[0.0, 1.0, 2.0, 5.0, 10.0, 25.0, 50.0, 100.0]).with_unit("1"),
68 MetricDescriptor::counter("", "background_tasks_total", "Total background task state transitions", &["kind", "status"]).with_unit("1"),
69 MetricDescriptor::gauge("", "background_tasks_pending", "Pending or retryable background task backlog", &[]).with_unit("1"),
70 MetricDescriptor::counter("", "background_task_retries_total", "Total background task retry transitions", &["kind"]).with_unit("1"),
71 MetricDescriptor::counter("", "external_operations_total", "Total operations against external systems", &["system", "operation", "status"]).with_unit("1"),
72 MetricDescriptor::histogram_with_buckets("", "external_operation_duration_seconds", "External system operation duration in seconds", &["system", "operation", "status"], &[
73 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0, 15.0, 60.0,
74 ]).with_unit("s"),
75 #[cfg(feature = "runtime-health")]
76 MetricDescriptor::gauge("", "health_report_status", "Aggregate health status for a health check scope: healthy=0, degraded=1, unhealthy=2", &["scope"]).with_unit("1"),
77 #[cfg(feature = "runtime-health")]
78 MetricDescriptor::histogram_with_buckets("", "health_report_duration_seconds", "Aggregate health check duration in seconds", &["scope", "status"], &[
79 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
80 ]).with_unit("s"),
81 #[cfg(feature = "runtime-health")]
82 MetricDescriptor::gauge("", "health_component_status", "Health component status for a health check scope: healthy=0, degraded=1, unhealthy=2", &["scope", "component"]).with_unit("1"),
83 #[cfg(feature = "runtime-health")]
84 MetricDescriptor::histogram_with_buckets("", "health_component_duration_seconds", "Health component check duration in seconds", &["scope", "component", "status"], &[
85 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
86 ]).with_unit("s"),
87 MetricDescriptor::gauge("", "process_memory_rss_bytes", "Process RSS memory in bytes", &[]).with_unit("By"),
88 MetricDescriptor::gauge("", "process_cpu_milliseconds_total", "Process accumulated CPU time in milliseconds", &[]).with_unit("ms"),
89 MetricDescriptor::gauge("", "process_uptime_seconds", "Process uptime in seconds", &[]).with_unit("s"),
90 #[cfg(feature = "allocator-metrics")]
91 MetricDescriptor::gauge("", "process_heap_memory_mib", "Allocator heap memory in MiB: allocated and peak_or_resident", &["kind"]).with_unit("MiBy"),
92 #[cfg(feature = "otlp")]
93 MetricDescriptor::gauge("telemetry", "export_successes", "Cumulative successful push exports observed locally.", &["exporter"]),
94 #[cfg(feature = "otlp")]
95 MetricDescriptor::gauge("telemetry", "export_failures", "Cumulative failed push exports observed locally.", &["exporter"]),
96 #[cfg(feature = "otlp")]
97 MetricDescriptor::gauge("telemetry", "export_last_success_timestamp_seconds", "Last acknowledged push export Unix timestamp.", &["exporter"]).with_unit("s"),
98];
99 let mut handles = provider.register_batch(&descriptors)?.into_iter();
100 Ok(Self {
101 http_requests_total: ProductCounter::from_handle(
102 handles.next().ok_or(MetricError::InvalidDescriptor)?,
103 )?,
104 http_request_duration_seconds: ProductHistogram::from_handle(
105 handles.next().ok_or(MetricError::InvalidDescriptor)?,
106 )?,
107 db_queries_total: ProductCounter::from_handle(
108 handles.next().ok_or(MetricError::InvalidDescriptor)?,
109 )?,
110 db_query_duration_seconds: ProductHistogram::from_handle(
111 handles.next().ok_or(MetricError::InvalidDescriptor)?,
112 )?,
113 auth_events_total: ProductCounter::from_handle(
114 handles.next().ok_or(MetricError::InvalidDescriptor)?,
115 )?,
116 application_events_total: ProductCounter::from_handle(
117 handles.next().ok_or(MetricError::InvalidDescriptor)?,
118 )?,
119 config_reloads_total: ProductCounter::from_handle(
120 handles.next().ok_or(MetricError::InvalidDescriptor)?,
121 )?,
122 config_reload_duration_seconds: ProductHistogram::from_handle(
123 handles.next().ok_or(MetricError::InvalidDescriptor)?,
124 )?,
125 config_reload_changed_keys: ProductHistogram::from_handle(
126 handles.next().ok_or(MetricError::InvalidDescriptor)?,
127 )?,
128 config_mutations_total: ProductCounter::from_handle(
129 handles.next().ok_or(MetricError::InvalidDescriptor)?,
130 )?,
131 config_mutation_changed_keys: ProductHistogram::from_handle(
132 handles.next().ok_or(MetricError::InvalidDescriptor)?,
133 )?,
134 background_tasks_total: ProductCounter::from_handle(
135 handles.next().ok_or(MetricError::InvalidDescriptor)?,
136 )?,
137 background_tasks_pending: ProductGauge::from_handle(
138 handles.next().ok_or(MetricError::InvalidDescriptor)?,
139 )?,
140 background_task_retries_total: ProductCounter::from_handle(
141 handles.next().ok_or(MetricError::InvalidDescriptor)?,
142 )?,
143 external_operations_total: ProductCounter::from_handle(
144 handles.next().ok_or(MetricError::InvalidDescriptor)?,
145 )?,
146 external_operation_duration_seconds: ProductHistogram::from_handle(
147 handles.next().ok_or(MetricError::InvalidDescriptor)?,
148 )?,
149 #[cfg(feature = "runtime-health")]
150 health_report_status: ProductGauge::from_handle(
151 handles.next().ok_or(MetricError::InvalidDescriptor)?,
152 )?,
153 #[cfg(feature = "runtime-health")]
154 health_report_duration_seconds: ProductHistogram::from_handle(
155 handles.next().ok_or(MetricError::InvalidDescriptor)?,
156 )?,
157 #[cfg(feature = "runtime-health")]
158 health_component_status: ProductGauge::from_handle(
159 handles.next().ok_or(MetricError::InvalidDescriptor)?,
160 )?,
161 #[cfg(feature = "runtime-health")]
162 health_component_duration_seconds: ProductHistogram::from_handle(
163 handles.next().ok_or(MetricError::InvalidDescriptor)?,
164 )?,
165 process_memory_rss_bytes: ProductGauge::from_handle(
166 handles.next().ok_or(MetricError::InvalidDescriptor)?,
167 )?,
168 process_cpu_milliseconds_total: ProductGauge::from_handle(
169 handles.next().ok_or(MetricError::InvalidDescriptor)?,
170 )?,
171 uptime_seconds: ProductGauge::from_handle(
172 handles.next().ok_or(MetricError::InvalidDescriptor)?,
173 )?,
174 #[cfg(feature = "allocator-metrics")]
175 process_heap_memory_mib: ProductGauge::from_handle(
176 handles.next().ok_or(MetricError::InvalidDescriptor)?,
177 )?,
178 #[cfg(feature = "otlp")]
179 exporter_successes: ProductGauge::from_handle(
180 handles.next().ok_or(MetricError::InvalidDescriptor)?,
181 )?,
182 #[cfg(feature = "otlp")]
183 exporter_failures: ProductGauge::from_handle(
184 handles.next().ok_or(MetricError::InvalidDescriptor)?,
185 )?,
186 #[cfg(feature = "otlp")]
187 exporter_last_success: ProductGauge::from_handle(
188 handles.next().ok_or(MetricError::InvalidDescriptor)?,
189 )?,
190 })
191 }
192 fn record_http_request(&self, method: &str, route: &str, status: u16, duration_seconds: f64) {
193 let metrics = self;
194
195 let status = status.to_string();
196 metrics
197 .http_requests_total
198 .inc(&[method, route, &status], 1);
199 metrics
200 .http_request_duration_seconds
201 .observe(&[method, route, &status], duration_seconds);
202 }
203
204 fn record_db_query(&self, metric: &DbQueryMetric) {
205 let metrics = self;
206
207 let backend = metric.backend.as_label();
208 let kind = metric.kind.as_label();
209 let status = metric.status_label();
210
211 metrics.db_queries_total.inc(&[backend, kind, status], 1);
212 metrics
213 .db_query_duration_seconds
214 .observe(&[backend, kind, status], metric.elapsed.as_secs_f64());
215 }
216
217 fn record_auth_event(&self, action: &'static str, status: &'static str, reason: &'static str) {
218 let metrics = self;
219
220 metrics.auth_events_total.inc(&[action, status, reason], 1);
221 }
222
223 fn record_application_event(
224 &self,
225 category: &'static str,
226 event: &'static str,
227 status: &'static str,
228 ) {
229 let metrics = self;
230
231 metrics
232 .application_events_total
233 .inc(&[category, event, status], 1);
234 }
235
236 fn record_config_reload(
237 &self,
238 source: &'static str,
239 decision: &'static str,
240 status: &'static str,
241 changed_keys: u64,
242 duration_seconds: f64,
243 ) {
244 let metrics = self;
245
246 metrics
247 .config_reloads_total
248 .inc(&[source, decision, status], 1);
249 metrics
250 .config_reload_duration_seconds
251 .observe(&[source, decision, status], duration_seconds);
252 metrics
253 .config_reload_changed_keys
254 .observe(&[source, decision, status], metric_u64(changed_keys));
255 }
256
257 fn record_config_mutation(
258 &self,
259 source: &'static str,
260 operation: &'static str,
261 status: &'static str,
262 changed_keys: u64,
263 ) {
264 let metrics = self;
265
266 metrics
267 .config_mutations_total
268 .inc(&[source, operation, status], 1);
269 metrics
270 .config_mutation_changed_keys
271 .observe(&[source, operation, status], metric_u64(changed_keys));
272 }
273
274 fn record_background_task_transition(&self, kind: &'static str, status: &'static str) {
275 let metrics = self;
276
277 metrics.background_tasks_total.inc(&[kind, status], 1);
278 if status == "retry" {
279 metrics.background_task_retries_total.inc(&[kind], 1);
280 }
281 }
282
283 fn set_background_tasks_pending(&self, pending: u64) {
284 let metrics = self;
285
286 metrics
287 .background_tasks_pending
288 .set(&[], metric_u64(pending));
289 }
290
291 fn record_external_operation(
292 &self,
293 system: &'static str,
294 operation: &'static str,
295 status: &'static str,
296 duration_seconds: f64,
297 ) {
298 let metrics = self;
299
300 metrics
301 .external_operations_total
302 .inc(&[system, operation, status], 1);
303 metrics
304 .external_operation_duration_seconds
305 .observe(&[system, operation, status], duration_seconds);
306 }
307
308 #[cfg(feature = "runtime-health")]
310 fn record_health_report(
311 &self,
312 scope: &'static str,
313 status_label: &'static str,
314 status_value: f64,
315 duration_seconds: f64,
316 ) {
317 let metrics = self;
318
319 metrics.health_report_status.set(&[scope], status_value);
320 metrics
321 .health_report_duration_seconds
322 .observe(&[scope, status_label], duration_seconds);
323 }
324
325 #[cfg(feature = "runtime-health")]
327 fn record_health_component(
328 &self,
329 scope: &'static str,
330 component: &'static str,
331 status_label: &'static str,
332 status_value: f64,
333 duration_seconds: f64,
334 ) {
335 let metrics = self;
336
337 metrics
338 .health_component_status
339 .set(&[scope, component], status_value);
340 metrics
341 .health_component_duration_seconds
342 .observe(&[scope, component, status_label], duration_seconds);
343 }
344}
345#[expect(
346 clippy::cast_precision_loss,
347 reason = "Prometheus gauges and histograms use f64 samples; large integer observations are diagnostic approximations."
348)]
349fn metric_u64(value: u64) -> f64 {
350 value as f64
351}
352
353#[derive(Clone)]
354pub struct InfrastructureMetricsRecorder {
355 provider: MetricsProvider,
356 metrics: Arc<InfrastructureMetricSet>,
357 started_at: Instant,
358}
359impl InfrastructureMetricsRecorder {
360 pub fn new(provider: MetricsProvider) -> MetricResult<Self> {
363 let metrics = Arc::new(InfrastructureMetricSet::new(&provider)?);
364 Ok(Self {
365 provider,
366 metrics,
367 started_at: Instant::now(),
368 })
369 }
370 #[must_use]
371 pub fn provider(&self) -> &MetricsProvider {
372 &self.provider
373 }
374}
375impl DbMetricsRecorder for InfrastructureMetricsRecorder {
376 fn enabled(&self) -> bool {
377 self.provider.enabled()
378 }
379
380 fn record_db_query(&self, metric: &DbQueryMetric) {
381 self.metrics.record_db_query(metric);
382 }
383}
384
385impl MetricsRecorder for InfrastructureMetricsRecorder {
386 fn provider(&self) -> Option<MetricsProvider> {
387 Some(self.provider.clone())
388 }
389 #[cfg(feature = "runtime-health")]
390 fn record_health(&self, scope: &'static str, report: &aster_forge_runtime::SystemHealthReport) {
391 report.record_metrics(scope, self);
392 }
393 fn record_http_request(&self, method: &str, route: &str, status: u16, duration_seconds: f64) {
394 self.metrics
395 .record_http_request(method, route, status, duration_seconds);
396 }
397
398 fn record_auth_event(&self, action: &'static str, status: &'static str, reason: &'static str) {
399 self.metrics.record_auth_event(action, status, reason);
400 }
401
402 fn record_application_event(
403 &self,
404 category: &'static str,
405 event: &'static str,
406 status: &'static str,
407 ) {
408 self.metrics
409 .record_application_event(category, event, status);
410 }
411
412 fn record_config_reload(
413 &self,
414 source: &'static str,
415 decision: &'static str,
416 status: &'static str,
417 changed_keys: u64,
418 duration_seconds: f64,
419 ) {
420 self.metrics
421 .record_config_reload(source, decision, status, changed_keys, duration_seconds);
422 }
423
424 fn record_config_mutation(
425 &self,
426 source: &'static str,
427 operation: &'static str,
428 status: &'static str,
429 changed_keys: u64,
430 ) {
431 self.metrics
432 .record_config_mutation(source, operation, status, changed_keys);
433 }
434
435 fn record_background_task_transition(&self, kind: &'static str, status: &'static str) {
436 self.metrics.record_background_task_transition(kind, status);
437 }
438
439 fn set_background_tasks_pending(&self, pending: u64) {
440 self.metrics.set_background_tasks_pending(pending);
441 }
442
443 fn record_external_operation(
444 &self,
445 system: &'static str,
446 operation: &'static str,
447 status: &'static str,
448 duration_seconds: f64,
449 ) {
450 self.metrics
451 .record_external_operation(system, operation, status, duration_seconds);
452 }
453
454 fn system_metrics_updater_task(
455 &self,
456 shutdown_token: CancellationToken,
457 ) -> Option<Pin<Box<dyn Future<Output = ()> + Send + 'static>>> {
458 if !self.provider.enabled() {
459 return None;
460 }
461 Some(Box::pin(system_metrics_updater_task(
462 self.clone(),
463 shutdown_token,
464 )))
465 }
466}
467
468#[cfg(feature = "runtime-health")]
469impl aster_forge_runtime::HealthMetricsRecorder for InfrastructureMetricsRecorder {
470 fn record_health_report(
471 &self,
472 scope: &'static str,
473 status: aster_forge_runtime::HealthStatus,
474 duration_seconds: f64,
475 ) {
476 self.metrics.record_health_report(
477 scope,
478 status.as_str(),
479 health_status_value(status.as_str()),
480 duration_seconds,
481 );
482 }
483
484 fn record_health_component(
485 &self,
486 scope: &'static str,
487 component: &aster_forge_runtime::HealthComponentReport,
488 duration_seconds: f64,
489 ) {
490 self.metrics.record_health_component(
491 scope,
492 component.name,
493 component.status.as_str(),
494 health_status_value(component.status.as_str()),
495 duration_seconds,
496 );
497 }
498}
499
500#[cfg(feature = "runtime-health")]
501fn health_status_value(status: &'static str) -> f64 {
502 match status {
503 "healthy" => 0.0,
504 "degraded" => 1.0,
505 _ => 2.0,
506 }
507}
508
509async fn system_metrics_updater_task(
510 recorder: InfrastructureMetricsRecorder,
511 shutdown: CancellationToken,
512) {
513 use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System};
514 let system = Arc::new(parking_lot::Mutex::new(System::new()));
515 let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
516 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
517 loop {
518 tokio::select! { biased; () = shutdown.cancelled() => break, _ = interval.tick() => {} }
519 if !recorder.provider.enabled() {
520 break;
521 }
522 let system = system.clone();
523 let recorder = recorder.clone();
524 let task = tokio::task::spawn_blocking(move || {
525 let mut system = system.lock();
526 let pid = Pid::from_u32(std::process::id());
527 system.refresh_processes_specifics(
528 ProcessesToUpdate::Some(&[pid]),
529 true,
530 ProcessRefreshKind::nothing().with_memory().with_cpu(),
531 );
532 if let Some(process) = system.process(pid) {
533 recorder
534 .metrics
535 .process_memory_rss_bytes
536 .set(&[], metric_u64(process.memory()));
537 recorder
538 .metrics
539 .process_cpu_milliseconds_total
540 .set(&[], metric_u64(process.accumulated_cpu_time()));
541 }
542 recorder
543 .metrics
544 .uptime_seconds
545 .set(&[], recorder.started_at.elapsed().as_secs_f64());
546 #[cfg(feature = "otlp")]
547 if let Some(diagnostics) = recorder.provider.otlp_diagnostics() {
548 let snapshot = diagnostics.snapshot();
549 recorder
550 .metrics
551 .exporter_successes
552 .set(&["otlp"], metric_u64(snapshot.successes));
553 recorder
554 .metrics
555 .exporter_failures
556 .set(&["otlp"], metric_u64(snapshot.failures));
557 recorder
558 .metrics
559 .exporter_last_success
560 .set(&["otlp"], metric_u64(snapshot.last_success_unix_seconds));
561 }
562 #[cfg(feature = "allocator-metrics")]
563 {
564 let (allocated, resident) = aster_forge_alloc::stats();
565 recorder
566 .metrics
567 .process_heap_memory_mib
568 .set(&["allocated"], allocated);
569 recorder
570 .metrics
571 .process_heap_memory_mib
572 .set(&["peak_or_resident"], resident);
573 }
574 });
575 if task.await.is_err() {
576 tracing::warn!("system metrics sampler failed");
577 }
578 }
579}