obs_core/
aggregator.rs

1use core_types::{HealthStatus, MetricsProvider, MetricsSnapshot};
2use serde_json::{Value, json};
3
4/// Health + metrics bundle for a single named component.
5#[derive(Clone, Debug)]
6pub struct ComponentHealth {
7    /// Component name as registered with the aggregator.
8    pub name: String,
9    /// Health status at the time of collection.
10    pub status: HealthStatus,
11    /// All metrics emitted by this component.
12    pub metrics: Vec<MetricsSnapshot>,
13}
14
15/// Overall system health report produced by [`MetricsAggregator`].
16#[derive(Clone, Debug)]
17pub struct HealthReport {
18    /// Per-component breakdown.
19    pub components: Vec<ComponentHealth>,
20    /// The worst status across all components (monotone escalation).
21    pub overall: HealthStatus,
22}
23
24impl HealthReport {
25    /// Returns `true` only when every component is `Healthy`.
26    pub fn is_fully_healthy(&self) -> bool {
27        matches!(self.overall, HealthStatus::Healthy)
28    }
29
30    /// Returns components whose health is not `Healthy`.
31    pub fn degraded_components(&self) -> impl Iterator<Item = &ComponentHealth> {
32        self.components.iter().filter(|c| !c.status.is_healthy())
33    }
34
35    pub fn to_json_value(&self) -> Value {
36        json!({
37            "schema": "robotrt.obs.health_report.v1",
38            "overall": health_status_to_json_value(&self.overall),
39            "components": self
40                .components
41                .iter()
42                .map(ComponentHealth::to_json_value)
43                .collect::<Vec<_>>()
44        })
45    }
46
47    pub fn to_json(&self) -> String {
48        self.to_json_value().to_string()
49    }
50
51    pub fn to_json_pretty(&self) -> Result<String, serde_json::Error> {
52        serde_json::to_string_pretty(&self.to_json_value())
53    }
54}
55
56impl ComponentHealth {
57    pub fn to_json_value(&self) -> Value {
58        json!({
59            "name": self.name,
60            "status": health_status_to_json_value(&self.status),
61            "metrics": self.metrics.iter().map(metrics_snapshot_to_json_value).collect::<Vec<_>>()
62        })
63    }
64}
65
66fn health_status_to_json_value(status: &HealthStatus) -> Value {
67    match status {
68        HealthStatus::Healthy => json!({
69            "state": "healthy"
70        }),
71        HealthStatus::Degraded { reason } => json!({
72            "state": "degraded",
73            "reason": reason
74        }),
75        HealthStatus::Unhealthy { reason } => json!({
76            "state": "unhealthy",
77            "reason": reason
78        }),
79    }
80}
81
82fn metrics_snapshot_to_json_value(metric: &MetricsSnapshot) -> Value {
83    json!({
84        "name": metric.name,
85        "value": metric.value,
86        "unit": metric.unit
87    })
88}
89
90// ─── Merge helper ─────────────────────────────────────────────────────────────
91
92fn merge_health(a: HealthStatus, b: &HealthStatus) -> HealthStatus {
93    match (&a, b) {
94        (_, HealthStatus::Unhealthy { .. }) => b.clone(),
95        (HealthStatus::Unhealthy { .. }, _) => a,
96        (_, HealthStatus::Degraded { .. }) => b.clone(),
97        (HealthStatus::Degraded { .. }, _) => a,
98        _ => HealthStatus::Healthy,
99    }
100}
101
102// ─── MetricsAggregator ────────────────────────────────────────────────────────
103
104struct Entry {
105    name: String,
106    provider: Box<dyn MetricsProvider + Send>,
107}
108
109/// Collects metrics and health from multiple [`MetricsProvider`] implementations.
110///
111/// # Example
112/// ```
113/// use obs_core::aggregator::MetricsAggregator;
114///
115/// let mut agg = MetricsAggregator::new();
116/// assert!(agg.is_empty());
117/// ```
118#[derive(Default)]
119pub struct MetricsAggregator {
120    entries: Vec<Entry>,
121}
122
123impl MetricsAggregator {
124    pub fn new() -> Self {
125        Self::default()
126    }
127
128    /// Register a named provider.
129    pub fn register(&mut self, name: impl Into<String>, provider: Box<dyn MetricsProvider + Send>) {
130        self.entries.push(Entry {
131            name: name.into(),
132            provider,
133        });
134    }
135
136    /// Returns `true` when no providers are registered.
137    pub fn is_empty(&self) -> bool {
138        self.entries.is_empty()
139    }
140
141    /// Number of registered providers.
142    pub fn len(&self) -> usize {
143        self.entries.len()
144    }
145
146    /// Collect all metrics from every registered provider.
147    pub fn collect_all(&self) -> Vec<(String, Vec<MetricsSnapshot>)> {
148        self.entries
149            .iter()
150            .map(|e| (e.name.clone(), e.provider.collect()))
151            .collect()
152    }
153
154    /// Build a complete [`HealthReport`] from all providers.
155    pub fn health_report(&self) -> HealthReport {
156        let mut overall = HealthStatus::Healthy;
157        let components = self
158            .entries
159            .iter()
160            .map(|e| {
161                let status = e.provider.health();
162                let metrics = e.provider.collect();
163                overall = merge_health(overall.clone(), &status);
164                ComponentHealth {
165                    name: e.name.clone(),
166                    status,
167                    metrics,
168                }
169            })
170            .collect();
171
172        HealthReport {
173            components,
174            overall,
175        }
176    }
177
178    /// Flat iterator over every `MetricsSnapshot` from every provider,
179    /// prefixed with the provider name: `"<name>/<metric_name>"`.
180    pub fn flat_metrics(&self) -> Vec<MetricsSnapshot> {
181        self.entries
182            .iter()
183            .flat_map(|e| {
184                e.provider.collect().into_iter().map(|mut m| {
185                    m.name = format!("{}/{}", e.name, m.name);
186                    m
187                })
188            })
189            .collect()
190    }
191}
obs_core/aggregator.rs

obs_core/
aggregator.rs
