feat: add free disk circuit breaker

Author: lzf575

agent/Cargo.lock

@@ -48,6 +48,7 @@ enum_dispatch = "0.3.7"
 envmnt = "0.10.4"
 flate2 = "1.0"
 flexi_logger = "0.29"
+fs2 = "0.4"
 futures = "~0.3"
 grpc = { path = "plugins/grpc" }
 hex = "0.4.3"

agent/Cargo.toml

@@ -2079,12 +2079,35 @@ impl Default for TxThroughput {
     }
 }
 
-#[derive(Clone, Copy, Default, Debug, Deserialize, PartialEq, Eq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct FreeDisk {
+    pub percentage_trigger_threshold: u8,
+    #[serde(deserialize_with = "deser_u64_with_giga_unit")]
+    pub absolute_trigger_threshold: u64,
+    pub directories: Vec<String>,
+}
+
+impl Default for FreeDisk {
+    fn default() -> Self {
+        Self {
+            percentage_trigger_threshold: 15,
+            absolute_trigger_threshold: 10 << 30,
+            #[cfg(not(target_os = "windows"))]
+            directories: vec!["/".to_string()],
+            #[cfg(target_os = "windows")]
+            directories: vec!["c:\\".to_string()],
+        }
+    }
+}
+
+#[derive(Clone, Default, Debug, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct CircuitBreakers {
     pub sys_memory_percentage: SysMemoryPercentage,
     pub relative_sys_load: RelativeSysLoad,
     pub tx_throughput: TxThroughput,
+    pub free_disk: FreeDisk,
 }
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
@@ -3329,6 +3352,13 @@ where
     Ok(usize::deserialize(deserializer)? << 20)
 }
 
+fn deser_u64_with_giga_unit<'de, D>(deserializer: D) -> Result<u64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    Ok(u64::deserialize(deserializer)? << 30)
+}
+
 // `humantime` will not parse "0" as Duration::ZERO
 // If "0" is a valid option for a duration field, use this deserializer
 //     #[serde(deserialize_with = "deser_humantime_with_zero")]

agent/src/config/config.rs

@@ -218,6 +218,9 @@ pub struct EnvironmentConfig {
     pub system_load_circuit_breaker_recover: f32,
     pub system_load_circuit_breaker_metric: agent::SystemLoadMetric,
     pub page_cache_reclaim_percentage: u8,
+    pub free_disk_circuit_breaker_percentage_threshold: u8,
+    pub free_disk_circuit_breaker_absolute_threshold: u64,
+    pub free_disk_circuit_breaker_directories: Vec<String>,
 }
 
 #[derive(Clone, PartialEq, Eq, Debug)]
