feat: add agent-side ICMP ping health monitoring

Implements health monitoring in the agent using ICMP ping, allowing devices assigned to local (non-cloud) agents to be monitored. Changes: - Add surge-ping library for async ICMP ping operations - Create ping.rs module with ping_device() function (5s timeout) - Rename JobType::MONITOR to JobType::PING in protobuf - Add monitoring_check channels for result communication - Implement execute_ping_job() to handle PING jobs - Update TUI to display ping result events The agent now receives PING jobs from Phoenix, executes ICMP pings, and sends MonitoringCheck results back for storage in the database. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-08 13:44:09 -06:00 · 2026-02-08 13:44:09 -06:00 · cd45e9ff51
commit cd45e9ff51
parent b29a38edb5
8 changed files with 300 additions and 1 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -1238,6 +1238,12 @@ dependencies = [
 "polyval",
 ]

+[[package]]
+name = "glob"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
+
 [[package]]
 name = "group"
 version = "0.13.0"
@ -1779,6 +1785,12 @@ dependencies = [
 "memoffset",
 ]

+[[package]]
+name = "no-std-net"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43794a0ace135be66a25d3ae77d41b91615fb68ae937f904090203e81f755b65"
+
 [[package]]
 name = "nom"
 version = "7.1.3"
@ -2241,6 +2253,48 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"

+[[package]]
+name = "pnet_base"
+version = "0.34.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fe4cf6fb3ab38b68d01ab2aea03ed3d1132b4868fa4e06285f29f16da01c5f4c"
+dependencies = [
+ "no-std-net",
+]
+
+[[package]]
+name = "pnet_macros"
+version = "0.34.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "688b17499eee04a0408aca0aa5cba5fc86401d7216de8a63fdf7a4c227871804"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "regex",
+ "syn 2.0.114",
+]
+
+[[package]]
+name = "pnet_macros_support"
+version = "0.34.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eea925b72f4bd37f8eab0f221bbe4c78b63498350c983ffa9dd4bcde7e030f56"
+dependencies = [
+ "pnet_base",
+]
+
+[[package]]
+name = "pnet_packet"
+version = "0.34.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9a005825396b7fe7a38a8e288dbc342d5034dac80c15212436424fef8ea90ba"
+dependencies = [
+ "glob",
+ "pnet_base",
+ "pnet_macros",
+ "pnet_macros_support",
+]
+
 [[package]]
 name = "poly1305"
 version = "0.8.0"
@ -3159,6 +3213,22 @@ version = "2.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"

