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(&self.node_id, outcome, declared_probability, true);
48
49        if anomaly {
50            crate::telemetry::emit_anomaly_event(
51                &self.node_id,
52                outcome,
53                declared_probability,
54                sla.observed_frequency(&self.node_id, outcome),
55            );
56        }
57
58        crate::telemetry::attach_node_span_attribute(&self.node_id);
59    }
60
61    pub fn record_failure(
62        &mut self,
63        operation_id: &str,
64        error: &dyn std::error::Error,
65        declared_probability: Option<f64>,
66    ) {
67        let key = format!("{}.failure", operation_id);
68        let outcome = "FAILURE";
69
70        if let Some(prob) = declared_probability {
71            let mut sla = self.sla_tracker.lock().unwrap();
72            let anomaly = sla.record(&key, outcome, prob, true);
73
74            if anomaly {
75                crate::telemetry::emit_anomaly_event(
76                    &key,
77                    outcome,
78                    prob,
79                    sla.observed_frequency(&key, outcome),
80                );
81            }
82        }
83
84        crate::telemetry::attach_node_span_attribute(&key);
85        eprintln!("[etdl] operation '{}' failed: {}", operation_id, error);
86    }
87
88    pub fn flush(&self) {
89        let sla = self.sla_tracker.lock().unwrap();
90        eprintln!(
91            "[etdl] node '{}': {} evaluations recorded",
92            self.node_id,
93            sla.total_evaluations()
94        );
95    }
96}
97
98impl Drop for BranchMonitor {
99    fn drop(&mut self) {
100        self.flush();
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107
108    #[test]
109    fn test_record_branch() {
110        let mut monitor = BranchMonitor::new("test_barrier");
111        monitor.record_branch("SUCCESS", 0.95);
112        monitor.record_branch("SUCCESS", 0.95);
113        monitor.record_branch("FAILURE", 0.05);
114    }
115}
etdl_core/monitor.rs

etdl_core/
monitor.rs
