traffic_core: продакшн-фиксы runtime и валидации

config.example.toml

@@ -6,8 +6,8 @@ peer_ipv4_cidr = "10.8.1.0/24"
 
 [traffic]
 enabled = true
-peer_limit = "50mbit"
+peer_limit = "70mbit"
-root_rate = "500mbit"
+root_rate = "1000mbit"
 reconcile_interval_secs = 15
 container_ready_timeout_secs = 120
 class_start_id = 10

traffic_core/src/cli.rs

@@ -5,7 +5,7 @@ use clap::{Parser, Subcommand};
 #[derive(Debug, Parser)]
 #[command(name = "awg2-harness")]
 pub struct Cli {
-    #[arg(long)]
+    #[arg(long, global = true)]
     pub config: Option<PathBuf>,
 
     #[command(subcommand)]
@@ -20,4 +20,5 @@ pub enum Command {
     Status,
     EnsureSwap,
     ValidateConfig,
+    ValidateRuntime,
 }

traffic_core/src/config.rs

@@ -61,13 +61,30 @@ impl Config {
         Ok(cfg)
     }
 
-    pub fn validate(&self) -> anyhow::Result<()> {
+    pub fn validate_config(&self) -> anyhow::Result<()> {
         if self.amnezia.container_names.is_empty() {
             anyhow::bail!("amnezia.container_names must not be empty");
         }
+        if self.amnezia.interface_candidates.is_empty() {
+            anyhow::bail!("amnezia.interface_candidates must not be empty");
+        }
+        if self.amnezia.interface_patterns.is_empty() {
+            anyhow::bail!("amnezia.interface_patterns must not be empty");
+        }
         if self.traffic.class_start_id < 10 {
             anyhow::bail!("traffic.class_start_id must be >= 10");
         }
+        if self.traffic.ifb_name.trim().is_empty() {
+            anyhow::bail!("traffic.ifb_name must not be empty");
+        }
+        if self.traffic.peer_limit.trim().is_empty() {
+            anyhow::bail!("traffic.peer_limit must not be empty");
+        }
+        if self.traffic.root_rate.trim().is_empty() {
+            anyhow::bail!("traffic.root_rate must not be empty");
+        }
+        parse_tc_rate_bps(&self.traffic.peer_limit)?;
+        parse_tc_rate_bps(&self.traffic.root_rate)?;
         if self.swap.swappiness > 100 {
             anyhow::bail!("swap.swappiness must be <= 100");
         }
@@ -77,12 +94,36 @@ impl Config {
         if self.traffic.reconcile_interval_secs == 0 {
             anyhow::bail!("traffic.reconcile_interval_secs must be > 0");
         }
+        if self.swap.enabled {
+            if self.swap.size_mb == 0 {
+                anyhow::bail!("swap.size_mb must be > 0 when swap.enabled=true");
+            }
+            if !self.swap.path.starts_with('/') {
+                anyhow::bail!("swap.path must be absolute");
+            }
+        }
+        if self.anti_torrent.enabled && self.anti_torrent.max_connections_per_peer == 0 {
+            anyhow::bail!("anti_torrent.max_connections_per_peer must be > 0");
+        }
         if self.reboot.enabled && self.reboot.window_start_utc == self.reboot.window_end_utc {
             anyhow::bail!("reboot window start and end must differ");
         }
+        if self.reboot.enabled {
+            parse_hhmm(&self.reboot.window_start_utc)?;
+            parse_hhmm(&self.reboot.window_end_utc)?;
+        }
         parse_cidr_v4(&self.amnezia.peer_ipv4_cidr)?;
         parse_cidr_v4(&self.anti_torrent.peer_ipv4_cidr)?;
-        self.check_host_tools()?;
+        Ok(())
+    }
+
+    pub fn validate_runtime(&self) -> anyhow::Result<()> {
+        self.check_host_tools()
+    }
+
+    pub fn validate_all(&self) -> anyhow::Result<()> {
+        self.validate_config()?;
+        self.validate_runtime()?;
         Ok(())
     }
 
@@ -106,6 +147,49 @@ impl Config {
     }
 }
 
+fn parse_hhmm(v: &str) -> anyhow::Result<()> {
+    let mut s = v.split(':');
+    let h = s
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("invalid time: {v}"))?
+        .parse::<u8>()?;
+    let m = s
+        .next()
+        .ok_or_else(|| anyhow::anyhow!("invalid time: {v}"))?
+        .parse::<u8>()?;
+    if s.next().is_some() {
+        anyhow::bail!("invalid time: {v}");
+    }
+    if h > 23 || m > 59 {
+        anyhow::bail!("invalid time: {v}");
+    }
+    Ok(())
+}
+
+fn parse_tc_rate_bps(v: &str) -> anyhow::Result<u64> {
+    let raw = v.trim().to_ascii_lowercase().replace(' ', "");
+    if raw.is_empty() {
+        anyhow::bail!("invalid tc rate: empty");
+    }
+    let (num, mul) = if let Some(n) = raw.strip_suffix("gbit") {
+        (n, 1_000_000_000u64)
+    } else if let Some(n) = raw.strip_suffix("mbit") {
+        (n, 1_000_000u64)
+    } else if let Some(n) = raw.strip_suffix("kbit") {
+        (n, 1_000u64)
+    } else if let Some(n) = raw.strip_suffix("bit") {
+        (n, 1u64)
+    } else {
+        anyhow::bail!("invalid tc rate unit: {v}");
+    };
+
+    let base = num
+        .parse::<u64>()
+        .map_err(|_| anyhow::anyhow!("invalid tc rate number: {v}"))?;
+    base.checked_mul(mul)
+        .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}"))
+}
+
 fn tool_exists(tool: &str) -> bool {
     let probes: &[&[&str]] = match tool {
         "tc" => &[&["-V"]],
@@ -171,4 +255,18 @@ mod tests {
         let no = ip_in_cidr("10.9.1.10".parse().expect("ip"), "10.8.1.0/24").expect("cidr");
         assert!(!no);
     }
+
+    #[test]
+    fn parse_hhmm_works() {
+        assert!(parse_hhmm("01:00").is_ok());
+        assert!(parse_hhmm("23:59").is_ok());
+        assert!(parse_hhmm("24:00").is_err());
+    }
+
+    #[test]
+    fn parse_tc_rate_works() {
+        assert_eq!(parse_tc_rate_bps("1gbit").expect("bps"), 1_000_000_000);
+        assert_eq!(parse_tc_rate_bps("1000mbit").expect("bps"), 1_000_000_000);
+        assert!(parse_tc_rate_bps("foo").is_err());
+    }
 }

traffic_core/src/daemon.rs

@@ -6,17 +6,21 @@ use std::{
     time::{Duration, Instant},
 };
 
-use crate::{config::Config, docker, firewall, iface, netns::build_transport, reboot, traffic};
+use crate::{config::Config, docker, firewall, iface, netns::build_transport, traffic};
 use tracing::{error, info};
 
 #[derive(Debug, Clone, Default)]
 struct LastActionState {
+    last_ok_at: Option<String>,
     last_ok: Option<String>,
+    last_error_at: Option<String>,
     last_error: Option<String>,
 }
 
 const LAST_STATE_FILE: &str = "/run/awg2-harness/last_action.state";
 const EVENTS_FILE: &str = "/run/awg2-harness/events.log";
+const EVENTS_MAX_LINES: usize = 1000;
+const EVENTS_MAX_BYTES: usize = 256 * 1024;
 
 pub fn run(cfg: Config) -> anyhow::Result<()> {
     info!("starting daemon");
@@ -36,22 +40,22 @@ pub fn run(cfg: Config) -> anyhow::Result<()> {
         set_ok(&state, "startup firewall reconcile ok");
         append_event("startup_firewall ok");
     }
-    let _ = reboot::maybe_schedule_reboot(&cfg);
-    let reboot_cfg = cfg.clone();
-    thread::spawn(move || {
-        if let Err(e) = reboot::reboot_scheduler_loop(reboot_cfg) {
-            error!(error = %e, "reboot scheduler loop error");
-        }
-    });
-
     let events_cfg = cfg.clone();
     let events_state = state.clone();
     thread::spawn(move || {
         loop {
-            if let Err(e) = docker_events_loop(&events_cfg, &events_state) {
+            match docker_events_loop(&events_cfg, &events_state) {
-                set_err(&events_state, format!("docker events loop crashed: {e}"));
+                Ok(()) => {
-                thread::sleep(Duration::from_secs(10));
+                    set_err(
+                        &events_state,
+                        "docker events stream ended, restarting".to_string(),
+                    );
                 }
+                Err(e) => {
+                    set_err(&events_state, format!("docker events loop crashed: {e}"));
+                }
+            }
+            thread::sleep(Duration::from_secs(10));
         }
     });
 
@@ -144,13 +148,16 @@ fn wait_for_interface(cfg: &Config, container_name: &str, timeout_secs: u64) ->
 
 fn set_ok(state: &Arc<Mutex<LastActionState>>, msg: &str) {
     if let Ok(mut s) = state.lock() {
+        s.last_ok_at = Some(now_rfc3339());
         s.last_ok = Some(msg.to_string());
+        s.last_error = None;
         persist_state(&s);
     }
 }
 
 fn set_err(state: &Arc<Mutex<LastActionState>>, msg: String) {
     if let Ok(mut s) = state.lock() {
+        s.last_error_at = Some(now_rfc3339());
         s.last_error = Some(msg);
         persist_state(&s);
     }
@@ -159,8 +166,10 @@ fn set_err(state: &Arc<Mutex<LastActionState>>, msg: String) {
 fn persist_state(state: &LastActionState) {
     let _ = fs::create_dir_all("/run/awg2-harness");
     let content = format!(
-        "last_ok={}\nlast_error={}\n",
+        "last_ok_at={}\nlast_ok={}\nlast_error_at={}\nlast_error={}\n",
+        state.last_ok_at.clone().unwrap_or_default(),
         state.last_ok.clone().unwrap_or_default(),
+        state.last_error_at.clone().unwrap_or_default(),
         state.last_error.clone().unwrap_or_default()
     );
     let _ = fs::write(LAST_STATE_FILE, content);
@@ -168,12 +177,34 @@ fn persist_state(state: &LastActionState) {
 
 fn append_event(event: &str) {
     let _ = fs::create_dir_all("/run/awg2-harness");
-    let ts = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+    let ts = now_rfc3339();
-    if let Ok(mut f) = fs::OpenOptions::new()
+    let mut lines: Vec<String> = fs::read_to_string(EVENTS_FILE)
-        .create(true)
+        .ok()
-        .append(true)
+        .map(|s| s.lines().map(|v| v.to_string()).collect())
-        .open(EVENTS_FILE)
+        .unwrap_or_default();
+    lines.push(format!("{ts} {event}"));
+    while lines.len() > EVENTS_MAX_LINES {
+        lines.remove(0);
+    }
+
+    let mut content = lines.join("\n");
+    if !content.is_empty() {
+        content.push('\n');
+    }
+    while content.len() > EVENTS_MAX_BYTES {
+        if let Some(pos) = content.find('\n') {
+            content.drain(..=pos);
+        } else {
+            break;
+        }
+    }
+
+    if let Ok(mut f) = fs::OpenOptions::new().create(true).write(true).truncate(true).open(EVENTS_FILE)
     {
-        let _ = writeln!(f, "{ts} {event}");
+        let _ = f.write_all(content.as_bytes());
     }
 }
+
+fn now_rfc3339() -> String {
+    chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+}

traffic_core/src/firewall.rs

@@ -20,16 +20,13 @@ struct FirewallActual {
 }
 
 pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
-    let transport = firewall_transport(cfg)?;
-
     if !cfg.anti_torrent.enabled {
-        run_allow_notfound(
+        apply_firewall_disabled(cfg)?;
-            transport.as_ref(),
-            &["nft", "delete", "table", "inet", "awg2_harness"],
-        )?;
         return Ok(());
     }
 
+    let transport = firewall_transport(cfg)?;
+
     let actual = inspect_firewall(transport.as_ref(), cfg)?;
     if !firewall_has_drift(&actual) {
         return Ok(());
@@ -72,6 +69,30 @@ pub fn apply_firewall(cfg: &Config) -> anyhow::Result<()> {
     Ok(())
 }
 
+fn apply_firewall_disabled(cfg: &Config) -> anyhow::Result<()> {
+    if cfg.anti_torrent.apply_in_amnezia_netns {
+        match docker::discover_target(cfg) {
+            Ok(target) => {
+                let transport = build_transport(target.container_pid, &target.container_name);
+                run_allow_notfound(
+                    transport.as_ref(),
+                    &["nft", "delete", "table", "inet", "awg2_harness"],
+                )?;
+            }
+            Err(_) => {
+                // container is absent now: nothing to remove in its netns
+            }
+        }
+    } else {
+        let transport = build_host_transport();
+        run_allow_notfound(
+            transport.as_ref(),
+            &["nft", "delete", "table", "inet", "awg2_harness"],
+        )?;
+    }
+    Ok(())
+}
+
 fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result<FirewallActual> {
     let table = match transport.run(&["nft", "list", "table", "inet", "awg2_harness"]) {
         Ok(out) => out,
@@ -199,11 +220,14 @@ fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Res
     match transport.run(args) {
         Ok(_) => Ok(()),
         Err(e) => {
-            let msg = e.to_string();
+            let msg = e.to_string().to_ascii_lowercase();
-            if msg.contains("No such file")
+            if msg.contains("no such file")
-                || msg.contains("No such")
+                || msg.contains("no such")
                 || msg.contains("not found")
                 || msg.contains("not exist")
+                || msg.contains("no such table")
+                || msg.contains("no such device")
+                || msg.contains("cannot find")
             {
                 return Ok(());
             }

traffic_core/src/main.rs

@@ -7,7 +7,6 @@ mod firewall;
 mod iface;
 mod lock;
 mod netns;
-mod reboot;
 mod status;
 mod swap;
 mod traffic;
@@ -36,35 +35,40 @@ fn main() -> anyhow::Result<()> {
     match cli.command {
         Command::ValidateConfig => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_config()?;
             println!("config valid: {}", config_path.display());
         }
+        Command::ValidateRuntime => {
+            let cfg = Config::load(&config_path)?;
+            cfg.validate_runtime()?;
+            println!("runtime valid");
+        }
         Command::ApplyLimits => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_all()?;
             traffic::reconcile_traffic(&cfg)?;
             println!("traffic limits reconciled");
         }
         Command::ApplyFirewall => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_all()?;
             firewall::apply_firewall(&cfg)?;
             println!("firewall reconciled");
         }
         Command::Status => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_config()?;
             status::print_status(&cfg)?;
         }
         Command::EnsureSwap => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_all()?;
             swap::ensure_swap(&cfg)?;
             println!("swap ensured");
         }
         Command::Daemon => {
             let cfg = Config::load(&config_path)?;
-            cfg.validate()?;
+            cfg.validate_all()?;
             daemon::run(cfg)?;
         }
     }

traffic_core/src/status.rs

@@ -24,6 +24,11 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
         Ok(t) => t,
         Err(e) => {
             println!("  container: not available ({e})");
+            println!("  container_id: n/a");
+            println!("  pid: n/a");
+            println!("  netns: n/a");
+            println!("  interface: n/a");
+            print_degraded_sections(cfg)?;
             return Ok(());
         }
     };
@@ -215,9 +220,112 @@ pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
         }
     }
 
+    print_reboot_section();
+
     Ok(())
 }
 
+fn print_degraded_sections(cfg: &Config) -> anyhow::Result<()> {
+    println!("Peers:");
+    println!("  unavailable (container/netns not discovered)");
+
+    println!("WireGuard clients:");
+    println!("  unavailable (container/netns not discovered)");
+
+    println!("Traffic desired:");
+    println!("  peer_limit: {}", cfg.traffic.peer_limit);
+    println!("  root_rate: {}", cfg.traffic.root_rate);
+    println!("  ifb_name: {}", cfg.traffic.ifb_name);
+    println!("  class_start_id: {}", cfg.traffic.class_start_id);
+
+    println!("Traffic actual:");
+    println!("  drift: unknown (container/netns not discovered)");
+
+    println!("Anti-torrent:");
+    println!("  enabled: {}", cfg.anti_torrent.enabled);
+    println!(
+        "  max_connections_per_peer: {}",
+        cfg.anti_torrent.max_connections_per_peer
+    );
+    if cfg.anti_torrent.enabled {
+        match firewall::antitorrent_drop_counter(cfg) {
+            Ok(Some((packets, bytes))) => {
+                println!("  dropped_packets: {}", packets);
+                println!("  dropped_bytes: {}", bytes);
+            }
+            Ok(None) => println!("  dropped_packets: n/a"),
+            Err(e) => println!("  dropped_packets: inspect error ({e})"),
+        }
+    }
+
+    println!("Swap:");
+    println!("  enabled: {}", cfg.swap.enabled);
+    println!("  path: {}", cfg.swap.path);
+    println!("  size_mb: {}", cfg.swap.size_mb);
+    println!("  priority: {}", cfg.swap.priority);
+
+    println!("Conntrack:");
+    let ct_count = Command::new("sysctl")
+        .args(["-n", "net.netfilter.nf_conntrack_count"])
+        .output()
+        .ok()
+        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
+        .unwrap_or_else(|| "n/a".to_string());
+    let ct_max = Command::new("sysctl")
+        .args(["-n", "net.netfilter.nf_conntrack_max"])
+        .output()
+        .ok()
+        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
+        .unwrap_or_else(|| "n/a".to_string());
+    println!("  nf_conntrack_count: {}", ct_count);
+    println!("  nf_conntrack_max: {}", ct_max);
+
+    println!("Last actions:");
+    let last = fs::read_to_string("/run/awg2-harness/last_action.state").unwrap_or_default();
+    if last.trim().is_empty() {
+        println!("  no persisted state");
+    } else {
+        for line in last.lines() {
+            println!("  {}", line);
+        }
+    }
+
+    println!("Recent events:");
+    let events = fs::read_to_string("/run/awg2-harness/events.log").unwrap_or_default();
+    let lines: Vec<&str> = events.lines().collect();
+    if lines.is_empty() {
+        println!("  no recent events");
+    } else {
+        for l in lines.iter().rev().take(10).rev() {
+            println!("  {l}");
+        }
+    }
+
+    print_reboot_section();
+    Ok(())
+}
+
+fn print_reboot_section() {
+    println!("Reboot scheduler:");
+    let svc = Command::new("systemctl")
+        .args(["is-active", "awg2-reboot-scheduler"])
+        .output();
+    match svc {
+        Ok(out) if out.status.success() => println!("  service: active"),
+        Ok(_) => println!("  service: inactive"),
+        Err(_) => println!("  service: unknown"),
+    }
+
+    let plan = fs::read_to_string("/run/awg2-harness/reboot.plan").unwrap_or_default();
+    if plan.trim().is_empty() {
+        println!("  planned: n/a");
+    } else {
+        for line in plan.lines() {
+            println!("  {}", line);
+        }
+    }
+}
+
 fn handshake_is_online(epoch: Option<u64>, threshold_secs: u64) -> &'static str {
     if let Some(ts) = epoch {
         let now = SystemTime::now()

traffic_core/src/traffic.rs

@@ -972,13 +972,40 @@ fn parse_class_rate_ceil_map(raw: &str, prefix: &str) -> BTreeMap<String, ClassR
 }
 
 fn rate_ceil_matches_limit(v: &ClassRateCeil, limit: &str) -> bool {
-    normalize_tc_rate(&v.rate) == limit && normalize_tc_rate(&v.ceil) == limit
+    let Ok(limit_bps) = parse_tc_rate_bps(limit) else {
+        return false;
+    };
+    let Ok(rate_bps) = parse_tc_rate_bps(&v.rate) else {
+        return false;
+    };
+    let Ok(ceil_bps) = parse_tc_rate_bps(&v.ceil) else {
+        return false;
+    };
+    rate_bps == limit_bps && ceil_bps == limit_bps
 }
 
 fn normalize_tc_rate(v: &str) -> String {
     v.trim().to_ascii_lowercase().replace(' ', "")
 }
 
+fn parse_tc_rate_bps(v: &str) -> anyhow::Result<u64> {
+    let n = normalize_tc_rate(v);
+    let (num, mul) = if let Some(x) = n.strip_suffix("gbit") {
+        (x, 1_000_000_000u64)
+    } else if let Some(x) = n.strip_suffix("mbit") {
+        (x, 1_000_000u64)
+    } else if let Some(x) = n.strip_suffix("kbit") {
+        (x, 1_000u64)
+    } else if let Some(x) = n.strip_suffix("bit") {
+        (x, 1u64)
+    } else {
+        anyhow::bail!("unknown tc rate unit: {v}")
+    };
+    let base: u64 = num.parse().map_err(|_| anyhow::anyhow!("invalid tc rate: {v}"))?;
+    base.checked_mul(mul)
+        .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}"))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1094,4 +1121,14 @@ mod tests {
         assert!(rate_ceil_matches_limit(&rc, "70mbit"));
         assert!(!rate_ceil_matches_limit(&rc, "50mbit"));
     }
+
+    #[test]
+    fn tc_rate_numeric_equivalence() {
+        let rc = ClassRateCeil {
+            rate: "1Gbit".to_string(),
+            ceil: "1000Mbit".to_string(),
+        };
+        assert!(rate_ceil_matches_limit(&rc, "1000mbit"));
+        assert!(rate_ceil_matches_limit(&rc, "1gbit"));
+    }
 }

traffic_core/src/wg.rs

@@ -36,11 +36,10 @@ pub fn discover_peers_with_transport(
     transport: &dyn NsTransport,
     peer_cidr: &str,
 ) -> anyhow::Result<Vec<Peer>> {
-    let primary = transport
+    let mut peers = match transport.run(&["wg", "show", "all", "allowed-ips"]) {
-        .run(&["wg", "show", "all", "allowed-ips"])
+        Ok(primary) => parse_allowed_ips(&primary, peer_cidr)?,
-        .map_err(|e| anyhow::anyhow!("wg show all allowed-ips failed: {e}"))?;
+        Err(_) => Vec::new(),
-
+    };
-    let mut peers = parse_allowed_ips(&primary, peer_cidr)?;
     if peers.is_empty() {
         let fallback = transport
             .run(&["wg", "show"])
@@ -261,16 +260,15 @@ mod tests {
     }
 
     #[test]
-    fn command_failure_is_propagated() {
+    fn primary_failure_falls_back_to_wg_show() {
         let transport = TestTransport::new(vec![(
             "wg show all allowed-ips",
             Err(anyhow::anyhow!("wg broken")),
-        )]);
+        ), ("wg show", Ok("allowed ips: 10.8.1.4/32"))]);
 
-        let err = discover_peers_with_transport(&transport, "10.8.1.0/24").err();
+        let peers = discover_peers_with_transport(&transport, "10.8.1.0/24").expect("peers");
-        assert!(err.is_some());
+        assert_eq!(peers.len(), 1);
-        let msg = err.expect("err").to_string();
+        assert_eq!(peers[0].ipv4.to_string(), "10.8.1.4");
-        assert!(msg.contains("wg show all allowed-ips failed"));
     }
 
     #[test]