From cd45e9ff51f0886d798214b44fcceadf9d29886d Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Sun, 8 Feb 2026 13:44:09 -0600 Subject: [PATCH] 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 --- Cargo.lock | 72 +++++++++++++++++++ Cargo.toml | 2 + proto/agent.proto | 1 + src/main.rs | 1 + src/ping.rs | 68 ++++++++++++++++++ src/tui/events.rs | 4 ++ src/tui/state.rs | 3 + src/websocket_client.rs | 150 +++++++++++++++++++++++++++++++++++++++- 8 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 src/ping.rs diff --git a/Cargo.lock b/Cargo.lock index 820b068..c2e7b0e 100644 --- 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", diff --git a/Cargo.toml b/Cargo.toml index 533c726..f087b7c 100644 --- 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 } diff --git a/proto/agent.proto b/proto/agent.proto index d84d479..b4668e8 100644 --- 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 { diff --git a/src/main.rs b/src/main.rs index 5386d94..3c41a4a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ mod mikrotik; +mod ping; mod proto; pub mod secret; mod snmp; diff --git a/src/ping.rs b/src/ping.rs new file mode 100644 index 0000000..81837d6 --- /dev/null +++ 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 { + 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 + } +} diff --git a/src/tui/events.rs b/src/tui/events.rs index 6870157..a0d6012 100644 --- 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, diff --git a/src/tui/state.rs b/src/tui/state.rs index 76aa743..16b1cab 100644 --- 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, diff --git a/src/websocket_client.rs b/src/websocket_client.rs index 66c706b..92bf44a 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -26,7 +26,7 @@ type Result = std::result::Result 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, credential_test_tx: mpsc::Sender, credential_test_rx: mpsc::Receiver, + monitoring_check_tx: mpsc::Sender, + monitoring_check_rx: mpsc::Receiver, 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, +) -> 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,