diff --git a/src/poller/executor.rs b/src/poller/executor.rs index cb9c63b..a77b381 100644 --- a/src/poller/executor.rs +++ b/src/poller/executor.rs @@ -42,6 +42,7 @@ use crate::metrics::Timestamp; use log::{error, info, warn}; /// Executor handles polling individual pieces of equipment +#[derive(Clone)] pub struct Executor { snmp_client: SnmpClient, storage: Storage, diff --git a/src/poller/scheduler.rs b/src/poller/scheduler.rs index 85a3a15..7d8a390 100644 --- a/src/poller/scheduler.rs +++ b/src/poller/scheduler.rs @@ -215,27 +215,45 @@ impl Scheduler { let poll_times = self.storage.get_all_last_poll_times()?; - for equipment in &config.equipment { - if !equipment.snmp.enabled { - continue; - } - - // Check if it's time to poll this equipment - let should_poll = match poll_times.get(&equipment.id) { - Some(last_poll) => { - let elapsed = last_poll.elapsed_secs() as u64; - elapsed >= equipment.poll_interval_seconds + // Collect equipment that needs polling + let equipment_to_poll: Vec<_> = config + .equipment + .iter() + .filter(|eq| eq.snmp.enabled) + .filter(|eq| { + match poll_times.get(&eq.id) { + Some(last_poll) => { + let elapsed = last_poll.elapsed_secs() as u64; + elapsed >= eq.poll_interval_seconds + } + None => true, // Never polled before } - None => true, // Never polled before - }; + }) + .collect(); - if should_poll { + if equipment_to_poll.is_empty() { + return Ok(()); + } + + info!( + "Polling {} equipment items in parallel", + equipment_to_poll.len() + ); + + // Spawn parallel polling tasks + let mut tasks = Vec::new(); + for equipment in equipment_to_poll { + let executor = self.executor.clone(); + let storage = self.storage.clone(); + let equipment = equipment.clone(); + + let task = tokio::spawn(async move { info!("Polling equipment: {}", equipment.name); // Poll sensors and interfaces in parallel let (sensor_result, interface_result) = tokio::join!( - self.executor.poll_sensors(equipment), - self.executor.poll_interfaces(equipment) + executor.poll_sensors(&equipment), + executor.poll_interfaces(&equipment) ); if let Err(e) = sensor_result { @@ -247,9 +265,18 @@ impl Scheduler { } // Update last poll time - if let Err(e) = self.storage.update_last_poll_time(&equipment.id) { + if let Err(e) = storage.update_last_poll_time(&equipment.id) { error!("Failed to update last poll time: {}", e); } + }); + + tasks.push(task); + } + + // Wait for all polling tasks to complete + for task in tasks { + if let Err(e) = task.await { + error!("Polling task failed: {}", e); } } diff --git a/src/snmp/client.rs b/src/snmp/client.rs index f21aed6..170e4fb 100644 --- a/src/snmp/client.rs +++ b/src/snmp/client.rs @@ -3,7 +3,7 @@ use snmp::SyncSession; use std::time::Duration; /// SNMP client for polling devices -#[derive(Debug)] +#[derive(Debug, Clone, Copy)] pub struct SnmpClient; impl SnmpClient {