etdl_core/
monitor.rs

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

etdl_core/
monitor.rs
