diff --git a/proto/agent.proto b/proto/agent.proto index 2e12807..428f861 100644 --- a/proto/agent.proto +++ b/proto/agent.proto @@ -17,6 +17,8 @@ message Device { uint32 poll_interval_seconds = 5; repeated Sensor sensors = 6; repeated Interface interfaces = 7; + bool monitoring_enabled = 8; + uint32 check_interval_seconds = 9; } message SnmpConfig { @@ -51,6 +53,7 @@ message Metric { SensorReading sensor_reading = 1; InterfaceStat interface_stat = 2; NeighborDiscovery neighbor_discovery = 3; + MonitoringCheck monitoring_check = 4; } } @@ -86,6 +89,13 @@ message NeighborDiscovery { int64 timestamp = 11; // Unix timestamp in seconds } +message MonitoringCheck { + string device_id = 1; + string status = 2; // "success" or "failure" + double response_time_ms = 3; // Optional - only present on success + int64 timestamp = 4; // Unix timestamp in seconds +} + // Heartbeat metadata message HeartbeatMetadata { string version = 1; @@ -126,6 +136,8 @@ message SnmpDevice { string community = 2; string version = 3; uint32 port = 4; + bool monitoring_enabled = 5; + uint32 check_interval_seconds = 6; } message SnmpQuery { diff --git a/src/websocket_client.rs b/src/websocket_client.rs index 0616e31..54092c0 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -19,7 +19,10 @@ use tokio_tungstenite::{ connect_async, tungstenite::protocol::Message as WsMessage, MaybeTlsStream, WebSocketStream, }; -use crate::proto::agent::{AgentHeartbeat, AgentJob, AgentJobList, JobType, QueryType, SnmpResult}; +use crate::ping::ping; +use crate::proto::agent::{ + AgentHeartbeat, AgentJob, AgentJobList, JobType, MonitoringCheck, QueryType, SnmpResult, +}; use crate::snmp::{SnmpClient, SnmpValue}; /// Phoenix channel message format (JSON wrapper around binary protobuf). @@ -38,6 +41,8 @@ pub struct AgentClient { agent_id: String, result_tx: mpsc::UnboundedSender, result_rx: mpsc::UnboundedReceiver, + monitoring_check_tx: mpsc::UnboundedSender, + monitoring_check_rx: mpsc::UnboundedReceiver, } impl AgentClient { @@ -69,6 +74,7 @@ impl AgentClient { let agent_id = generate_agent_id(); let (result_tx, result_rx) = mpsc::unbounded_channel(); + let (monitoring_check_tx, monitoring_check_rx) = mpsc::unbounded_channel(); // Join Phoenix channel with token in payload let join_msg = PhoenixMessage { @@ -90,6 +96,8 @@ impl AgentClient { agent_id, result_tx, result_rx, + monitoring_check_tx, + monitoring_check_rx, }) } @@ -135,6 +143,11 @@ impl AgentClient { self.send_result(result).await?; } + // Receive monitoring checks from ping tasks + Some(check) = self.monitoring_check_rx.recv() => { + self.send_monitoring_check(check).await?; + } + // Send periodic heartbeats _ = heartbeat_interval.tick() => { self.send_heartbeat().await?; @@ -189,13 +202,55 @@ impl AgentClient { let job_type = JobType::try_from(job.job_type).unwrap_or(JobType::Poll); log::info!("Executing job: {} (type: {:?})", job.job_id, job_type); - // Spawn task to execute job + // Extract monitoring configuration before moving job + let monitoring_info = job.snmp_device.as_ref().and_then(|snmp_device| { + if snmp_device.monitoring_enabled { + Some(( + snmp_device.ip.clone(), + if snmp_device.check_interval_seconds > 0 { + snmp_device.check_interval_seconds as u64 + } else { + 60 + }, + )) + } else { + None + } + }); + + // Spawn task to execute SNMP job let result_tx = self.result_tx.clone(); + let monitoring_check_tx = self.monitoring_check_tx.clone(); + let device_id = job.device_id.clone(); + tokio::spawn(async move { if let Err(e) = execute_job(job, result_tx).await { log::error!("Job execution failed: {}", e); } }); + + // Start monitoring task if enabled + if let Some((ip, interval_seconds)) = monitoring_info { + log::info!( + "Starting ICMP monitoring for device {} every {} seconds", + device_id, + interval_seconds + ); + + let device_id_clone = device_id.clone(); + tokio::spawn(async move { + if let Err(e) = run_monitoring_task( + device_id_clone, + ip, + interval_seconds, + monitoring_check_tx, + ) + .await + { + log::error!("Monitoring task failed: {}", e); + } + }); + } } Ok(()) @@ -244,6 +299,33 @@ impl AgentClient { log::debug!("Sent SNMP result for device {}", result.device_id); Ok(()) } + + /// Send monitoring check result to server. + async fn send_monitoring_check(&mut self, check: MonitoringCheck) -> Result<()> { + let binary = check.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_stream.send(WsMessage::Text(text)).await?; + + log::debug!( + "Sent monitoring check for device {}: {}", + check.device_id, + check.status + ); + Ok(()) + } + + /// Get the monitoring check sender for spawning ping tasks. + pub fn monitoring_check_sender(&self) -> mpsc::UnboundedSender { + self.monitoring_check_tx.clone() + } } /// Execute an SNMP job and collect results. @@ -367,3 +449,70 @@ fn get_local_ip() -> Option { // Could use local_ip_address crate or parse network interfaces None } + +/// Run continuous ICMP monitoring for a device. +async fn run_monitoring_task( + device_id: String, + ip: String, + interval_seconds: u64, + check_tx: mpsc::UnboundedSender, +) -> Result<()> { + let mut interval_timer = interval(Duration::from_secs(interval_seconds)); + + loop { + interval_timer.tick().await; + + let ip_addr: std::net::IpAddr = match ip.parse() { + Ok(addr) => addr, + Err(e) => { + log::error!("Invalid IP address for device {}: {}", device_id, e); + continue; + } + }; + + let timeout = Duration::from_secs(5); + + match ping(ip_addr, timeout).await { + Ok(rtt) => { + let check = MonitoringCheck { + device_id: device_id.clone(), + status: "success".to_string(), + response_time_ms: rtt.as_secs_f64() * 1000.0, + timestamp: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() as i64, + }; + + if let Err(e) = check_tx.send(check) { + log::error!("Failed to send monitoring check: {}", e); + break; + } + + log::debug!( + "ICMP ping successful for {}: {:.2}ms", + ip, + rtt.as_secs_f64() * 1000.0 + ); + } + Err(e) => { + let check = MonitoringCheck { + device_id: device_id.clone(), + status: "failure".to_string(), + response_time_ms: 0.0, + timestamp: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_secs() as i64, + }; + + if let Err(send_err) = check_tx.send(check) { + log::error!("Failed to send monitoring check: {}", send_err); + break; + } + + log::warn!("ICMP ping failed for {}: {}", ip, e); + } + } + } + + Ok(()) +}