camel_component_api/
runtime_observability.rs

1//! `RuntimeObservability` — narrow trait exposing `metrics()` and `health()`
2//! to Endpoint consumers/producers. Defined per ADR-0012 §Signal-replacement-API
3//! and the Phase A closure spec
4//! (`docs/superpowers/specs/2026-06-05-adr-0012-closure-design.md`).
5//!
6//! **Why a separate trait:** `ComponentContext` carries `resolve_component`,
7//! `resolve_language`, `platform_service` — far more authority than an
8//! emitter needs. Passing `Arc<dyn ComponentContext>` would be a
9//! service-locator anti-pattern inviting future misuse. This trait exposes
10//! exactly the two observability surfaces an ADR-0012 emitter calls.
11
12use std::sync::Arc;
13
14use camel_api::{ComponentMetrics, InFlightGauge, MetricsCollector};
15
16use crate::{ComponentContext, HealthCheckRegistry};
17
18/// Narrow observability surface available to Endpoint consumers and producers.
19///
20/// Implemented blanket for every `T: ComponentContext`, so
21/// `CamelContext` (the spec's "DefaultRuntime") and any test ctx
22/// automatically satisfy this trait. Endpoints receive
23/// `Arc<dyn RuntimeObservability>` at `create_consumer` / `create_producer`
24/// time (per Phase A closure spec).
25///
26/// `Send + Sync` is mandatory — Endpoints spawn across tokio tasks.
27pub trait RuntimeObservability: Send + Sync {
28    /// Active metrics collector. Used for `increment_errors(route_id, label)`
29    /// per ADR-0012 categories (b′) and (e).
30    fn metrics(&self) -> Arc<dyn MetricsCollector>;
31
32    /// Active health-check registry. Used for
33    /// `force_unhealthy_for_route(route_id, name, reason)` per ADR-0012
34    /// category (g).
35    fn health(&self) -> Arc<dyn HealthCheckRegistry>;
36
37    /// Facade for the uniform component-operations family
38    /// (`camel_component_operations_total{component,operation,outcome}`,
39    /// dashboard-observability Task 4.1), scoped to `component` so the
40    /// handle carries both the identity and the resolved switch.
41    ///
42    /// The per-component lever snapshot is baked in at facade
43    /// construction: the component series flow only when
44    /// `component_metric_levers().enabled_for(component)` is true, while
45    /// failures always reach the non-disableable error family — error
46    /// emission is never lever-gated.
47    ///
48    /// Default: lever off — impls that do not override
49    /// `component_metric_levers()` (manual test runtimes AND any context
50    /// wired without a levers snapshot) keep the component series
51    /// suppressed and only forward errors. In production the blanket impl
52    /// resolves the context's `MetricsLeversConfig` snapshot per
53    /// component.
54    fn component_metrics(&self, component: &str) -> ComponentMetrics {
55        ComponentMetrics::new(self.metrics(), component, false)
56    }
57
58    /// Context-global gauge of accepted-not-completed exchanges
59    /// (drainclaim), forwarded from [`ComponentContext::in_flight_counter`].
60    /// Default `None` keeps runtimes that install no gauge uncounted.
61    fn in_flight_counter(&self) -> Option<Arc<InFlightGauge>> {
62        None
63    }
64}
65
66// Blanket impl: every ComponentContext is automatically a RuntimeObservability.
67// This covers `CamelContext` (production), `NoOpComponentContext` (tests),
68// and any future impl.
69//
70// No `?Sized` bound: ComponentContext itself requires Sized (the trait has
71// methods returning `Arc<dyn …>` which need a concrete owner type). If a
72// future dynamically-sized ctx is needed, add a separate explicit impl.
73impl<T: ComponentContext> RuntimeObservability for T {
74    fn metrics(&self) -> Arc<dyn MetricsCollector> {
75        <Self as ComponentContext>::metrics(self)
76    }
77
78    fn health(&self) -> Arc<dyn HealthCheckRegistry> {
79        <Self as ComponentContext>::health(self)
80    }
81
82    fn component_metrics(&self, component: &str) -> ComponentMetrics {
83        let enabled =
84            <Self as ComponentContext>::component_metric_levers(self).enabled_for(component);
85        ComponentMetrics::new(
86            <Self as ComponentContext>::metrics(self),
87            component,
88            enabled,
89        )
90    }
91
92    fn in_flight_counter(&self) -> Option<Arc<InFlightGauge>> {
93        <Self as ComponentContext>::in_flight_counter(self)
94    }
95}
96
97#[cfg(test)]
98mod tests {
99    use std::collections::BTreeMap;
100    use std::sync::{Arc, Mutex};
101    use std::time::Duration;
102
103    use camel_api::{ComponentMetricsLevers, MetricsCollector, PlatformService};
104    use camel_language_api::Language;
105
106    use super::*;
107    use crate::{Component, NoOpComponentContext};
108
109    /// `MetricsCollector` capturing both emission families.
110    #[derive(Default)]
111    struct RecordingCollector {
112        errors: Mutex<Vec<(String, String)>>,
113        ops: Mutex<Vec<(String, String, String)>>,
114    }
115
116    impl RecordingCollector {
117        fn errors(&self) -> Vec<(String, String)> {
118            self.errors.lock().expect("errors lock").clone()
119        }
120
121        fn ops(&self) -> Vec<(String, String, String)> {
122            self.ops.lock().expect("ops lock").clone()
123        }
124    }
125
126    impl MetricsCollector for RecordingCollector {
127        fn record_exchange_duration(&self, _: &str, _: Duration) {}
128        fn increment_errors(&self, component: &str, error_type: &str) {
129            self.errors
130                .lock()
131                .expect("errors lock")
132                .push((component.to_string(), error_type.to_string()));
133        }
134        fn increment_exchanges(&self, _: &str) {}
135        fn set_queue_depth(&self, _: &str, _: usize) {}
136        fn record_circuit_breaker_change(&self, _: &str, _: &str, _: &str) {}
137        fn record_component_operation(&self, component: &str, operation: &str, outcome: &str) {
138            self.ops.lock().expect("ops lock").push((
139                component.to_string(),
140                operation.to_string(),
141                outcome.to_string(),
142            ));
143        }
144    }
145
146    /// Test context whose levers enable only `wasm`.
147    struct ScopedLeverContext {
148        inner: NoOpComponentContext,
149        collector: Arc<RecordingCollector>,
150    }
151
152    impl ComponentContext for ScopedLeverContext {
153        fn resolve_component(&self, scheme: &str) -> Option<Arc<dyn Component>> {
154            self.inner.resolve_component(scheme)
155        }
156
157        fn resolve_language(&self, name: &str) -> Option<Arc<dyn Language>> {
158            self.inner.resolve_language(name)
159        }
160
161        fn metrics(&self) -> Arc<dyn MetricsCollector> {
162            Arc::clone(&self.collector) as Arc<dyn MetricsCollector>
163        }
164
165        fn component_metric_levers(&self) -> ComponentMetricsLevers {
166            ComponentMetricsLevers::new(true, false, BTreeMap::from([("wasm".to_string(), true)]))
167        }
168
169        fn platform_service(&self) -> Arc<dyn PlatformService> {
170            self.inner.platform_service()
171        }
172
173        fn register_route_health_check(
174            &self,
175            route_id: &str,
176            check: Arc<dyn camel_api::AsyncHealthCheck>,
177        ) {
178            self.inner.register_route_health_check(route_id, check);
179        }
180
181        fn unregister_route_health_check(&self, route_id: &str) {
182            self.inner.unregister_route_health_check(route_id);
183        }
184    }
185
186    #[test]
187    fn runtime_observability_is_send_sync() {
188        fn assert_send_sync<T: Send + Sync + ?Sized>() {}
189        assert_send_sync::<dyn RuntimeObservability>();
190    }
191
192    #[test]
193    fn blanket_impl_uses_context_levers() {
194        let collector = Arc::new(RecordingCollector::default());
195        let ctx = ScopedLeverContext {
196            inner: NoOpComponentContext,
197            collector: Arc::clone(&collector),
198        };
199
200        RuntimeObservability::component_metrics(&ctx, "wasm").observe("invoke", false);
201        assert_eq!(
202            collector.ops(),
203            vec![(
204                "wasm".to_string(),
205                "invoke".to_string(),
206                "success".to_string()
207            )],
208            "the override must enable the wasm component series"
209        );
210
211        RuntimeObservability::component_metrics(&ctx, "kafka").observe("consume", true);
212        assert_eq!(
213            collector.ops().len(),
214            1,
215            "kafka inherits components=false, so no operation is recorded"
216        );
217        assert_eq!(
218            collector.errors(),
219            vec![("kafka".to_string(), "e:kafka:consume".to_string())],
220            "errors must flow regardless of the per-component lever"
221        );
222    }
223}
camel_component_api/runtime_observability.rs

camel_component_api/
runtime_observability.rs
