cbtop/federated_metrics/
host.rs

1//! Host state, aggregated metrics, and configuration for federated metrics.
2
3use std::collections::HashMap;
4use std::time::{Duration, Instant};
5
6use super::crdt::GCounter;
7
8/// Host state in the federation
9#[derive(Debug, Clone)]
10pub struct FederatedHost {
11    /// Host identifier
12    pub host_id: String,
13    /// Last seen timestamp
14    pub last_seen: Instant,
15    /// Sample count from this host
16    pub sample_count: GCounter,
17    /// Health status (0.0 = dead, 1.0 = healthy)
18    pub health: f64,
19    /// Current sampling rate (samples per second)
20    pub sampling_rate: f64,
21    /// Network latency to this host (milliseconds)
22    pub latency_ms: f64,
23    /// Logical clock value
24    pub logical_clock: u64,
25}
26
27impl FederatedHost {
28    /// Create a new federated host
29    pub fn new(host_id: impl Into<String>) -> Self {
30        Self {
31            host_id: host_id.into(),
32            last_seen: Instant::now(),
33            sample_count: GCounter::new(),
34            health: 1.0,
35            sampling_rate: 100.0, // Default 100 Hz
36            latency_ms: 0.0,
37            logical_clock: 0,
38        }
39    }
40
41    /// Update last seen time
42    pub fn touch(&mut self) {
43        self.last_seen = Instant::now();
44    }
45
46    /// Check if host is stale (not seen recently)
47    pub fn is_stale(&self, timeout: Duration) -> bool {
48        self.last_seen.elapsed() > timeout
49    }
50
51    /// Increment logical clock
52    pub fn tick(&mut self) -> u64 {
53        self.logical_clock += 1;
54        self.logical_clock
55    }
56
57    /// Update logical clock from received message
58    pub fn sync_clock(&mut self, received_time: u64) {
59        self.logical_clock = self.logical_clock.max(received_time) + 1;
60    }
61}
62
63/// Aggregated metrics across federation
64#[derive(Debug, Clone, Default)]
65pub struct AggregatedMetrics {
66    /// Metric name
67    pub metric_name: String,
68    /// All sample values
69    pub values: Vec<f64>,
70    /// Per-host sample counts
71    pub host_counts: HashMap<String, usize>,
72    /// Minimum value
73    pub min: f64,
74    /// Maximum value
75    pub max: f64,
76    /// Sum for mean calculation
77    pub sum: f64,
78}
79
80impl AggregatedMetrics {
81    /// Create new aggregated metrics
82    pub fn new(metric_name: impl Into<String>) -> Self {
83        Self {
84            metric_name: metric_name.into(),
85            values: Vec::new(),
86            host_counts: HashMap::new(),
87            min: f64::INFINITY,
88            max: f64::NEG_INFINITY,
89            sum: 0.0,
90        }
91    }
92
93    /// Add a sample
94    pub fn add_sample(&mut self, host_id: &str, value: f64) {
95        self.values.push(value);
96        *self.host_counts.entry(host_id.to_string()).or_insert(0) += 1;
97        self.min = self.min.min(value);
98        self.max = self.max.max(value);
99        self.sum += value;
100    }
101
102    /// Get mean value
103    pub fn mean(&self) -> f64 {
104        if self.values.is_empty() {
105            0.0
106        } else {
107            self.sum / self.values.len() as f64
108        }
109    }
110
111    /// Get percentile value
112    pub fn percentile(&self, p: f64) -> f64 {
113        if self.values.is_empty() {
114            return 0.0;
115        }
116
117        let mut sorted = self.values.clone();
118        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
119
120        let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize;
121        sorted[idx.min(sorted.len() - 1)]
122    }
123
124    /// Get p50
125    pub fn p50(&self) -> f64 {
126        self.percentile(50.0)
127    }
128
129    /// Get p95
130    pub fn p95(&self) -> f64 {
131        self.percentile(95.0)
132    }
133
134    /// Get p99
135    pub fn p99(&self) -> f64 {
136        self.percentile(99.0)
137    }
138
139    /// Detect skewed host (significantly slower)
140    pub fn detect_skew(&self, threshold_percent: f64) -> Vec<String> {
141        let mean = self.mean();
142        if mean == 0.0 {
143            return Vec::new();
144        }
145
146        let mut skewed = Vec::new();
147        for (host_id, count) in &self.host_counts {
148            // Calculate host-specific mean (simplified: use count as proxy)
149            let expected_count = self.values.len() / self.host_counts.len().max(1);
150            let deviation =
151                ((*count as f64 - expected_count as f64) / expected_count as f64).abs() * 100.0;
152
153            if deviation > threshold_percent {
154                skewed.push(host_id.clone());
155            }
156        }
157        skewed
158    }
159}
160
161/// Configuration for federation
162#[derive(Debug, Clone)]
163pub struct FederationConfig {
164    /// Maximum clock drift tolerance (milliseconds)
165    pub max_clock_drift_ms: i64,
166    /// Host timeout before considered stale
167    pub host_timeout: Duration,
168    /// Memory limit per federation (bytes)
169    pub memory_limit_bytes: usize,
170    /// Default sampling rate (Hz)
171    pub default_sampling_rate: f64,
172    /// Skew detection threshold (percent)
173    pub skew_threshold_percent: f64,
174    /// Partition recovery timeout
175    pub partition_recovery_timeout: Duration,
176}
177
178impl Default for FederationConfig {
179    fn default() -> Self {
180        Self {
181            max_clock_drift_ms: 100,
182            host_timeout: Duration::from_secs(30),
183            memory_limit_bytes: 100 * 1024 * 1024, // 100MB
184            default_sampling_rate: 100.0,
185            skew_threshold_percent: 40.0,
186            partition_recovery_timeout: Duration::from_secs(30),
187        }
188    }
189}
cbtop/federated_metrics/host.rs

cbtop/federated_metrics/
host.rs
