1use crate::{
11 DbMetricsRecorder, DbQueryMetric, MetricDescriptor, MetricKind, MetricsRecorder,
12 SharedMetricsRecorder,
13};
14use prometheus::{
15 Encoder, Gauge, GaugeVec, HistogramOpts, HistogramVec, IntCounterVec, IntGauge, Opts, Registry,
16 TextEncoder,
17};
18use std::collections::BTreeMap;
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::{Mutex, OnceLock};
22use std::time::Instant;
23use tokio_util::sync::CancellationToken;
24
25static METRICS: OnceLock<PrometheusMetrics> = OnceLock::new();
26static PROCESS_STARTED_AT: OnceLock<Instant> = OnceLock::new();
27
28fn boxed_collector<C>(collector: C) -> Box<dyn prometheus::core::Collector>
29where
30 C: prometheus::core::Collector + 'static,
31{
32 Box::new(collector)
33}
34
35pub struct PrometheusMetrics {
37 registry: Registry,
38 http_requests_total: IntCounterVec,
39 http_request_duration_seconds: HistogramVec,
40 db_queries_total: IntCounterVec,
41 db_query_duration_seconds: HistogramVec,
42 auth_events_total: IntCounterVec,
43 application_events_total: IntCounterVec,
44 config_reloads_total: IntCounterVec,
45 config_reload_duration_seconds: HistogramVec,
46 config_reload_changed_keys: HistogramVec,
47 config_mutations_total: IntCounterVec,
48 config_mutation_changed_keys: HistogramVec,
49 background_tasks_total: IntCounterVec,
50 background_tasks_pending: IntGauge,
51 background_task_retries_total: IntCounterVec,
52 external_operations_total: IntCounterVec,
53 external_operation_duration_seconds: HistogramVec,
54 health_report_status: GaugeVec,
55 health_report_duration_seconds: HistogramVec,
56 health_component_status: GaugeVec,
57 health_component_duration_seconds: HistogramVec,
58 process_memory_rss_bytes: Gauge,
59 process_cpu_milliseconds_total: IntGauge,
60 uptime_seconds: Gauge,
61 #[cfg(feature = "allocator-metrics")]
62 process_heap_memory_mib: GaugeVec,
63 product_metrics: Mutex<ProductMetricRegistry>,
64}
65
66impl PrometheusMetrics {
67 #[expect(
68 clippy::too_many_lines,
69 reason = "Built-in metric construction and registration stay together so initialization remains atomic."
70 )]
71 fn new() -> Result<Self, prometheus::Error> {
72 let registry = Registry::new();
73
74 let http_requests_total = IntCounterVec::new(
75 Opts::new("http_requests_total", "Total HTTP requests"),
76 &["method", "route", "status"],
77 )?;
78 let http_request_duration_seconds = HistogramVec::new(
79 HistogramOpts::new(
80 "http_request_duration_seconds",
81 "HTTP request duration in seconds",
82 )
83 .buckets(vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0]),
84 &["method", "route", "status"],
85 )?;
86 let db_queries_total = IntCounterVec::new(
87 Opts::new(
88 "db_queries_total",
89 "Total database queries observed through the shared database metrics adapter",
90 ),
91 &["backend", "kind", "status"],
92 )?;
93 let db_query_duration_seconds = HistogramVec::new(
94 HistogramOpts::new(
95 "db_query_duration_seconds",
96 "Database query duration in seconds",
97 )
98 .buckets(vec![
99 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
100 ]),
101 &["backend", "kind", "status"],
102 )?;
103 let auth_events_total = IntCounterVec::new(
104 Opts::new("auth_events_total", "Total authentication events"),
105 &["action", "status", "reason"],
106 )?;
107 let application_events_total = IntCounterVec::new(
108 Opts::new(
109 "application_events_total",
110 "Total low-cardinality application events",
111 ),
112 &["category", "event", "status"],
113 )?;
114 let config_reloads_total = IntCounterVec::new(
115 Opts::new(
116 "config_reloads_total",
117 "Total runtime config reload attempts",
118 ),
119 &["source", "decision", "status"],
120 )?;
121 let config_reload_duration_seconds = HistogramVec::new(
122 HistogramOpts::new(
123 "config_reload_duration_seconds",
124 "Runtime config reload duration in seconds",
125 )
126 .buckets(vec![
127 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
128 ]),
129 &["source", "decision", "status"],
130 )?;
131 let config_reload_changed_keys = HistogramVec::new(
132 HistogramOpts::new(
133 "config_reload_changed_keys",
134 "Number of changed keys observed by runtime config reload attempts",
135 )
136 .buckets(vec![0.0, 1.0, 2.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0]),
137 &["source", "decision", "status"],
138 )?;
139 let config_mutations_total = IntCounterVec::new(
140 Opts::new(
141 "config_mutations_total",
142 "Total runtime config mutation attempts",
143 ),
144 &["source", "operation", "status"],
145 )?;
146 let config_mutation_changed_keys = HistogramVec::new(
147 HistogramOpts::new(
148 "config_mutation_changed_keys",
149 "Number of changed keys in runtime config mutation attempts",
150 )
151 .buckets(vec![0.0, 1.0, 2.0, 5.0, 10.0, 25.0, 50.0, 100.0]),
152 &["source", "operation", "status"],
153 )?;
154 let background_tasks_total = IntCounterVec::new(
155 Opts::new(
156 "background_tasks_total",
157 "Total background task state transitions",
158 ),
159 &["kind", "status"],
160 )?;
161 let background_tasks_pending = IntGauge::new(
162 "background_tasks_pending",
163 "Pending or retryable background task backlog",
164 )?;
165 let background_task_retries_total = IntCounterVec::new(
166 Opts::new(
167 "background_task_retries_total",
168 "Total background task retry transitions",
169 ),
170 &["kind"],
171 )?;
172 let external_operations_total = IntCounterVec::new(
173 Opts::new(
174 "external_operations_total",
175 "Total operations against external systems",
176 ),
177 &["system", "operation", "status"],
178 )?;
179 let external_operation_duration_seconds = HistogramVec::new(
180 HistogramOpts::new(
181 "external_operation_duration_seconds",
182 "External system operation duration in seconds",
183 )
184 .buckets(vec![
185 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,
186 ]),
187 &["system", "operation", "status"],
188 )?;
189 let health_report_status = GaugeVec::new(
190 Opts::new(
191 "health_report_status",
192 "Aggregate health status for a health check scope: healthy=0, degraded=1, unhealthy=2",
193 ),
194 &["scope"],
195 )?;
196 let health_report_duration_seconds = HistogramVec::new(
197 HistogramOpts::new(
198 "health_report_duration_seconds",
199 "Aggregate health check duration in seconds",
200 )
201 .buckets(vec![
202 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
203 ]),
204 &["scope", "status"],
205 )?;
206 let health_component_status = GaugeVec::new(
207 Opts::new(
208 "health_component_status",
209 "Health component status for a health check scope: healthy=0, degraded=1, unhealthy=2",
210 ),
211 &["scope", "component"],
212 )?;
213 let health_component_duration_seconds = HistogramVec::new(
214 HistogramOpts::new(
215 "health_component_duration_seconds",
216 "Health component check duration in seconds",
217 )
218 .buckets(vec![
219 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0,
220 ]),
221 &["scope", "component", "status"],
222 )?;
223 let process_memory_rss_bytes =
224 Gauge::new("process_memory_rss_bytes", "Process RSS memory in bytes")?;
225 let process_cpu_milliseconds_total = IntGauge::new(
226 "process_cpu_milliseconds_total",
227 "Process accumulated CPU time in milliseconds",
228 )?;
229 let uptime_seconds = Gauge::new("process_uptime_seconds", "Process uptime in seconds")?;
230 #[cfg(feature = "allocator-metrics")]
231 let process_heap_memory_mib = GaugeVec::new(
232 Opts::new(
233 "process_heap_memory_mib",
234 "Allocator heap memory in MiB: allocated and peak_or_resident",
235 ),
236 &["kind"],
237 )?;
238
239 let collectors: Vec<Box<dyn prometheus::core::Collector>> = vec![
240 boxed_collector(http_requests_total.clone()),
241 boxed_collector(http_request_duration_seconds.clone()),
242 boxed_collector(db_queries_total.clone()),
243 boxed_collector(db_query_duration_seconds.clone()),
244 boxed_collector(auth_events_total.clone()),
245 boxed_collector(application_events_total.clone()),
246 boxed_collector(config_reloads_total.clone()),
247 boxed_collector(config_reload_duration_seconds.clone()),
248 boxed_collector(config_reload_changed_keys.clone()),
249 boxed_collector(config_mutations_total.clone()),
250 boxed_collector(config_mutation_changed_keys.clone()),
251 boxed_collector(background_tasks_total.clone()),
252 boxed_collector(background_tasks_pending.clone()),
253 boxed_collector(background_task_retries_total.clone()),
254 boxed_collector(external_operations_total.clone()),
255 boxed_collector(external_operation_duration_seconds.clone()),
256 boxed_collector(health_report_status.clone()),
257 boxed_collector(health_report_duration_seconds.clone()),
258 boxed_collector(health_component_status.clone()),
259 boxed_collector(health_component_duration_seconds.clone()),
260 boxed_collector(process_memory_rss_bytes.clone()),
261 boxed_collector(process_cpu_milliseconds_total.clone()),
262 boxed_collector(uptime_seconds.clone()),
263 #[cfg(feature = "allocator-metrics")]
264 boxed_collector(process_heap_memory_mib.clone()),
265 ];
266
267 for collector in collectors {
268 registry.register(collector)?;
269 }
270
271 Ok(Self {
272 registry,
273 http_requests_total,
274 http_request_duration_seconds,
275 db_queries_total,
276 db_query_duration_seconds,
277 auth_events_total,
278 application_events_total,
279 config_reloads_total,
280 config_reload_duration_seconds,
281 config_reload_changed_keys,
282 config_mutations_total,
283 config_mutation_changed_keys,
284 background_tasks_total,
285 background_tasks_pending,
286 background_task_retries_total,
287 external_operations_total,
288 external_operation_duration_seconds,
289 health_report_status,
290 health_report_duration_seconds,
291 health_component_status,
292 health_component_duration_seconds,
293 process_memory_rss_bytes,
294 process_cpu_milliseconds_total,
295 uptime_seconds,
296 #[cfg(feature = "allocator-metrics")]
297 process_heap_memory_mib,
298 product_metrics: Mutex::new(ProductMetricRegistry::default()),
299 })
300 }
301
302 fn export(&self) -> Result<String, String> {
303 self.refresh_allocator_metrics();
304 let encoder = TextEncoder::new();
305 let metric_families = self.registry.gather();
306 let mut buf = Vec::new();
307 encoder
308 .encode(&metric_families, &mut buf)
309 .map_err(|error| error.to_string())?;
310 String::from_utf8(buf).map_err(|error| error.to_string())
311 }
312
313 #[cfg(feature = "allocator-metrics")]
314 fn refresh_allocator_metrics(&self) {
315 let (allocated_mib, peak_or_resident_mib) = aster_forge_alloc::stats();
316 self.process_heap_memory_mib
317 .with_label_values(&["allocated"])
318 .set(allocated_mib);
319 self.process_heap_memory_mib
320 .with_label_values(&["peak_or_resident"])
321 .set(peak_or_resident_mib);
322 }
323
324 #[cfg(not(feature = "allocator-metrics"))]
325 fn refresh_allocator_metrics(&self) {}
326}
327
328#[derive(Default)]
329struct ProductMetricRegistry {
330 collectors: BTreeMap<ProductMetricKey, ProductMetricCollector>,
331}
332
333#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
334struct ProductMetricKey {
335 subsystem: &'static str,
336 name: &'static str,
337}
338
339#[derive(Clone)]
340enum ProductMetricCollector {
341 Counter {
342 labels: &'static [&'static str],
343 collector: IntCounterVec,
344 },
345 Gauge {
346 labels: &'static [&'static str],
347 collector: GaugeVec,
348 },
349 Histogram {
350 labels: &'static [&'static str],
351 collector: HistogramVec,
352 },
353}
354
355impl ProductMetricCollector {
356 fn kind(&self) -> MetricKind {
357 match self {
358 Self::Counter { .. } => MetricKind::Counter,
359 Self::Gauge { .. } => MetricKind::Gauge,
360 Self::Histogram { .. } => MetricKind::Histogram,
361 }
362 }
363
364 fn label_count(&self) -> usize {
365 match self {
366 Self::Counter { labels, .. }
367 | Self::Gauge { labels, .. }
368 | Self::Histogram { labels, .. } => labels.len(),
369 }
370 }
371}
372
373#[derive(Debug, Clone, Copy, PartialEq, Eq)]
375pub struct ProductMetricHandle {
376 subsystem: &'static str,
377 name: &'static str,
378 kind: MetricKind,
379 label_count: usize,
380}
381
382impl ProductMetricHandle {
383 const fn new(descriptor: &MetricDescriptor) -> Self {
385 Self {
386 subsystem: descriptor.subsystem,
387 name: descriptor.name,
388 kind: descriptor.kind,
389 label_count: descriptor.labels.len(),
390 }
391 }
392
393 #[must_use]
395 pub const fn subsystem(&self) -> &'static str {
396 self.subsystem
397 }
398
399 #[must_use]
401 pub const fn name(&self) -> &'static str {
402 self.name
403 }
404
405 #[must_use]
407 pub const fn kind(&self) -> MetricKind {
408 self.kind
409 }
410
411 #[must_use]
413 pub const fn label_count(&self) -> usize {
414 self.label_count
415 }
416}
417
418#[derive(Debug, Clone, Copy, PartialEq, Eq)]
420pub struct ProductCounter {
421 handle: ProductMetricHandle,
422}
423
424impl ProductCounter {
425 pub fn from_handle(handle: ProductMetricHandle) -> ProductMetricResult<Self> {
431 typed_product_metric_handle(handle, MetricKind::Counter).map(|handle| Self { handle })
432 }
433
434 #[must_use]
436 pub const fn handle(&self) -> ProductMetricHandle {
437 self.handle
438 }
439
440 pub fn inc(&self, label_values: &[&str], value: u64) {
442 if let Err(error) = self.try_inc(label_values, value) {
443 log_product_metric_error(&error);
444 }
445 }
446
447 pub fn try_inc(&self, label_values: &[&str], value: u64) -> ProductMetricResult<()> {
453 inc_product_counter(self.handle, label_values, value)
454 }
455}
456
457#[derive(Debug, Clone, Copy, PartialEq, Eq)]
459pub struct ProductGauge {
460 handle: ProductMetricHandle,
461}
462
463impl ProductGauge {
464 pub fn from_handle(handle: ProductMetricHandle) -> ProductMetricResult<Self> {
470 typed_product_metric_handle(handle, MetricKind::Gauge).map(|handle| Self { handle })
471 }
472
473 #[must_use]
475 pub const fn handle(&self) -> ProductMetricHandle {
476 self.handle
477 }
478
479 pub fn set(&self, label_values: &[&str], value: f64) {
481 if let Err(error) = self.try_set(label_values, value) {
482 log_product_metric_error(&error);
483 }
484 }
485
486 pub fn try_set(&self, label_values: &[&str], value: f64) -> ProductMetricResult<()> {
492 set_product_gauge(self.handle, label_values, value)
493 }
494
495 pub fn add(&self, label_values: &[&str], value: f64) {
497 if let Err(error) = self.try_add(label_values, value) {
498 log_product_metric_error(&error);
499 }
500 }
501
502 pub fn try_add(&self, label_values: &[&str], value: f64) -> ProductMetricResult<()> {
508 add_product_gauge(self.handle, label_values, value)
509 }
510}
511
512#[derive(Debug, Clone, Copy, PartialEq, Eq)]
514pub struct ProductHistogram {
515 handle: ProductMetricHandle,
516}
517
518impl ProductHistogram {
519 pub fn from_handle(handle: ProductMetricHandle) -> ProductMetricResult<Self> {
525 typed_product_metric_handle(handle, MetricKind::Histogram).map(|handle| Self { handle })
526 }
527
528 #[must_use]
530 pub const fn handle(&self) -> ProductMetricHandle {
531 self.handle
532 }
533
534 pub fn observe(&self, label_values: &[&str], value: f64) {
536 if let Err(error) = self.try_observe(label_values, value) {
537 log_product_metric_error(&error);
538 }
539 }
540
541 pub fn try_observe(&self, label_values: &[&str], value: f64) -> ProductMetricResult<()> {
547 observe_product_histogram(self.handle, label_values, value)
548 }
549}
550
551fn log_product_metric_error(error: &ProductMetricError) {
552 tracing::warn!(error = %error, "failed to record product metric");
553}
554
555fn typed_product_metric_handle(
556 handle: ProductMetricHandle,
557 expected_kind: MetricKind,
558) -> ProductMetricResult<ProductMetricHandle> {
559 if handle.kind != expected_kind {
560 return Err(ProductMetricError::WrongKind {
561 subsystem: handle.subsystem,
562 name: handle.name,
563 actual: handle.kind,
564 expected: expected_kind,
565 });
566 }
567
568 Ok(handle)
569}
570
571#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
573pub enum ProductMetricError {
574 #[error("metrics registry is not initialized")]
576 NotInitialized,
577 #[error("duplicate product metric registration: {subsystem}.{name}")]
579 DuplicateRegistration {
580 subsystem: &'static str,
582 name: &'static str,
584 },
585 #[error("unknown product metric: {subsystem}.{name}")]
587 UnknownMetric {
588 subsystem: &'static str,
590 name: &'static str,
592 },
593 #[error("product metric {subsystem}.{name} has kind {actual:?}, expected {expected:?}")]
595 WrongKind {
596 subsystem: &'static str,
598 name: &'static str,
600 actual: MetricKind,
602 expected: MetricKind,
604 },
605 #[error(
607 "product metric {subsystem}.{name} expected {expected} label values, received {actual}"
608 )]
609 LabelCountMismatch {
610 subsystem: &'static str,
612 name: &'static str,
614 expected: usize,
616 actual: usize,
618 },
619 #[error("prometheus metric error: {0}")]
621 Prometheus(String),
622 #[error("product metric registry lock is poisoned")]
624 LockPoisoned,
625}
626
627pub type ProductMetricResult<T> = std::result::Result<T, ProductMetricError>;
629
630pub fn init_metrics() -> Result<(), prometheus::Error> {
636 if METRICS.get().is_some() {
637 return Ok(());
638 }
639
640 let _ = PROCESS_STARTED_AT.get_or_init(Instant::now);
641 let metrics = PrometheusMetrics::new()?;
642 let _ = METRICS.set(metrics);
643 Ok(())
644}
645
646pub fn init_or_noop() -> SharedMetricsRecorder {
648 crate::init_metrics_or_noop(init_metrics, || PrometheusMetricsRecorder)
649}
650
651pub fn is_initialized() -> bool {
653 METRICS.get().is_some()
654}
655
656pub fn export_metrics() -> Result<String, String> {
662 let metrics = METRICS
663 .get()
664 .ok_or_else(|| "metrics registry is not initialized".to_string())?;
665 metrics.export()
666}
667
668pub fn register_product_metric(
678 descriptor: MetricDescriptor,
679) -> ProductMetricResult<ProductMetricHandle> {
680 let metrics = METRICS.get().ok_or(ProductMetricError::NotInitialized)?;
681 let key = ProductMetricKey {
682 subsystem: descriptor.subsystem,
683 name: descriptor.name,
684 };
685 let handle = ProductMetricHandle::new(&descriptor);
686 let collector = build_product_metric_collector(&descriptor)?;
687 let mut product_metrics = metrics
688 .product_metrics
689 .lock()
690 .map_err(|_| ProductMetricError::LockPoisoned)?;
691 if product_metrics.collectors.contains_key(&key) {
692 return Err(ProductMetricError::DuplicateRegistration {
693 subsystem: descriptor.subsystem,
694 name: descriptor.name,
695 });
696 }
697 metrics
698 .registry
699 .register(product_collector_box(&collector))
700 .map_err(|error| ProductMetricError::Prometheus(error.to_string()))?;
701 product_metrics.collectors.insert(key, collector);
702 Ok(handle)
703}
704
705pub fn register_product_metrics<I>(descriptors: I) -> ProductMetricResult<Vec<ProductMetricHandle>>
715where
716 I: IntoIterator<Item = MetricDescriptor>,
717{
718 let mut handles = Vec::new();
719 for descriptor in descriptors {
720 match register_product_metric(descriptor) {
721 Ok(handle) => handles.push(handle),
722 Err(error) => {
723 for handle in &handles {
724 if let Err(rollback_error) =
725 unregister_product_metric(handle.subsystem(), handle.name())
726 {
727 tracing::warn!(
728 error = %rollback_error,
729 subsystem = handle.subsystem(),
730 name = handle.name(),
731 "product metric rollback after batch registration failure failed"
732 );
733 }
734 }
735 return Err(error);
736 }
737 }
738 }
739 Ok(handles)
740}
741
742fn unregister_product_metric(
747 subsystem: &'static str,
748 name: &'static str,
749) -> ProductMetricResult<()> {
750 let metrics = METRICS.get().ok_or(ProductMetricError::NotInitialized)?;
751 let key = ProductMetricKey { subsystem, name };
752 let mut product_metrics = metrics
753 .product_metrics
754 .lock()
755 .map_err(|_| ProductMetricError::LockPoisoned)?;
756 let Some(collector) = product_metrics.collectors.remove(&key) else {
757 return Ok(());
758 };
759 metrics
760 .registry
761 .unregister(product_collector_box(&collector))
762 .map_err(|error| ProductMetricError::Prometheus(error.to_string()))
763}
764
765pub fn register_product_counter(
771 descriptor: MetricDescriptor,
772) -> ProductMetricResult<ProductCounter> {
773 ensure_descriptor_kind(&descriptor, MetricKind::Counter)?;
774 ProductCounter::from_handle(register_product_metric(descriptor)?)
775}
776
777pub fn register_product_gauge(descriptor: MetricDescriptor) -> ProductMetricResult<ProductGauge> {
783 ensure_descriptor_kind(&descriptor, MetricKind::Gauge)?;
784 ProductGauge::from_handle(register_product_metric(descriptor)?)
785}
786
787pub fn register_product_histogram(
793 descriptor: MetricDescriptor,
794) -> ProductMetricResult<ProductHistogram> {
795 ensure_descriptor_kind(&descriptor, MetricKind::Histogram)?;
796 ProductHistogram::from_handle(register_product_metric(descriptor)?)
797}
798
799fn ensure_descriptor_kind(
800 descriptor: &MetricDescriptor,
801 expected_kind: MetricKind,
802) -> ProductMetricResult<()> {
803 if descriptor.kind != expected_kind {
804 return Err(ProductMetricError::WrongKind {
805 subsystem: descriptor.subsystem,
806 name: descriptor.name,
807 actual: descriptor.kind,
808 expected: expected_kind,
809 });
810 }
811
812 Ok(())
813}
814
815fn build_product_metric_collector(
816 descriptor: &MetricDescriptor,
817) -> ProductMetricResult<ProductMetricCollector> {
818 match descriptor.kind {
819 MetricKind::Counter => {
820 let metric_name = product_metric_name(descriptor);
821 let collector =
822 IntCounterVec::new(Opts::new(metric_name, descriptor.help), descriptor.labels)
823 .map_err(|error| ProductMetricError::Prometheus(error.to_string()))?;
824 Ok(ProductMetricCollector::Counter {
825 labels: descriptor.labels,
826 collector,
827 })
828 }
829 MetricKind::Gauge => {
830 let metric_name = product_metric_name(descriptor);
831 let collector =
832 GaugeVec::new(Opts::new(metric_name, descriptor.help), descriptor.labels)
833 .map_err(|error| ProductMetricError::Prometheus(error.to_string()))?;
834 Ok(ProductMetricCollector::Gauge {
835 labels: descriptor.labels,
836 collector,
837 })
838 }
839 MetricKind::Histogram => {
840 let metric_name = product_metric_name(descriptor);
841 let mut opts = HistogramOpts::new(metric_name, descriptor.help);
842 if !descriptor.buckets.is_empty() {
843 opts = opts.buckets(descriptor.buckets.to_vec());
844 }
845 let collector = HistogramVec::new(opts, descriptor.labels)
846 .map_err(|error| ProductMetricError::Prometheus(error.to_string()))?;
847 Ok(ProductMetricCollector::Histogram {
848 labels: descriptor.labels,
849 collector,
850 })
851 }
852 }
853}
854
855fn product_metric_name(descriptor: &MetricDescriptor) -> String {
856 format!("{}_{}", descriptor.subsystem, descriptor.name)
857}
858
859fn product_collector_box(
860 collector: &ProductMetricCollector,
861) -> Box<dyn prometheus::core::Collector> {
862 match collector {
863 ProductMetricCollector::Counter { collector, .. } => Box::new(collector.clone()),
864 ProductMetricCollector::Gauge { collector, .. } => Box::new(collector.clone()),
865 ProductMetricCollector::Histogram { collector, .. } => Box::new(collector.clone()),
866 }
867}
868
869pub fn inc_product_counter(
875 handle: ProductMetricHandle,
876 label_values: &[&str],
877 value: u64,
878) -> ProductMetricResult<()> {
879 with_product_metric(handle, MetricKind::Counter, label_values, |collector| {
880 let ProductMetricCollector::Counter { collector, .. } = collector else {
881 return;
882 };
883 collector.with_label_values(label_values).inc_by(value);
884 })
885}
886
887pub fn set_product_gauge(
893 handle: ProductMetricHandle,
894 label_values: &[&str],
895 value: f64,
896) -> ProductMetricResult<()> {
897 with_product_metric(handle, MetricKind::Gauge, label_values, |collector| {
898 let ProductMetricCollector::Gauge { collector, .. } = collector else {
899 return;
900 };
901 collector.with_label_values(label_values).set(value);
902 })
903}
904
905pub fn add_product_gauge(
911 handle: ProductMetricHandle,
912 label_values: &[&str],
913 value: f64,
914) -> ProductMetricResult<()> {
915 with_product_metric(handle, MetricKind::Gauge, label_values, |collector| {
916 let ProductMetricCollector::Gauge { collector, .. } = collector else {
917 return;
918 };
919 collector.with_label_values(label_values).add(value);
920 })
921}
922
923pub fn observe_product_histogram(
929 handle: ProductMetricHandle,
930 label_values: &[&str],
931 value: f64,
932) -> ProductMetricResult<()> {
933 with_product_metric(handle, MetricKind::Histogram, label_values, |collector| {
934 let ProductMetricCollector::Histogram { collector, .. } = collector else {
935 return;
936 };
937 collector.with_label_values(label_values).observe(value);
938 })
939}
940
941fn with_product_metric<F>(
942 handle: ProductMetricHandle,
943 expected_kind: MetricKind,
944 label_values: &[&str],
945 record: F,
946) -> ProductMetricResult<()>
947where
948 F: FnOnce(&ProductMetricCollector),
949{
950 let metrics = METRICS.get().ok_or(ProductMetricError::NotInitialized)?;
951 let product_metrics = metrics
952 .product_metrics
953 .lock()
954 .map_err(|_| ProductMetricError::LockPoisoned)?;
955 let key = ProductMetricKey {
956 subsystem: handle.subsystem,
957 name: handle.name,
958 };
959 let collector =
960 product_metrics
961 .collectors
962 .get(&key)
963 .ok_or(ProductMetricError::UnknownMetric {
964 subsystem: handle.subsystem,
965 name: handle.name,
966 })?;
967 if collector.kind() != expected_kind || handle.kind != expected_kind {
968 return Err(ProductMetricError::WrongKind {
969 subsystem: handle.subsystem,
970 name: handle.name,
971 actual: collector.kind(),
972 expected: expected_kind,
973 });
974 }
975 let expected_labels = collector.label_count();
976 if expected_labels != label_values.len() || handle.label_count != label_values.len() {
977 return Err(ProductMetricError::LabelCountMismatch {
978 subsystem: handle.subsystem,
979 name: handle.name,
980 expected: expected_labels,
981 actual: label_values.len(),
982 });
983 }
984
985 record(collector);
986 Ok(())
987}
988
989#[macro_export]
1017macro_rules! product_metrics {
1018 (
1019 $(#[$struct_meta:meta])*
1020 $vis:vis struct $name:ident {
1021 $(
1022 $(#[$field_meta:meta])*
1023 $field:ident : $kind:ident (
1024 $subsystem:expr,
1025 $metric_name:expr,
1026 $help:expr,
1027 $labels:expr
1028 $(, $buckets:expr)?
1029 $(,)?
1030 )
1031 ),* $(,)?
1032 }
1033 ) => {
1034 $(#[$struct_meta])*
1035 $vis struct $name {
1036 $(
1037 $(#[$field_meta])*
1038 pub $field: $crate::product_metrics!(@field_type $kind),
1039 )*
1040 }
1041
1042 impl $name {
1043 pub fn register() -> $crate::prometheus::ProductMetricResult<Self> {
1045 Ok(Self {
1046 $(
1047 $field: $crate::product_metrics!(
1048 @register
1049 $kind,
1050 $subsystem,
1051 $metric_name,
1052 $help,
1053 $labels
1054 $(, $buckets)?
1055 )?,
1056 )*
1057 })
1058 }
1059 }
1060 };
1061 (@field_type counter) => {
1062 $crate::prometheus::ProductCounter
1063 };
1064 (@field_type gauge) => {
1065 $crate::prometheus::ProductGauge
1066 };
1067 (@field_type histogram) => {
1068 $crate::prometheus::ProductHistogram
1069 };
1070 (@field_type histogram_with_buckets) => {
1071 $crate::prometheus::ProductHistogram
1072 };
1073 (@register counter, $subsystem:expr, $metric_name:expr, $help:expr, $labels:expr) => {
1074 $crate::prometheus::register_product_counter($crate::MetricDescriptor::counter(
1075 $subsystem,
1076 $metric_name,
1077 $help,
1078 $labels,
1079 ))
1080 };
1081 (@register gauge, $subsystem:expr, $metric_name:expr, $help:expr, $labels:expr) => {
1082 $crate::prometheus::register_product_gauge($crate::MetricDescriptor::gauge(
1083 $subsystem,
1084 $metric_name,
1085 $help,
1086 $labels,
1087 ))
1088 };
1089 (@register histogram, $subsystem:expr, $metric_name:expr, $help:expr, $labels:expr) => {
1090 $crate::prometheus::register_product_histogram($crate::MetricDescriptor::histogram(
1091 $subsystem,
1092 $metric_name,
1093 $help,
1094 $labels,
1095 ))
1096 };
1097 (@register histogram_with_buckets, $subsystem:expr, $metric_name:expr, $help:expr, $labels:expr, $buckets:expr) => {
1098 $crate::prometheus::register_product_histogram(
1099 $crate::MetricDescriptor::histogram_with_buckets(
1100 $subsystem,
1101 $metric_name,
1102 $help,
1103 $labels,
1104 $buckets,
1105 ),
1106 )
1107 };
1108}
1109
1110#[derive(Debug, Clone, Copy, Default)]
1112pub struct PrometheusMetricsRecorder;
1113
1114impl DbMetricsRecorder for PrometheusMetricsRecorder {
1115 fn enabled(&self) -> bool {
1116 is_initialized()
1120 }
1121
1122 fn record_db_query(&self, metric: &DbQueryMetric) {
1123 record_db_query(metric);
1124 }
1125}
1126
1127impl MetricsRecorder for PrometheusMetricsRecorder {
1128 fn record_http_request(&self, method: &str, route: &str, status: u16, duration_seconds: f64) {
1129 record_http_request(method, route, status, duration_seconds);
1130 }
1131
1132 fn record_auth_event(&self, action: &'static str, status: &'static str, reason: &'static str) {
1133 record_auth_event(action, status, reason);
1134 }
1135
1136 fn record_application_event(
1137 &self,
1138 category: &'static str,
1139 event: &'static str,
1140 status: &'static str,
1141 ) {
1142 record_application_event(category, event, status);
1143 }
1144
1145 fn record_config_reload(
1146 &self,
1147 source: &'static str,
1148 decision: &'static str,
1149 status: &'static str,
1150 changed_keys: u64,
1151 duration_seconds: f64,
1152 ) {
1153 record_config_reload(source, decision, status, changed_keys, duration_seconds);
1154 }
1155
1156 fn record_config_mutation(
1157 &self,
1158 source: &'static str,
1159 operation: &'static str,
1160 status: &'static str,
1161 changed_keys: u64,
1162 ) {
1163 record_config_mutation(source, operation, status, changed_keys);
1164 }
1165
1166 fn record_background_task_transition(&self, kind: &'static str, status: &'static str) {
1167 record_background_task_transition(kind, status);
1168 }
1169
1170 fn set_background_tasks_pending(&self, pending: u64) {
1171 set_background_tasks_pending(pending);
1172 }
1173
1174 fn record_external_operation(
1175 &self,
1176 system: &'static str,
1177 operation: &'static str,
1178 status: &'static str,
1179 duration_seconds: f64,
1180 ) {
1181 record_external_operation(system, operation, status, duration_seconds);
1182 }
1183
1184 fn system_metrics_updater_task(
1185 &self,
1186 shutdown_token: CancellationToken,
1187 ) -> Option<Pin<Box<dyn Future<Output = ()> + Send + 'static>>> {
1188 Some(Box::pin(system_metrics_updater_task(shutdown_token)))
1189 }
1190}
1191
1192#[cfg(feature = "runtime-health")]
1193impl aster_forge_runtime::HealthMetricsRecorder for PrometheusMetricsRecorder {
1194 fn record_health_report(
1195 &self,
1196 scope: &'static str,
1197 status: aster_forge_runtime::HealthStatus,
1198 duration_seconds: f64,
1199 ) {
1200 record_health_report(
1201 scope,
1202 status.as_str(),
1203 health_status_value(status.as_str()),
1204 duration_seconds,
1205 );
1206 }
1207
1208 fn record_health_component(
1209 &self,
1210 scope: &'static str,
1211 component: &aster_forge_runtime::HealthComponentReport,
1212 duration_seconds: f64,
1213 ) {
1214 record_health_component(
1215 scope,
1216 component.name,
1217 component.status.as_str(),
1218 health_status_value(component.status.as_str()),
1219 duration_seconds,
1220 );
1221 }
1222}
1223
1224fn record_http_request(method: &str, route: &str, status: u16, duration_seconds: f64) {
1225 let Some(metrics) = METRICS.get() else {
1226 return;
1227 };
1228
1229 let status = status.to_string();
1230 metrics
1231 .http_requests_total
1232 .with_label_values(&[method, route, &status])
1233 .inc();
1234 metrics
1235 .http_request_duration_seconds
1236 .with_label_values(&[method, route, &status])
1237 .observe(duration_seconds);
1238}
1239
1240fn record_db_query(metric: &DbQueryMetric) {
1241 let Some(metrics) = METRICS.get() else {
1242 return;
1243 };
1244
1245 let backend = metric.backend.as_label();
1246 let kind = metric.kind.as_label();
1247 let status = metric.status_label();
1248
1249 metrics
1250 .db_queries_total
1251 .with_label_values(&[backend, kind, status])
1252 .inc();
1253 metrics
1254 .db_query_duration_seconds
1255 .with_label_values(&[backend, kind, status])
1256 .observe(metric.elapsed.as_secs_f64());
1257}
1258
1259fn record_auth_event(action: &'static str, status: &'static str, reason: &'static str) {
1260 let Some(metrics) = METRICS.get() else {
1261 return;
1262 };
1263
1264 metrics
1265 .auth_events_total
1266 .with_label_values(&[action, status, reason])
1267 .inc();
1268}
1269
1270fn record_application_event(category: &'static str, event: &'static str, status: &'static str) {
1271 let Some(metrics) = METRICS.get() else {
1272 return;
1273 };
1274
1275 metrics
1276 .application_events_total
1277 .with_label_values(&[category, event, status])
1278 .inc();
1279}
1280
1281#[expect(
1282 clippy::cast_precision_loss,
1283 reason = "Prometheus gauges and histograms use f64 samples; large integer observations are diagnostic approximations."
1284)]
1285fn prometheus_u64(value: u64) -> f64 {
1286 value as f64
1287}
1288
1289fn record_config_reload(
1290 source: &'static str,
1291 decision: &'static str,
1292 status: &'static str,
1293 changed_keys: u64,
1294 duration_seconds: f64,
1295) {
1296 let Some(metrics) = METRICS.get() else {
1297 return;
1298 };
1299
1300 metrics
1301 .config_reloads_total
1302 .with_label_values(&[source, decision, status])
1303 .inc();
1304 metrics
1305 .config_reload_duration_seconds
1306 .with_label_values(&[source, decision, status])
1307 .observe(duration_seconds);
1308 metrics
1309 .config_reload_changed_keys
1310 .with_label_values(&[source, decision, status])
1311 .observe(prometheus_u64(changed_keys));
1312}
1313
1314fn record_config_mutation(
1315 source: &'static str,
1316 operation: &'static str,
1317 status: &'static str,
1318 changed_keys: u64,
1319) {
1320 let Some(metrics) = METRICS.get() else {
1321 return;
1322 };
1323
1324 metrics
1325 .config_mutations_total
1326 .with_label_values(&[source, operation, status])
1327 .inc();
1328 metrics
1329 .config_mutation_changed_keys
1330 .with_label_values(&[source, operation, status])
1331 .observe(prometheus_u64(changed_keys));
1332}
1333
1334fn record_background_task_transition(kind: &'static str, status: &'static str) {
1335 let Some(metrics) = METRICS.get() else {
1336 return;
1337 };
1338
1339 metrics
1340 .background_tasks_total
1341 .with_label_values(&[kind, status])
1342 .inc();
1343 if status == "retry" {
1344 metrics
1345 .background_task_retries_total
1346 .with_label_values(&[kind])
1347 .inc();
1348 }
1349}
1350
1351fn set_background_tasks_pending(pending: u64) {
1352 let Some(metrics) = METRICS.get() else {
1353 return;
1354 };
1355
1356 metrics
1357 .background_tasks_pending
1358 .set(i64::try_from(pending).unwrap_or(i64::MAX));
1359}
1360
1361fn record_external_operation(
1362 system: &'static str,
1363 operation: &'static str,
1364 status: &'static str,
1365 duration_seconds: f64,
1366) {
1367 let Some(metrics) = METRICS.get() else {
1368 return;
1369 };
1370
1371 metrics
1372 .external_operations_total
1373 .with_label_values(&[system, operation, status])
1374 .inc();
1375 metrics
1376 .external_operation_duration_seconds
1377 .with_label_values(&[system, operation, status])
1378 .observe(duration_seconds);
1379}
1380
1381pub fn record_health_report(
1383 scope: &'static str,
1384 status_label: &'static str,
1385 status_value: f64,
1386 duration_seconds: f64,
1387) {
1388 let Some(metrics) = METRICS.get() else {
1389 return;
1390 };
1391
1392 metrics
1393 .health_report_status
1394 .with_label_values(&[scope])
1395 .set(status_value);
1396 metrics
1397 .health_report_duration_seconds
1398 .with_label_values(&[scope, status_label])
1399 .observe(duration_seconds);
1400}
1401
1402pub fn record_health_component(
1404 scope: &'static str,
1405 component: &'static str,
1406 status_label: &'static str,
1407 status_value: f64,
1408 duration_seconds: f64,
1409) {
1410 let Some(metrics) = METRICS.get() else {
1411 return;
1412 };
1413
1414 metrics
1415 .health_component_status
1416 .with_label_values(&[scope, component])
1417 .set(status_value);
1418 metrics
1419 .health_component_duration_seconds
1420 .with_label_values(&[scope, component, status_label])
1421 .observe(duration_seconds);
1422}
1423
1424#[cfg(feature = "runtime-health")]
1425fn health_status_value(status: &'static str) -> f64 {
1426 match status {
1427 "healthy" => 0.0,
1428 "degraded" => 1.0,
1429 _ => 2.0,
1430 }
1431}
1432
1433async fn system_metrics_updater_task(shutdown_token: CancellationToken) {
1434 use std::sync::Mutex;
1435 use sysinfo::{Pid, ProcessRefreshKind, ProcessesToUpdate, System};
1436
1437 static SYSTEM: OnceLock<Mutex<System>> = OnceLock::new();
1438
1439 let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
1440 loop {
1441 tokio::select! {
1442 biased;
1443 () = shutdown_token.cancelled() => break,
1444 _ = interval.tick() => {}
1445 }
1446
1447 if shutdown_token.is_cancelled() {
1448 break;
1449 }
1450
1451 let Some(metrics) = METRICS.get() else {
1452 continue;
1453 };
1454
1455 let update = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1456 let pid = Pid::from_u32(std::process::id());
1457 let sys_mutex = SYSTEM.get_or_init(|| Mutex::new(System::new()));
1458 let Ok(mut sys) = sys_mutex.lock() else {
1459 tracing::warn!("system metrics updater lock is poisoned");
1460 return;
1461 };
1462 sys.refresh_processes_specifics(
1463 ProcessesToUpdate::Some(&[pid]),
1464 true,
1465 ProcessRefreshKind::nothing().with_memory().with_cpu(),
1466 );
1467 if let Some(process) = sys.process(pid) {
1468 metrics
1469 .process_memory_rss_bytes
1470 .set(prometheus_u64(process.memory()));
1471 let cpu_millis = i64::try_from(process.accumulated_cpu_time()).unwrap_or(i64::MAX);
1472 metrics.process_cpu_milliseconds_total.set(cpu_millis);
1473 }
1474 let uptime = PROCESS_STARTED_AT
1475 .get()
1476 .map(Instant::elapsed)
1477 .unwrap_or_default()
1478 .as_secs_f64();
1479 metrics.uptime_seconds.set(uptime);
1480 metrics.refresh_allocator_metrics();
1481 }));
1482
1483 if let Err(panic) = update {
1484 tracing::error!(panic = %panic_message(panic.as_ref()), "system metrics updater panicked");
1485 }
1486 }
1487}
1488
1489fn panic_message(panic: &(dyn std::any::Any + Send)) -> String {
1490 if let Some(message) = panic.downcast_ref::<&str>() {
1491 (*message).to_string()
1492 } else if let Some(message) = panic.downcast_ref::<String>() {
1493 message.clone()
1494 } else {
1495 "unknown panic payload".to_string()
1496 }
1497}
1498
1499#[cfg(test)]
1500mod tests {
1501 use super::{
1502 ProductCounter, ProductGauge, ProductHistogram, ProductMetricError, ProductMetricHandle,
1503 PrometheusMetricsRecorder, export_metrics, inc_product_counter, init_metrics,
1504 is_initialized, observe_product_histogram, record_health_component, record_health_report,
1505 register_product_counter, register_product_gauge, register_product_histogram,
1506 register_product_metric, register_product_metrics, set_product_gauge,
1507 };
1508 use crate::{DbMetricsRecorder, MetricDescriptor, MetricKind, MetricsRecorder};
1509
1510 #[test]
1511 fn prometheus_recorder_exports_low_cardinality_metrics() {
1512 init_metrics().expect("metrics registry should initialize");
1513 let recorder = PrometheusMetricsRecorder;
1514
1515 recorder.record_http_request("GET", "/health", 200, 0.01);
1516 recorder.record_config_reload("pubsub", "reloaded", "ok", 3, 0.02);
1517 recorder.record_config_mutation("api", "upsert", "error", 1);
1518 record_health_report("diagnostics", "degraded", 1.0, 0.25);
1519 record_health_component("diagnostics", "cache", "degraded", 1.0, 0.05);
1520
1521 let body = export_metrics().expect("metrics should export");
1522 assert!(recorder.enabled());
1523 assert!(body.contains("http_requests_total"));
1524 assert!(body.contains("config_reloads_total"));
1525 assert!(body.contains("config_reload_duration_seconds_count"));
1526 assert!(body.contains("config_mutations_total"));
1527 assert!(body.contains("health_report_status"));
1528 assert!(body.contains("health_component_status"));
1529 assert!(body.contains("source=\"pubsub\""));
1530 assert!(body.contains("decision=\"reloaded\""));
1531 assert!(body.contains("operation=\"upsert\""));
1532 }
1533
1534 #[cfg(feature = "allocator-metrics")]
1535 #[test]
1536 fn allocator_metrics_export_heap_memory_kinds() {
1537 init_metrics().expect("metrics registry should initialize");
1538
1539 let body = export_metrics().expect("metrics should export");
1540
1541 assert!(body.contains("process_heap_memory_mib"));
1542 assert!(body.contains("kind=\"allocated\""));
1543 assert!(body.contains("kind=\"peak_or_resident\""));
1544 }
1545
1546 #[test]
1547 fn recorder_enabled_tracks_registry_initialization() {
1548 init_metrics().expect("metrics registry should initialize");
1549
1550 assert_eq!(PrometheusMetricsRecorder.enabled(), is_initialized());
1557 }
1558
1559 #[test]
1560 fn register_product_metrics_rolls_back_partial_registrations_on_failure() {
1561 init_metrics().expect("metrics registry should initialize");
1562 let descriptor = || {
1563 MetricDescriptor::counter(
1564 "batch_rollback_test",
1565 "events_total",
1566 "Batch rollback test events.",
1567 &[],
1568 )
1569 };
1570
1571 let error = register_product_metrics(vec![descriptor(), descriptor()])
1574 .expect_err("duplicate registration should fail the batch");
1575 assert!(matches!(
1576 error,
1577 ProductMetricError::DuplicateRegistration { .. }
1578 ));
1579
1580 register_product_metrics(vec![descriptor()])
1583 .expect("retry after rollback should register cleanly");
1584 }
1585
1586 #[test]
1587 fn product_metric_operations_fail_before_registration() {
1588 init_metrics().expect("metrics registry should initialize");
1589 let handle = ProductMetricHandle {
1590 subsystem: "missing_product_metric_test",
1591 name: "events_total",
1592 kind: MetricKind::Counter,
1593 label_count: 0,
1594 };
1595
1596 let error = inc_product_counter(handle, &[], 1)
1597 .expect_err("unknown metric handle should be rejected");
1598
1599 assert_eq!(
1600 error,
1601 ProductMetricError::UnknownMetric {
1602 subsystem: "missing_product_metric_test",
1603 name: "events_total"
1604 }
1605 );
1606 }
1607
1608 #[test]
1609 fn typed_product_metric_handles_record_and_validate_kind() {
1610 init_metrics().expect("metrics registry should initialize");
1611
1612 let counter = register_product_counter(MetricDescriptor::counter(
1613 "typed_product_metric_test",
1614 "events_total",
1615 "Typed product metric events.",
1616 &["status"],
1617 ))
1618 .expect("counter should register");
1619 let gauge = register_product_gauge(MetricDescriptor::gauge(
1620 "typed_product_metric_test",
1621 "queue_depth",
1622 "Typed product metric queue depth.",
1623 &["queue"],
1624 ))
1625 .expect("gauge should register");
1626 let histogram = register_product_histogram(MetricDescriptor::histogram_with_buckets(
1627 "typed_product_metric_test",
1628 "duration_seconds",
1629 "Typed product metric duration.",
1630 &["kind"],
1631 &[0.1, 1.0],
1632 ))
1633 .expect("histogram should register");
1634
1635 counter.try_inc(&["ok"], 2).expect("counter should record");
1636 counter.inc(&["ok"], 1);
1637 gauge.try_set(&["mail"], 4.0).expect("gauge should set");
1638 gauge.add(&["mail"], 1.0);
1639 histogram
1640 .try_observe(&["dispatch"], 0.2)
1641 .expect("histogram should observe");
1642 histogram.observe(&["dispatch"], 0.3);
1643
1644 let wrong_kind = ProductGauge::from_handle(counter.handle())
1645 .expect_err("counter handle should not become a gauge");
1646 assert_eq!(
1647 wrong_kind,
1648 ProductMetricError::WrongKind {
1649 subsystem: "typed_product_metric_test",
1650 name: "events_total",
1651 actual: MetricKind::Counter,
1652 expected: MetricKind::Gauge
1653 }
1654 );
1655 assert!(ProductCounter::from_handle(counter.handle()).is_ok());
1656 assert!(ProductHistogram::from_handle(histogram.handle()).is_ok());
1657
1658 let body = export_metrics().expect("metrics should export");
1659 assert!(body.contains("typed_product_metric_test_events_total"));
1660 assert!(body.contains("typed_product_metric_test_queue_depth"));
1661 assert!(body.contains("typed_product_metric_test_duration_seconds_bucket"));
1662 }
1663
1664 #[test]
1665 fn product_metrics_macro_registers_typed_metric_set() {
1666 init_metrics().expect("metrics registry should initialize");
1667
1668 crate::product_metrics! {
1669 #[derive(Clone, Copy)]
1670 pub struct MacroProductMetrics {
1671 requests: counter(
1673 "macro_product_metric_test",
1674 "requests_total",
1675 "Macro product metric requests.",
1676 &["status"],
1677 ),
1678 queue_depth: gauge(
1679 "macro_product_metric_test",
1680 "queue_depth",
1681 "Macro product metric queue depth.",
1682 &["queue"],
1683 ),
1684 latency: histogram_with_buckets(
1685 "macro_product_metric_test",
1686 "latency_seconds",
1687 "Macro product metric latency.",
1688 &["route"],
1689 &[0.01, 0.1],
1690 ),
1691 }
1692 }
1693
1694 let metrics = MacroProductMetrics::register().expect("metric set should register");
1695 metrics.requests.inc(&["ok"], 1);
1696 metrics.queue_depth.set(&["mail"], 3.0);
1697 metrics.latency.observe(&["/healthz"], 0.02);
1698
1699 let body = export_metrics().expect("metrics should export");
1700 assert!(body.contains("macro_product_metric_test_requests_total"));
1701 assert!(body.contains("macro_product_metric_test_queue_depth"));
1702 assert!(body.contains("macro_product_metric_test_latency_seconds_bucket"));
1703 }
1704
1705 #[test]
1706 fn product_metrics_register_record_and_validate_boundaries() {
1707 init_metrics().expect("metrics registry should initialize");
1708
1709 let counter = register_product_metric(MetricDescriptor::counter(
1710 "product_registration_test",
1711 "events_total",
1712 "Product registration test events.",
1713 &["kind", "status"],
1714 ))
1715 .expect("counter should register");
1716 assert_eq!(counter.subsystem(), "product_registration_test");
1717 assert_eq!(counter.name(), "events_total");
1718 assert_eq!(counter.kind(), MetricKind::Counter);
1719 assert_eq!(counter.label_count(), 2);
1720
1721 let duplicate = register_product_metric(MetricDescriptor::counter(
1722 "product_registration_test",
1723 "events_total",
1724 "Duplicate product registration test events.",
1725 &["kind", "status"],
1726 ))
1727 .expect_err("duplicate should be rejected before touching prometheus registry");
1728 assert_eq!(
1729 duplicate,
1730 ProductMetricError::DuplicateRegistration {
1731 subsystem: "product_registration_test",
1732 name: "events_total"
1733 }
1734 );
1735
1736 let handles = register_product_metrics([
1737 MetricDescriptor::gauge(
1738 "product_registration_test",
1739 "queue_depth",
1740 "Product registration test queue depth.",
1741 &["queue"],
1742 ),
1743 MetricDescriptor::histogram_with_buckets(
1744 "product_registration_test",
1745 "job_duration_seconds",
1746 "Product registration test job duration.",
1747 &["kind"],
1748 &[0.1, 0.5, 1.0],
1749 ),
1750 ])
1751 .expect("gauge and histogram should register");
1752 assert_eq!(handles.len(), 2);
1753
1754 inc_product_counter(counter, &["dispatch", "ok"], 3)
1755 .expect("counter should record with matching labels");
1756 set_product_gauge(handles[0], &["mail"], 7.0).expect("gauge should record");
1757 observe_product_histogram(handles[1], &["dispatch"], 0.25)
1758 .expect("histogram should record");
1759
1760 let wrong_label_count = inc_product_counter(counter, &["dispatch"], 1)
1761 .expect_err("wrong label count should be rejected");
1762 assert_eq!(
1763 wrong_label_count,
1764 ProductMetricError::LabelCountMismatch {
1765 subsystem: "product_registration_test",
1766 name: "events_total",
1767 expected: 2,
1768 actual: 1
1769 }
1770 );
1771
1772 let wrong_kind = set_product_gauge(counter, &["dispatch", "ok"], 1.0)
1773 .expect_err("wrong recording kind should be rejected");
1774 assert_eq!(
1775 wrong_kind,
1776 ProductMetricError::WrongKind {
1777 subsystem: "product_registration_test",
1778 name: "events_total",
1779 actual: MetricKind::Counter,
1780 expected: MetricKind::Gauge
1781 }
1782 );
1783
1784 let body = export_metrics().expect("metrics should export");
1785 assert!(body.contains("product_registration_test_events_total"));
1786 assert!(body.contains("kind=\"dispatch\""));
1787 assert!(body.contains("status=\"ok\""));
1788 assert!(body.contains("product_registration_test_queue_depth"));
1789 assert!(body.contains("queue=\"mail\""));
1790 assert!(body.contains("product_registration_test_job_duration_seconds_bucket"));
1791 assert!(body.contains("le=\"0.5\""));
1792 }
1793}