+[[package]]
+name = "surge-ping"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "30498e9c9feba213c3df6ed675bdf75519ccbee493517e7225305898c86cac05"
+dependencies = [
+ "hex",
+ "parking_lot",
+ "pnet_packet",
+ "rand 0.9.2",
+ "socket2",
+ "thiserror 1.0.69",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "syn"
 version = "1.0.109"
@ -3426,11 +3496,13 @@ dependencies = [
 "prost",
 "prost-build",
 "prost-types",
+ "rand 0.8.5",
 "ratatui",
 "russh",
 "serde",
 "serde_json",
 "snmp2",
+ "surge-ping",
 "thiserror 2.0.18",
 "tokio",
 "tokio-rustls",
--- a/Cargo.toml
+++ b/Cargo.toml
@ -23,6 +23,8 @@ home = "=0.5.12"
 zeroize = { version = "1", features = ["derive"] }
 hostname = "0.4"
 anyhow = "1.0"
+surge-ping = "0.8"
+rand = "0.8"

 # TUI dependencies (optional feature)
 ratatui = { version = "0.30", optional = true }
--- a/proto/agent.proto
+++ b/proto/agent.proto
@ -115,6 +115,7 @@ enum JobType {
  POLL = 1;
  MIKROTIK = 2;
  TEST_CREDENTIALS = 3;
+  PING = 4;
 }

 enum QueryType {
--- a/src/main.rs
+++ b/src/main.rs
@ -1,4 +1,5 @@
 mod mikrotik;
+mod ping;
 mod proto;
 pub mod secret;
 mod snmp;
--- a/src/ping.rs
+++ b/src/ping.rs
@ -0,0 +1,68 @@
+use anyhow::{Context, Result};
+use std::net::IpAddr;
+use std::time::Duration;
+use surge_ping::{Client, Config, PingIdentifier, PingSequence};
+use tokio::time::timeout;
+
+/// Ping a device and return response time in milliseconds.
+///
+/// Returns Ok(response_time_ms) on success, Err on failure.
+pub async fn ping_device(ip_address: &str, timeout_ms: u64) -> Result<f64> {
+    let ip: IpAddr = ip_address
+        .parse()
+        .context(format!("Invalid IP address: {}", ip_address))?;
+
+    // Create ICMP client
+    let client = Client::new(&Config::default())
+        .context("Failed to create ping client - may require root/admin privileges")?;
+
+    // Send ping with timeout
+    let payload = [0; 56]; // Standard ping payload size
+    let identifier = PingIdentifier(rand::random());
+    let sequence = PingSequence(1);
+
+    let ping_future = async {
+        match ip {
+            IpAddr::V4(addr) => {
+                client
+                    .pinger(addr.into(), identifier)
+                    .await
+                    .ping(sequence, &payload)
+                    .await
+            }
+            IpAddr::V6(addr) => {
+                client
+                    .pinger(addr.into(), identifier)
+                    .await
+                    .ping(sequence, &payload)
+                    .await
+            }
+        }
+    };
+
+    // Apply timeout
+    let (_, duration) = timeout(Duration::from_millis(timeout_ms), ping_future)
+        .await
+        .context("Ping timeout")?
+        .context("Ping failed")?;
+
+    // Convert Duration to milliseconds (f64 for sub-millisecond precision)
+    let ms = duration.as_secs_f64() * 1000.0;
+
+    Ok(ms)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    #[ignore] // Requires network access and elevated privileges
+    async fn test_ping_localhost() {
+        let result = ping_device("127.0.0.1", 5000).await;
+        assert!(result.is_ok());
+        let response_time = result.unwrap();
+        assert!(response_time > 0.0);
+        assert!(response_time < 100.0); // Localhost should be fast
+    }
+}
--- a/src/tui/events.rs
+++ b/src/tui/events.rs
@ -29,6 +29,10 @@ pub enum AgentEvent {
        device_id: String,
        sentence_count: usize,
    },
+    MonitoringCheckSent {
+        device_id: String,
+        status: String,
+    },
    PollerCreated {
        device_ip: String,
        total_count: usize,
--- a/src/tui/state.rs
+++ b/src/tui/state.rs
@ -105,6 +105,9 @@ impl AgentState {
                    device_id, sentence_count
                )
            }
+            AgentEvent::MonitoringCheckSent { device_id, status } => {
+                format!("Monitoring check sent for {} (status: {})", device_id, status)
+            }
            AgentEvent::PollerCreated {
                device_ip,
                total_count,
--- a/src/websocket_client.rs
+++ b/src/websocket_client.rs
@ -26,7 +26,7 @@ type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>

 use crate::proto::agent::{
    AgentHeartbeat, AgentJob, AgentJobList, CredentialTestResult, JobType, MikrotikResult,
-    MikrotikSentence, QueryType, SnmpResult,
+    MikrotikSentence, MonitoringCheck, QueryType, SnmpResult,
 };
 use crate::snmp::{DeviceConfig, PollerRegistry, SnmpValue};

@ -62,6 +62,8 @@ pub struct AgentClient {
    mikrotik_result_rx: mpsc::Receiver<MikrotikResult>,
    credential_test_tx: mpsc::Sender<CredentialTestResult>,
    credential_test_rx: mpsc::Receiver<CredentialTestResult>,
+    monitoring_check_tx: mpsc::Sender<MonitoringCheck>,
+    monitoring_check_rx: mpsc::Receiver<MonitoringCheck>,
    poller_registry: PollerRegistry,
    /// Counter for Phoenix transport heartbeat refs
    phx_heartbeat_ref: u64,
@ -139,6 +141,7 @@ impl AgentClient {
        let (result_tx, result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
        let (mikrotik_result_tx, mikrotik_result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
        let (credential_test_tx, credential_test_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
+        let (monitoring_check_tx, monitoring_check_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);

        // Split the WebSocket stream so reads and writes can proceed concurrently.
        // The write half is owned by a dedicated writer task.
@ -175,6 +178,8 @@ impl AgentClient {
            mikrotik_result_rx,
            credential_test_tx,
            credential_test_rx,
+            monitoring_check_tx,
+            monitoring_check_rx,
            poller_registry: PollerRegistry::new(),
            phx_heartbeat_ref: 0,
            cached_hostname: get_hostname(),
@ -228,6 +233,7 @@ impl AgentClient {
        let (result_tx, result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
        let (mikrotik_result_tx, mikrotik_result_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
        let (credential_test_tx, credential_test_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);
+        let (monitoring_check_tx, monitoring_check_rx) = mpsc::channel(RESULT_CHANNEL_CAPACITY);

        // Split the WebSocket stream so reads and writes can proceed concurrently.
        // The write half is owned by a dedicated writer task.
@ -264,6 +270,8 @@ impl AgentClient {
            mikrotik_result_rx,
            credential_test_tx,
            credential_test_rx,
+            monitoring_check_tx,
+            monitoring_check_rx,
            poller_registry: PollerRegistry::new(),
            phx_heartbeat_ref: 0,
            cached_hostname: get_hostname(),
@ -380,6 +388,20 @@ impl AgentClient {
                    }
                }

+                // Receive monitoring check results from job tasks
+                Some(monitoring_check) = self.monitoring_check_rx.recv() => {
+                    #[cfg(feature = "tui")]
+                    let device_id = monitoring_check.device_id.clone();
+
+                    if let Err(e) = self.send_monitoring_check(monitoring_check).await {
+                        tracing::error!("Error sending monitoring check: {}", e);
+                        #[cfg(feature = "tui")]
+                        self.publish_event(crate::tui::AgentEvent::Error {
+                            message: format!("Monitoring check send failed for {}: {}", device_id, e),
+                        });
+                    }
+                }
+
                // Send periodic heartbeats
                _ = heartbeat_interval.tick() => {
                    if let Err(e) = self.send_heartbeat().await {
@ -573,6 +595,42 @@ impl AgentClient {
                        }
                    });
                }
+                JobType::Ping => {
+                    // Execute ICMP ping health check
+                    let monitoring_check_tx = self.monitoring_check_tx.clone();
+                    #[cfg(feature = "tui")]
+                    let event_bus = self.event_bus.clone();
+                    let job_id = job.job_id.clone();
+                    let device_id = job.device_id.clone();
+
+                    tokio::spawn(async move {
+                        let start_time = std::time::Instant::now();
+                        match execute_ping_job(job, monitoring_check_tx).await {
+                            Ok(_) =>
+                            {
+                                #[cfg(feature = "tui")]
+                                if let Some(ref bus) = event_bus {
+                                    let _ = bus.send(crate::tui::AgentEvent::JobCompleted {
+                                        job_id,
+                                        device_id,
+                                        duration_ms: start_time.elapsed().as_millis() as u64,
+                                    });
+                                }
+                            }
+                            Err(e) => {
+                                tracing::error!("Ping job execution failed: {}", e);
+                                #[cfg(feature = "tui")]
+                                if let Some(ref bus) = event_bus {
+                                    let _ = bus.send(crate::tui::AgentEvent::JobFailed {
+                                        job_id,
+                                        device_id,
+                                        error: e.to_string(),
+                                    });
+                                }
+                            }
+                        }
+                    });
+                }
                _ => {
                    // Execute SNMP job (discovery or polling)
                    let result_tx = self.result_tx.clone();
@ -765,6 +823,38 @@ impl AgentClient {
        );
        Ok(())
    }
+
+    /// Send monitoring check result to server.
+    async fn send_monitoring_check(&mut self, result: MonitoringCheck) -> Result<()> {
+        let binary = result.encode_to_vec();
+
+        let msg = PhoenixMessage {
+            topic: format!("agent:{}", self.agent_id),
+            event: "monitoring_check".to_string(),
+            payload: serde_json::json!({"binary": base64_encode(&binary)}),
+            reference: None,
+        };
+
+        let text = serde_json::to_string(&msg)?;
+        self.ws_write_tx
+            .send(WsMessage::Text(text.into()))
+            .await
+            .map_err(|e| format!("Writer task closed: {}", e))?;
+
+        tracing::debug!(
+            "Sent monitoring check for device {} (status: {})",
+            result.device_id,
+            result.status
+        );
+
+        #[cfg(feature = "tui")]
+        self.publish_event(crate::tui::AgentEvent::MonitoringCheckSent {
+            device_id: result.device_id.clone(),
+            status: result.status.clone(),
+        });
+
+        Ok(())
+    }
 }

 /// Dedicated writer task that owns the WebSocket write half.
@ -1082,6 +1172,64 @@ async fn execute_credential_test(
    Ok(())
 }

+/// Execute a ping job using ICMP ping to check device health.
+async fn execute_ping_job(
+    job: AgentJob,
+    result_tx: mpsc::Sender<MonitoringCheck>,
+) -> Result<()> {
+    let device_id = job.device_id.clone();
+    let snmp_device = job.snmp_device.ok_or("Job missing SNMP device info")?;
+    let ip_address = &snmp_device.ip;
+
+    // Use 5-second timeout for pings (same as Phoenix DeviceMonitorWorker)
+    let timeout_ms = 5000;
+
+    tracing::debug!("Executing health check for device {} at {}", device_id, ip_address);
+
+    let timestamp = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)?
+        .as_secs() as i64;
+
+    // Execute ping
+    let result = match crate::ping::ping_device(ip_address, timeout_ms).await {
+        Ok(response_time_ms) => {
+            tracing::info!(
+                "✓ Device {} is up (response time: {:.1}ms)",
+                device_id,
+                response_time_ms
+            );
+
+            MonitoringCheck {
+                device_id: device_id.clone(),
+                status: "success".to_string(),
+                response_time_ms,
+                timestamp,
+            }
+        }
+        Err(e) => {
+            tracing::warn!("✗ Device {} is down: {}", device_id, e);
+
+            MonitoringCheck {
+                device_id: device_id.clone(),
+                status: "failure".to_string(),
+                response_time_ms: 0.0,
+                timestamp,
+            }
+        }
+    };
+
+    // Send result back to main client task
+    if let Err(e) = result_tx.send(result).await {
+        tracing::warn!(
+            "Failed to send monitoring check for device {}: channel closed",
+            device_id
+        );
+        return Err(format!("Result channel closed: {}", e).into());
+    }
+
+    Ok(())
+}
+
 /// Execute a MikroTik API job and collect results.
 async fn execute_mikrotik_job(
    job: AgentJob,