@@ -1665,6 +1668,22 @@ impl TryFrom<(Config, UserConfig)> for ModuleConfig {
                     .relative_sys_load
                     .metric,
                 page_cache_reclaim_percentage: conf.global.tunning.page_cache_reclaim_percentage,
+                free_disk_circuit_breaker_percentage_threshold: conf
+                    .global
+                    .circuit_breakers
+                    .free_disk
+                    .percentage_trigger_threshold,
+                free_disk_circuit_breaker_absolute_threshold: conf
+                    .global
+                    .circuit_breakers
+                    .free_disk
+                    .absolute_trigger_threshold,
+                free_disk_circuit_breaker_directories: {
+                    let mut v = conf.global.circuit_breakers.free_disk.directories.clone();
+                    v.sort();
+                    v.dedup();
+                    v
+                },
             },
             synchronizer: SynchronizerConfig {
                 sync_interval: conf.global.communication.proactive_request_interval,
@@ -4011,6 +4030,34 @@ impl ConfigHandler {
             limits.max_sockets_tolerate_interval = new_limits.max_sockets_tolerate_interval;
         }
 
+        let free_disk = &mut config.global.circuit_breakers.free_disk;
+        let new_free_disk = &mut new_config.user_config.global.circuit_breakers.free_disk;
+        if free_disk.percentage_trigger_threshold != new_free_disk.percentage_trigger_threshold {
+            info!(
+                "Update global.circuit_breakers.free_disk.percentage_trigger_threshold from {:?} to {:?}.",
+                free_disk.percentage_trigger_threshold, new_free_disk.percentage_trigger_threshold
+            );
+            free_disk.percentage_trigger_threshold = new_free_disk.percentage_trigger_threshold;
+        }
+
+        if free_disk.absolute_trigger_threshold != new_free_disk.absolute_trigger_threshold {
+            info!(
+                "Update global.circuit_breakers.free_disk.absolute_trigger_threshold from {:?} to {:?}.",
+                free_disk.absolute_trigger_threshold, new_free_disk.absolute_trigger_threshold
+            );
+            free_disk.absolute_trigger_threshold = new_free_disk.absolute_trigger_threshold;
+        }
+
+        new_free_disk.directories.sort();
+        new_free_disk.directories.dedup();
+        if free_disk.directories != new_free_disk.directories {
+            info!(
+                "Update global.circuit_breakers.free_disk.directories from {:?} to {:?}.",
+                free_disk.directories, new_free_disk.directories
+            );
+            free_disk.directories = new_free_disk.directories.clone();
+        }
+
         let ntp = &mut config.global.ntp;
         let new_ntp = &mut new_config.user_config.global.ntp;
         if ntp.enabled != new_ntp.enabled {

agent/src/config/handler.rs

@@ -30,7 +30,10 @@ use sysinfo::NetworkExt;
 use sysinfo::{get_current_pid, Pid, ProcessExt, ProcessRefreshKind, System, SystemExt};
 
 #[cfg(target_os = "linux")]
-use crate::utils::{cgroups, environment::SocketInfo};
+use crate::utils::{
+    cgroups,
+    environment::{get_disk_usage, SocketInfo},
+};
 use crate::{
     config::handler::EnvironmentAccess,
     error::{Error, Result},
@@ -227,7 +230,8 @@ impl RefCountable for SysStatusBroker {
             CounterValue::Unsigned(current_sys_available_memory_percentage as u64),
         ));
 
-        let sys_memory_limit = self.config.load().sys_memory_limit as f64;
+        let config = self.config.load();
+        let sys_memory_limit = config.sys_memory_limit as f64;
 
         let (sys_free_memory_limit_ratio, sys_available_memory_limit_ratio) =
             if sys_memory_limit > 0.0 {
@@ -266,6 +270,7 @@ impl RefCountable for SysStatusBroker {
                 warn!("get file and size sum failed: {:?}", e);
             }
         }
+
         match system_guard.process(self.pid) {
             Some(process) => {
                 let cpu_usage = process.cpu_usage() as f64;
@@ -279,7 +284,7 @@ impl RefCountable for SysStatusBroker {
                 metrics.push((
                     "max_millicpus_ratio",
                     CounterType::Gauged,
-                    CounterValue::Float(cpu_usage * 10.0 / self.config.load().max_millicpus as f64),
+                    CounterValue::Float(cpu_usage * 10.0 / config.max_millicpus as f64),
                 ));
                 metrics.push((
                     "memory",
@@ -289,7 +294,7 @@ impl RefCountable for SysStatusBroker {
                 metrics.push((
                     "max_memory_ratio",
                     CounterType::Gauged,
-                    CounterValue::Float(mem_used as f64 / self.config.load().max_memory as f64),
+                    CounterValue::Float(mem_used as f64 / config.max_memory as f64),
                 ));
                 metrics.push((
                     "create_time",
@@ -373,13 +378,54 @@ impl stats::Module for NetStats<'_> {
     }
 }
 
+struct FreeDiskUsage {
+    directory: String,
+}
+
+impl stats::Module for FreeDiskUsage {
+    fn name(&self) -> &'static str {
+        "free_disk"
+    }
+
+    fn tags(&self) -> Vec<StatsOption> {
+        vec![StatsOption::Tag("directory", self.directory.clone())]
+    }
+}
+
+impl RefCountable for FreeDiskUsage {
+    fn get_counters(&self) -> Vec<Counter> {
+        let mut metrics = vec![];
+        match get_disk_usage(&self.directory) {
+            Ok((total, free)) => {
+                metrics.push((
+                    "free_disk_percentage",
+                    CounterType::Gauged,
+                    CounterValue::Float(free as f64 * 100.0 / total as f64),
+                ));
+                metrics.push((
+                    "free_disk_absolute",
+                    CounterType::Gauged,
+                    CounterValue::Unsigned(free as u64),
+                ));
+            }
+            Err(e) => {
+                warn!("get disk free usage failed: {:?}", e);
+            }
+        }
+        metrics
+    }
+}
+
 pub struct Monitor {
     stats: Arc<Collector>,
     running: AtomicBool,
     sys_monitor: Arc<SysStatusBroker>,
     sys_load: Arc<SysLoad>,
     link_map: Arc<Mutex<HashMap<String, Arc<LinkStatusBroker>>>>,
     system: Arc<Mutex<System>>,
+    config: EnvironmentAccess,
+    free_disks_config: Arc<Mutex<Vec<String>>>,
+    free_disk_counters: Arc<Mutex<Vec<Arc<FreeDiskUsage>>>>,
 }
 
 impl Monitor {
@@ -399,6 +445,9 @@ impl Monitor {
             sys_load: Arc::new(SysLoad(system.clone())),
             link_map: Arc::new(Mutex::new(HashMap::new())),
             system,
+            config: config.clone(),
+            free_disks_config: Arc::new(Mutex::new(vec![])),
+            free_disk_counters: Arc::new(Mutex::new(vec![])),
         })
     }
 
@@ -519,6 +568,42 @@ impl Monitor {
             Countable::Ref(Arc::downgrade(&self.sys_load) as Weak<dyn RefCountable>),
         );
 
+        let config = self.config.clone();
+        let stats_collector = self.stats.clone();
+        let free_disks_config = self.free_disks_config.clone();
+        let free_disk_counters = self.free_disk_counters.clone();
+        self.stats.register_pre_hook(Box::new(move || {
+            let config_load = config.load();
+            let mut free_disks_config = free_disks_config.lock().unwrap();
+            if config_load.free_disk_circuit_breaker_directories == *free_disks_config {
+                return;
+            }
+
+            let mut locked_counters = free_disk_counters.lock().unwrap();
+            let old_data = std::mem::take(&mut *locked_counters);
+            stats_collector
+                .deregister_countables(old_data.iter().map(|c| c.as_ref() as &dyn stats::Module));
+
+            for free_disk in &config_load.free_disk_circuit_breaker_directories {
+                let free_disk_counter = Arc::new(FreeDiskUsage {
+                    directory: free_disk.clone(),
+                });
+                stats_collector.register_countable(
+                    &FreeDiskUsage {
+                        directory: free_disk.clone(),
+                    },
+                    Countable::Ref(Arc::downgrade(&free_disk_counter) as Weak<dyn RefCountable>),
+                );
+                locked_counters.push(free_disk_counter);
+            }
+
+            info!(
+                "update free disk monitor from {:?} to {:?}",
+                free_disks_config, config_load.free_disk_circuit_breaker_directories
+            );
+            *free_disks_config = config_load.free_disk_circuit_breaker_directories.clone();
+        }));
+
         info!("monitor started");
     }
 

agent/src/monitor.rs

@@ -17,7 +17,7 @@
 use std::{
     cell::OnceCell,
     env::{self, VarError},
-    fs,
+    fs, io,
     iter::Iterator,
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
     path::Path,
@@ -26,6 +26,7 @@ use std::{
 };
 
 use bytesize::ByteSize;
+use fs2::{free_space, total_space};
 use log::{error, warn};
 use sysinfo::{DiskExt, System, SystemExt};
 
@@ -425,3 +426,18 @@ pub fn get_ctrl_ip_and_mac(dest: &IpAddr) -> Result<(IpAddr, MacAddr)> {
         "failed getting control ip and mac, deepflow-agent restart...".to_owned(),
     ))
 }
+
+pub fn get_disk_usage(directory: &str) -> io::Result<(u64, u64)> {
+    let path = std::path::Path::new(directory);
+    if !path.exists() {
+        return Err(io::Error::new(
+            io::ErrorKind::NotFound,
+            format!("path does not exist: {}", path.display()),
+        ));
+    }
+
+    let total = total_space(path)?;
+    let free = free_space(path)?;
+
+    Ok((total, free))
+}