camel_api/
metrics.rs

1use std::time::Duration;
2
3/// Trait for collecting metrics from the Camel runtime.
4/// Implementations can integrate with Prometheus, OpenTelemetry, etc.
5pub trait MetricsCollector: Send + Sync {
6    /// Record exchange processing time
7    fn record_exchange_duration(&self, route_id: &str, duration: Duration);
8
9    /// Increment error counter
10    fn increment_errors(&self, route_id: &str, error_type: &str);
11
12    /// Increment exchange counter
13    fn increment_exchanges(&self, route_id: &str);
14
15    /// Update queue depth
16    fn set_queue_depth(&self, route_id: &str, depth: usize);
17
18    /// Record circuit breaker state change
19    fn record_circuit_breaker_change(&self, route_id: &str, from: &str, to: &str);
20
21    /// Record a histogram observation (e.g., cost, latency distribution).
22    /// Default: no-op (backward-compatible).
23    fn record_histogram(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
24
25    /// Record a monotonically-increasing counter (e.g. `foo_total`).
26    /// Default: no-op (backward-compatible).
27    fn record_counter(&self, _name: &str, _value: f64, _labels: &[(&str, &str)]) {}
28}
29
30/// No-op metrics collector for default behavior
31pub struct NoOpMetrics;
32
33impl MetricsCollector for NoOpMetrics {
34    fn record_exchange_duration(&self, _route_id: &str, _duration: Duration) {}
35    fn increment_errors(&self, _route_id: &str, _error_type: &str) {}
36    fn increment_exchanges(&self, _route_id: &str) {}
37    fn set_queue_depth(&self, _route_id: &str, _depth: usize) {}
38    fn record_circuit_breaker_change(&self, _route_id: &str, _from: &str, _to: &str) {}
39}
40
41#[cfg(test)]
42mod tests {
43    use super::*;
44    use std::sync::Arc;
45
46    #[test]
47    fn test_noop_metrics_implements_trait() {
48        let metrics = NoOpMetrics;
49        let metrics_arc: Arc<dyn MetricsCollector> = Arc::new(metrics);
50
51        // All methods should execute without panicking
52        metrics_arc.record_exchange_duration("test-route", Duration::from_millis(100));
53        metrics_arc.increment_errors("test-route", "test-error");
54        metrics_arc.increment_exchanges("test-route");
55        metrics_arc.set_queue_depth("test-route", 5);
56        metrics_arc.record_circuit_breaker_change("test-route", "closed", "open");
57    }
58
59    #[test]
60    fn test_custom_metrics_collector() {
61        struct TestMetrics {
62            exchange_count: std::sync::atomic::AtomicU64,
63        }
64
65        impl MetricsCollector for TestMetrics {
66            fn record_exchange_duration(&self, route_id: &str, duration: Duration) {
67                // In a real implementation, this would record the duration
68                println!("Route {} took {}ms", route_id, duration.as_millis());
69            }
70
71            fn increment_errors(&self, route_id: &str, error_type: &str) {
72                // In a real implementation, this would increment an error counter
73                println!("Route {} had error: {}", route_id, error_type);
74            }
75
76            fn increment_exchanges(&self, route_id: &str) {
77                // In a real implementation, this would increment an exchange counter
78                self.exchange_count
79                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
80                println!("Route {} processed exchange", route_id);
81            }
82
83            fn set_queue_depth(&self, route_id: &str, depth: usize) {
84                // In a real implementation, this would update a gauge
85                println!("Route {} queue depth: {}", route_id, depth);
86            }
87
88            fn record_circuit_breaker_change(&self, route_id: &str, from: &str, to: &str) {
89                // In a real implementation, this would record the state change
90                println!("Route {} circuit breaker: {} -> {}", route_id, from, to);
91            }
92        }
93
94        let test_metrics = TestMetrics {
95            exchange_count: std::sync::atomic::AtomicU64::new(0),
96        };
97        let metrics_arc: Arc<dyn MetricsCollector> = Arc::new(test_metrics);
98
99        // Test that all methods work
100        metrics_arc.record_exchange_duration("test-route", Duration::from_millis(100));
101        metrics_arc.increment_errors("test-route", "test-error");
102        metrics_arc.increment_exchanges("test-route");
103        metrics_arc.set_queue_depth("test-route", 5);
104        metrics_arc.record_circuit_breaker_change("test-route", "closed", "open");
105
106        // Note: We can't easily test the counter value without additional accessors
107        // This is just to verify the trait implementation works
108    }
109}
camel_api/metrics.rs

camel_api/
metrics.rs
