etdl_core/
monitor.rs

1use crate::chaos::ChaosController;
2use crate::observation::{NoopSink, ReliabilityObservation, SharedSink};
3use crate::sla::SlaTracker;
4use std::sync::Arc;
5use std::sync::Mutex;
6
7pub struct BranchMonitor {
8    node_id: String,
9    sla_tracker: Arc<Mutex<SlaTracker>>,
10    chaos: Arc<Mutex<ChaosController>>,
11    observation_sink: SharedSink,
12}
13
14impl BranchMonitor {
15    pub fn new(node_id: &str) -> Self {
16        BranchMonitor {
17            node_id: node_id.to_string(),
18            sla_tracker: Arc::new(Mutex::new(SlaTracker::new())),
19            chaos: Arc::new(Mutex::new(ChaosController::new())),
20            observation_sink: Arc::new(NoopSink),
21        }
22    }
23
24    /// Attach an observation sink (JSON Lines, OTel, database adapter, ...).
25    pub fn with_sink(mut self, sink: SharedSink) -> Self {
26        self.observation_sink = sink;
27        self
28    }
29
30    /// Record an immutable failure observation for offline analysis.
31    /// Lightweight: does not run statistics, query databases, or call AI.
32    pub fn record_failure_observation(&self, observation: ReliabilityObservation) {
33        self.observation_sink.emit(&observation);
34    }
35
36    pub fn with_sla(node_id: &str, sla_tracker: Arc<Mutex<SlaTracker>>) -> Self {
37        BranchMonitor {
38            node_id: node_id.to_string(),
39            sla_tracker,
40            chaos: Arc::new(Mutex::new(ChaosController::new())),
41            observation_sink: Arc::new(NoopSink),
42        }
43    }
44
45    pub fn with_chaos(node_id: &str, chaos: Arc<Mutex<ChaosController>>) -> Self {
46        BranchMonitor {
47            node_id: node_id.to_string(),
48            sla_tracker: Arc::new(Mutex::new(SlaTracker::new())),
49            chaos,
50            observation_sink: Arc::new(NoopSink),
51        }
52    }
53
54    pub fn record_branch(&mut self, outcome: &str, declared_probability: f64) {
55        let should_chaos = {
56            let mut chaos = self.chaos.lock().unwrap();
57            chaos.should_inject_chaos(&self.node_id)
58        };
59        if should_chaos {
60            return;
61        }
62
63        let mut sla = self.sla_tracker.lock().unwrap();
64        let anomaly = sla.record(&self.node_id, outcome, declared_probability, true);
65
66        if anomaly {
67            crate::telemetry::emit_anomaly_event(
68                &self.node_id,
69                outcome,
70                declared_probability,
71                sla.observed_frequency(&self.node_id, outcome),
72            );
73        }
74
75        crate::telemetry::attach_node_span_attribute(&self.node_id);
76    }
77
78    pub fn record_failure(
79        &mut self,
80        operation_id: &str,
81        error: &dyn std::error::Error,
82        declared_probability: Option<f64>,
83    ) {
84        let key = format!("{}.failure", operation_id);
85        let outcome = "FAILURE";
86
87        if let Some(prob) = declared_probability {
88            let mut sla = self.sla_tracker.lock().unwrap();
89            let anomaly = sla.record(&key, outcome, prob, true);
90
91            if anomaly {
92                crate::telemetry::emit_anomaly_event(
93                    &key,
94                    outcome,
95                    prob,
96                    sla.observed_frequency(&key, outcome),
97                );
98            }
99        }
100
101        crate::telemetry::attach_node_span_attribute(&key);
102        eprintln!("[etdl] operation '{}' failed: {}", operation_id, error);
103    }
104
105    pub fn flush(&self) {
106        let sla = self.sla_tracker.lock().unwrap();
107        eprintln!(
108            "[etdl] node '{}': {} evaluations recorded",
109            self.node_id,
110            sla.total_evaluations()
111        );
112    }
113}
114
115impl Drop for BranchMonitor {
116    fn drop(&mut self) {
117        self.flush();
118    }
119}
120
121#[cfg(test)]
122mod tests {
123    use super::*;
124
125    #[test]
126    fn test_record_branch() {
127        let mut monitor = BranchMonitor::new("test_barrier");
128        monitor.record_branch("SUCCESS", 0.95);
129        monitor.record_branch("SUCCESS", 0.95);
130        monitor.record_branch("FAILURE", 0.05);
131    }
132}
etdl_core/monitor.rs

etdl_core/
monitor.rs
