Agent overhaul with zero persistence

docker-compose.example.yml

@@ -0,0 +1,41 @@
+version: '3.8'
+
+services:
+  towerops-agent:
+    image: registry.gitlab.com/towerops/towerops-agent:latest
+    container_name: towerops-agent
+    restart: unless-stopped
+
+    environment:
+      # Required: Your Towerops API URL
+      - TOWEROPS_API_URL=https://app.towerops.com
+
+      # Required: Agent authentication token (from Towerops web UI)
+      # Get this from: Organization > Agents > Create New Agent
+      - TOWEROPS_AGENT_TOKEN=your-agent-token-here
+
+      # Optional: How often to refresh configuration (seconds)
+      # Default: 300 (5 minutes)
+      # - CONFIG_REFRESH_SECONDS=300
+
+      # Optional: Database path for metrics buffering
+      # Default: /data/towerops-agent.db
+      # - DATABASE_PATH=/data/towerops-agent.db
+
+      # Optional: Log level (error, warn, info, debug, trace)
+      # Default: info
+      # - RUST_LOG=info
+
+    volumes:
+      # Persistent storage for metrics buffering
+      - ./data:/data
+
+    # Optional: Resource limits
+    deploy:
+      resources:
+        limits:
+          cpus: '0.5'
+          memory: 512M
+        reservations:
+          cpus: '0.1'
+          memory: 128M

proto/agent.proto

@@ -136,8 +136,6 @@ message SnmpDevice {
   string community = 2;
   string version = 3;
   uint32 port = 4;
-  bool monitoring_enabled = 5;
-  uint32 check_interval_seconds = 6;
 }
 
 message SnmpQuery {

src/api_client.rs

@@ -0,0 +1,171 @@
+use crate::config::{AgentConfig, HeartbeatMetadata};
+use crate::metrics::Metric;
+use crate::proto::agent;
+use prost::Message;
+use serde_json::json;
+use std::time::Duration;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ApiError {
+    #[error("HTTP request failed: {0}")]
+    RequestFailed(String),
+
+    #[error("HTTP status error: {0}")]
+    StatusError(u16),
+
+    #[error("JSON parsing error: {0}")]
+    JsonError(#[from] std::io::Error),
+
+    #[error("Task join error: {0}")]
+    JoinError(String),
+}
+
+pub type Result<T> = std::result::Result<T, ApiError>;
+
+/// API client for communicating with the Towerops server
+#[derive(Clone)]
+pub struct ApiClient {
+    base_url: String,
+    token: String,
+}
+
+impl ApiClient {
+    /// Create a new API client
+    pub fn new(base_url: String, token: String) -> Result<Self> {
+        Ok(Self { base_url, token })
+    }
+
+    /// Fetch configuration from the API
+    pub async fn fetch_config(&self) -> Result<AgentConfig> {
+        let url = format!("{}/api/v1/agent/config", self.base_url);
+        let token = self.token.clone();
+
+        let config = tokio::task::spawn_blocking(move || {
+            let response = ureq::get(&url)
+                .set("Authorization", &format!("Bearer {}", token))
+                .timeout(Duration::from_secs(30))
+                .call()
+                .map_err(|e| ApiError::RequestFailed(e.to_string()))?;
+
+            let status = response.status();
+            if status != 200 {
+                return Err(ApiError::StatusError(status));
+            }
+
+            let config: AgentConfig = response
+                .into_json()
+                .map_err(|e| ApiError::RequestFailed(e.to_string()))?;
+
+            Ok(config)
+        })
+        .await
+        .map_err(|e| ApiError::JoinError(e.to_string()))??;
+
+        Ok(config)
+    }
+
+    /// Submit metrics to the API using Protocol Buffers
+    pub async fn submit_metrics(&self, metrics: Vec<Metric>) -> Result<()> {
+        if metrics.is_empty() {
+            return Ok(());
+        }
+
+        let url = format!("{}/api/v1/agent/metrics", self.base_url);
+        let token = self.token.clone();
+
+        tokio::task::spawn_blocking(move || {
+            // Convert metrics to protobuf
+            let proto_metrics: Vec<agent::Metric> = metrics
+                .into_iter()
+                .map(|m| convert_metric_to_proto(&m))
+                .collect();
+
+            let batch = agent::MetricBatch {
+                metrics: proto_metrics,
+            };
+
+            // Encode to bytes
+            let mut buf = Vec::new();
+            batch
+                .encode(&mut buf)
+                .map_err(|e| ApiError::RequestFailed(format!("Protobuf encoding error: {}", e)))?;
+
+            // Send as protobuf
+            let response = ureq::post(&url)
+                .set("Authorization", &format!("Bearer {}", token))
+                .set("Content-Type", "application/x-protobuf")
+                .timeout(Duration::from_secs(30))
+                .send_bytes(&buf)
+                .map_err(|e| ApiError::RequestFailed(e.to_string()))?;
+
+            let status = response.status();
+            if status != 200 {
+                return Err(ApiError::StatusError(status));
+            }
+
+            Ok(())
+        })
+        .await
+        .map_err(|e| ApiError::JoinError(e.to_string()))??;
+
+        Ok(())
+    }
+
+    /// Send heartbeat to the API
+    pub async fn heartbeat(&self, metadata: HeartbeatMetadata) -> Result<()> {
+        let url = format!("{}/api/v1/agent/heartbeat", self.base_url);
+        let token = self.token.clone();
+
+        tokio::task::spawn_blocking(move || {
+            let response = ureq::post(&url)
+                .set("Authorization", &format!("Bearer {}", token))
+                .timeout(Duration::from_secs(30))
+                .send_json(&metadata)
+                .map_err(|e| ApiError::RequestFailed(e.to_string()))?;
+
+            let status = response.status();
+            if status != 200 {
+                return Err(ApiError::StatusError(status));
+            }
+
+            Ok(())
+        })
+        .await
+        .map_err(|e| ApiError::JoinError(e.to_string()))??;
+
+        Ok(())
+    }
+}
+
+/// Convert internal Metric type to protobuf Metric
+fn convert_metric_to_proto(metric: &Metric) -> agent::Metric {
+    use crate::metrics::Metric as M;
+
+    match metric {
+        M::SensorReading(sr) => agent::Metric {
+            metric_type: Some(agent::metric::MetricType::SensorReading(
+                agent::SensorReading {
+                    sensor_id: sr.sensor_id.clone(),
+                    value: sr.value,
+                    status: sr.status.clone(),
+                    timestamp: sr.timestamp.to_unix_timestamp(),
+                },
+            )),
+        },
+        M::InterfaceStat(is) => agent::Metric {
+            metric_type: Some(agent::metric::MetricType::InterfaceStat(
+                agent::InterfaceStat {
+                    interface_id: is.interface_id.clone(),
+                    if_in_octets: is.if_in_octets,
+                    if_out_octets: is.if_out_octets,
+                    if_in_errors: is.if_in_errors,
+                    if_out_errors: is.if_out_errors,
+                    if_in_discards: is.if_in_discards,
+                    if_out_discards: is.if_out_discards,
+                    timestamp: is.timestamp.to_unix_timestamp(),
+                },
+            )),
+        },
+    }
+}

src/buffer/mod.rs

@@ -0,0 +1,3 @@
+mod storage;
+
+pub use storage::{Storage, StorageError};

src/buffer/storage.rs

@@ -0,0 +1,180 @@
+use crate::metrics::{Metric, Timestamp};
+use rusqlite::{params, Connection};
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum StorageError {
+    #[error("Database error: {0}")]
+    Database(#[from] rusqlite::Error),
+
+    #[error("Serialization error: {0}")]
+    Serialization(#[from] serde_json::Error),
+
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+}
+
+pub type Result<T> = std::result::Result<T, StorageError>;
+
+/// SQLite storage for buffering metrics when API is unavailable
+#[derive(Clone)]
+pub struct Storage {
+    conn: Arc<Mutex<Connection>>,
+}
+
+impl Storage {
+    /// Create a new storage instance
+    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
+        let conn = Connection::open(path)?;
+
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS metrics (
+                id INTEGER PRIMARY KEY,
+                metric_type TEXT NOT NULL,
+                data TEXT NOT NULL,
+                timestamp TEXT NOT NULL,
+                sent INTEGER DEFAULT 0,
+                created_at TEXT DEFAULT CURRENT_TIMESTAMP
+            )",
+            [],
+        )?;
+
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_metrics_sent ON metrics(sent, created_at)",
+            [],
+        )?;
+
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS last_poll_times (
+                equipment_id TEXT PRIMARY KEY,
+                last_poll_time TEXT NOT NULL
+            )",
+            [],
+        )?;
+
+        Ok(Self {
+            conn: Arc::new(Mutex::new(conn)),
+        })
+    }
+
+    /// Store a metric in the buffer
+    pub fn store_metric(&self, metric: &Metric) -> Result<()> {
+        let data = serde_json::to_string(metric)?;
+        let conn = self.conn.lock().unwrap();
+
+        conn.execute(
+            "INSERT INTO metrics (metric_type, data, timestamp) VALUES (?1, ?2, ?3)",
+            params![metric.metric_type(), data, metric.timestamp().to_rfc3339()],
+        )?;
+
+        Ok(())
+    }
+
+    /// Get pending metrics that haven't been sent yet
+    pub fn get_pending_metrics(&self, limit: usize) -> Result<Vec<(i64, Metric)>> {
+        let conn = self.conn.lock().unwrap();
+
+        let mut stmt = conn
+            .prepare("SELECT id, data FROM metrics WHERE sent = 0 ORDER BY created_at LIMIT ?1")?;
+
+        let metrics = stmt
+            .query_map([limit], |row| {
+                let id: i64 = row.get(0)?;
+                let data: String = row.get(1)?;
+                Ok((id, data))
+            })?
+            .filter_map(|r| r.ok())
+            .filter_map(|(id, data)| {
+                serde_json::from_str::<Metric>(&data)
+                    .ok()
+                    .map(|metric| (id, metric))
+            })
+            .collect();
+
+        Ok(metrics)
+    }
+
+    /// Mark metrics as sent
+    pub fn mark_metrics_sent(&self, ids: &[i64]) -> Result<()> {
+        if ids.is_empty() {
+            return Ok(());
+        }
+
+        let conn = self.conn.lock().unwrap();
+        let placeholders = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
+        let query = format!("UPDATE metrics SET sent = 1 WHERE id IN ({})", placeholders);
+
+        let params: Vec<_> = ids.iter().map(|id| id as &dyn rusqlite::ToSql).collect();
+        conn.execute(&query, params.as_slice())?;
+
+        Ok(())
+    }
+
+    /// Clean up old metrics that have been sent
+    pub fn cleanup_old_metrics(&self) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+
+        conn.execute(
+            "DELETE FROM metrics WHERE sent = 1 AND created_at < datetime('now', '-24 hours')",
+            [],
+        )?;
+
+        Ok(())
+    }
+
+    /// Get the last poll time for equipment
+    #[allow(dead_code)] // Used in full SNMP implementation
+    pub fn get_last_poll_time(&self, equipment_id: &str) -> Result<Option<Timestamp>> {
+        let conn = self.conn.lock().unwrap();
+
+        let result: std::result::Result<String, _> = conn.query_row(
+            "SELECT last_poll_time FROM last_poll_times WHERE equipment_id = ?1",
+            params![equipment_id],
+            |row| row.get(0),
+        );
+
+        match result {
+            Ok(_timestamp_str) => {
+                // Return a Timestamp - parsing from string not implemented yet
+                // For now just return None since this function isn't used yet
+                Ok(None)
+            }
+            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
+            Err(e) => Err(e.into()),
+        }
+    }
+
+    /// Update the last poll time for equipment
+    pub fn update_last_poll_time(&self, equipment_id: &str) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        let now = Timestamp::now().to_rfc3339();
+
+        conn.execute(
+            "INSERT OR REPLACE INTO last_poll_times (equipment_id, last_poll_time) VALUES (?1, ?2)",
+            params![equipment_id, now],
+        )?;
+
+        Ok(())
+    }
+
+    /// Get all last poll times
+    pub fn get_all_last_poll_times(&self) -> Result<std::collections::HashMap<String, Timestamp>> {
+        let conn = self.conn.lock().unwrap();
+
+        let mut stmt = conn.prepare("SELECT equipment_id, last_poll_time FROM last_poll_times")?;
+
+        let times: std::collections::HashMap<String, Timestamp> = stmt
+            .query_map([], |row| {
+                let equipment_id: String = row.get(0)?;
+                let _timestamp_str: String = row.get(1)?;
+                Ok(equipment_id)
+            })?
+            .filter_map(|r| r.ok())
+            .map(|id| (id, Timestamp::now())) // Simplified - not parsing timestamps yet
+            .collect();
+
+        Ok(times)
+    }
+}

src/config.rs

@@ -0,0 +1,58 @@
+use serde::{Deserialize, Serialize};
+
+/// Configuration received from the API
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct AgentConfig {
+    pub version: String,
+    pub poll_interval_seconds: u64,
+    pub equipment: Vec<EquipmentConfig>,
+}
+
+/// Configuration for a single piece of equipment
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct EquipmentConfig {
+    pub id: String,
+    pub name: String,
+    pub ip_address: String,
+    pub snmp: SnmpConfig,
+    pub poll_interval_seconds: u64,
+    pub sensors: Vec<SensorConfig>,
+    pub interfaces: Vec<InterfaceConfig>,
+}
+
+/// SNMP configuration
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct SnmpConfig {
+    pub enabled: bool,
+    pub version: String,
+    pub community: String,
+    pub port: u16,
+}
+
+/// Sensor configuration
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct SensorConfig {
+    pub id: String,
+    #[serde(rename = "type")]
+    pub sensor_type: String,
+    pub oid: String,
+    pub divisor: Option<i32>,
+    pub unit: Option<String>,
+    pub metadata: Option<serde_json::Value>,
+}
+
+/// Interface configuration
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct InterfaceConfig {
+    pub id: String,
+    pub if_index: i32,
+    pub if_name: String,
+}
+
+/// Heartbeat metadata sent to the API
+#[derive(Debug, Serialize)]
+pub struct HeartbeatMetadata {
+    pub version: String,
+    pub hostname: String,
+    pub uptime_seconds: u64,
+}

src/main.rs

@@ -1,4 +1,3 @@
-mod ping;
 mod proto;
 mod snmp;
 mod version;

src/metrics/mod.rs

@@ -0,0 +1,149 @@
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+/// Timestamp that serializes to RFC3339 format
+#[derive(Debug, Clone, Copy)]
+pub struct Timestamp {
+    secs: i64,
+}
+
+impl Timestamp {
+    pub fn now() -> Self {
+        let duration = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("Time went backwards");
+        Self {
+            secs: duration.as_secs() as i64,
+        }
+    }
+
+    pub fn elapsed_secs(&self) -> i64 {
+        Self::now().secs - self.secs
+    }
+
+    pub fn to_unix_timestamp(&self) -> i64 {
+        self.secs
+    }
+
+    pub fn to_rfc3339(self) -> String {
+        // Convert Unix timestamp to RFC3339 format
+        // This is a simplified implementation that produces UTC timestamps
+        let secs = self.secs;
+        let days = secs / 86400;
+        let rem_secs = secs % 86400;
+        let hours = rem_secs / 3600;
+        let minutes = (rem_secs % 3600) / 60;
+        let seconds = rem_secs % 60;
+
+        // Calculate year/month/day from days since epoch (1970-01-01)
+        let (year, month, day) = days_to_ymd(days);
+
+        format!(
+            "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
+            year, month, day, hours, minutes, seconds
+        )
+    }
+}
+
+impl Serialize for Timestamp {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.to_rfc3339())
+    }
+}
+
+impl<'de> Deserialize<'de> for Timestamp {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let _s = String::deserialize(deserializer)?;
+        // Simple RFC3339 parsing - accepts the format we produce
+        // For a production system, you might want more robust parsing
+        Ok(Timestamp { secs: 0 }) // Simplified - we mainly serialize, not deserialize
+    }
+}
+
+/// Convert days since Unix epoch to year/month/day
+fn days_to_ymd(mut days: i64) -> (i32, u8, u8) {
+    let mut year = 1970;
+    loop {
+        let days_in_year = if is_leap_year(year) { 366 } else { 365 };
+        if days < days_in_year {
+            break;
+        }
+        days -= days_in_year;
+        year += 1;
+    }
+
+    let days_in_months = if is_leap_year(year) {
+        [31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
+    } else {
+        [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
+    };
+
+    let mut month = 1;
+    for &dim in &days_in_months {
+        if days < dim as i64 {
+            break;
+        }
+        days -= dim as i64;
+        month += 1;
+    }
+
+    (year, month, days as u8 + 1)
+}
+
+fn is_leap_year(year: i32) -> bool {
+    (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0)
+}
+
+/// Metric types that can be submitted to the API
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum Metric {
+    #[serde(rename = "sensor_reading")]
+    SensorReading(SensorReading),
+    #[serde(rename = "interface_stat")]
+    InterfaceStat(InterfaceStat),
+}
+
+impl Metric {
+    pub fn metric_type(&self) -> &str {
+        match self {
+            Metric::SensorReading(_) => "sensor_reading",
+            Metric::InterfaceStat(_) => "interface_stat",
+        }
+    }
+
+    pub fn timestamp(&self) -> &Timestamp {
+        match self {
+            Metric::SensorReading(sr) => &sr.timestamp,
+            Metric::InterfaceStat(is) => &is.timestamp,
+        }
+    }
+}
+
+/// Sensor reading metric
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SensorReading {
+    pub sensor_id: String,
+    pub value: f64,
+    pub status: String,
+    pub timestamp: Timestamp,
+}
+
+/// Interface statistics metric
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct InterfaceStat {
+    pub interface_id: String,
+    pub if_in_octets: i64,
+    pub if_out_octets: i64,
+    pub if_in_errors: i64,
+    pub if_out_errors: i64,
+    pub if_in_discards: i64,
+    pub if_out_discards: i64,
+    pub timestamp: Timestamp,
+}

src/ping.rs

@@ -1,259 +0,0 @@
-use std::net::IpAddr;
-use std::process::Stdio;
-use std::time::Duration;
-use tokio::process::Command;
-use tokio::time::timeout;
-
-type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
-
-/// Pings an IP address with multiple attempts for reliability.
-///
-/// This function sends multiple ping packets and tolerates some packet loss,
-/// making it more resilient to temporary network issues. It only fails if
-/// all packets are lost.
-///
-/// # Arguments
-///
-/// * `ip` - The IP address to ping
-/// * `timeout_duration` - Maximum time to wait for responses
-/// * `count` - Number of ping packets to send (default: 3)
-///
-/// # Returns
-///
-/// * `Ok(Duration)` - The average round-trip time if at least one packet succeeds
-/// * `Err(anyhow::Error)` - If all pings fail or 100% packet loss
-pub async fn ping_with_retries(
-    ip: IpAddr,
-    timeout_duration: Duration,
-    count: u32,
-) -> Result<Duration> {
-    let ip_str = ip.to_string();
-    let timeout_secs = timeout_duration.as_secs().max(1);
-    let packet_count = count.max(1).to_string();
-
-    // Build ping command arguments based on OS
-    let args = if cfg!(target_os = "macos") {
-        // macOS: -W is timeout in ms
-        vec![
-            "-c".to_string(),
-            packet_count,
-            "-W".to_string(),
-            (timeout_secs * 1000).to_string(),
-            ip_str.clone(),
-        ]
-    } else {
-        // Linux: -W is timeout in seconds
-        vec![
-            "-c".to_string(),
-            packet_count,
-            "-W".to_string(),
-            timeout_secs.to_string(),
-            ip_str.clone(),
-        ]
-    };
-
-    // Execute ping command with timeout (extra time for multiple packets)
-    let total_timeout = timeout_duration * count + Duration::from_secs(2);
-    let result = timeout(total_timeout, async {
-        let output = Command::new("ping")
-            .args(&args)
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .output()
-            .await?;
-
-        let stdout = String::from_utf8_lossy(&output.stdout);
-
-        // Check for 100% packet loss - this is a hard failure
-        if stdout.contains("100% packet loss") || stdout.contains("100.0% packet loss") {
-            return Err(format!("All {} ping packets lost to {}", count, ip_str).into());
-        }
-
-        // If we got any successful packets, parse the average RTT
-        if output.status.success() || !stdout.is_empty() {
-            parse_ping_output(&stdout)
-        } else {
-            let stderr = String::from_utf8_lossy(&output.stderr);
-            Err(format!(
-                "Ping failed for {}: {}{}",
-                ip_str,
-                stdout.trim(),
-                stderr.trim()
-            )
-            .into())
-        }
-    })
-    .await;
-
-    match result {
-        Ok(inner_result) => inner_result,
-        Err(_) => Err(format!("Ping timeout for {} after {} attempts", ip_str, count).into()),
-    }
-}
-
-/// Parse the ping output to extract round-trip time as Duration.
-///
-/// Supports multiple output formats:
-/// - macOS: "round-trip min/avg/max/stddev = 1.234/1.234/1.234/0.000 ms"
-/// - Linux: "rtt min/avg/max/mdev = 1.234/1.234/1.234/0.000 ms"
-/// - Both: "time=X.XX ms" in the reply line
-fn parse_ping_output(output: &str) -> Result<Duration> {
-    // Try macOS format: "round-trip min/avg/max/stddev = X/Y/Z/W ms"
-    if let Some(caps) = regex_lite::Regex::new(r"round-trip.*=\s*[\d.]+/([\d.]+)/")
-        .ok()
-        .and_then(|re| re.captures(output))
-    {
-        if let Some(avg_ms) = caps.get(1) {
-            if let Ok(ms) = avg_ms.as_str().parse::<f64>() {
-                return Ok(Duration::from_secs_f64(ms / 1000.0));
-            }
-        }
-    }
-
-    // Try Linux format: "rtt min/avg/max/mdev = X/Y/Z/W ms"
-    if let Some(caps) = regex_lite::Regex::new(r"rtt.*=\s*[\d.]+/([\d.]+)/")
-        .ok()
-        .and_then(|re| re.captures(output))
-    {
-        if let Some(avg_ms) = caps.get(1) {
-            if let Ok(ms) = avg_ms.as_str().parse::<f64>() {
-                return Ok(Duration::from_secs_f64(ms / 1000.0));
-            }
-        }
-    }
-
-    // Try extracting from "time=X.XX ms" in the reply line
-    if let Some(caps) = regex_lite::Regex::new(r"time[=<]([\d.]+)\s*ms")
-        .ok()
-        .and_then(|re| re.captures(output))
-    {
-        if let Some(time_ms) = caps.get(1) {
-            if let Ok(ms) = time_ms.as_str().parse::<f64>() {
-                return Ok(Duration::from_secs_f64(ms / 1000.0));
-            }
-        }
-    }
-
-    Err(format!("Could not parse ping output: {}", output.trim()).into())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_parse_ping_output_macos() {
-        let output = r#"PING 192.168.1.1 (192.168.1.1): 56 data bytes
-64 bytes from 192.168.1.1: icmp_seq=0 ttl=64 time=1.234 ms
-
---- 192.168.1.1 ping statistics ---
-1 packets transmitted, 1 packets received, 0.0% packet loss
-round-trip min/avg/max/stddev = 1.234/1.234/1.234/0.000 ms"#;
-
-        let result = parse_ping_output(output);
-        assert!(result.is_ok());
-        let duration = result.unwrap();
-        // Should be approximately 1.234ms
-        assert!(duration.as_secs_f64() > 0.001);
-        assert!(duration.as_secs_f64() < 0.002);
-    }
-
-    #[test]
-    fn test_parse_ping_output_linux() {
-        let output = r#"PING 192.168.1.1 (192.168.1.1) 56(84) bytes of data.
-64 bytes from 192.168.1.1: icmp_seq=1 ttl=64 time=0.543 ms
-
---- 192.168.1.1 ping statistics ---
-1 packets transmitted, 1 received, 0% packet loss, time 0ms
-rtt min/avg/max/mdev = 0.543/0.543/0.543/0.000 ms"#;
-
-        let result = parse_ping_output(output);
-        assert!(result.is_ok());
-        let duration = result.unwrap();
-        // Should be approximately 0.543ms
-        assert!(duration.as_secs_f64() > 0.0005);
-        assert!(duration.as_secs_f64() < 0.001);
-    }
-
-    #[test]
-    fn test_parse_ping_output_time_only() {
-        // Some systems may not include the summary line
-        let output = "64 bytes from 192.168.1.1: icmp_seq=0 ttl=64 time=2.567 ms";
-
-        let result = parse_ping_output(output);
-        assert!(result.is_ok());
-        let duration = result.unwrap();
-        // Should be approximately 2.567ms
-        assert!(duration.as_secs_f64() > 0.002);
-        assert!(duration.as_secs_f64() < 0.003);
-    }
-
-    #[test]
-    fn test_parse_ping_output_time_less_than() {
-        // Some ping implementations use time<1 ms for very fast responses
-        let output = "64 bytes from 127.0.0.1: icmp_seq=0 ttl=64 time<1 ms";
-
-        let result = parse_ping_output(output);
-        assert!(result.is_ok());
-        let duration = result.unwrap();
-        // Should be approximately 1ms (the < becomes the value)
-        assert!(duration.as_secs_f64() < 0.002);
-    }
-
-    #[test]
-    fn test_parse_ping_output_invalid() {
-        let output = "some random text without ping data";
-        let result = parse_ping_output(output);
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_parse_ping_output_empty() {
-        let result = parse_ping_output("");
-        assert!(result.is_err());
-    }
-
-    #[tokio::test]
-    async fn test_ping_with_retries_localhost() {
-        // Test multi-packet ping to localhost
-        let ip: IpAddr = "127.0.0.1".parse().unwrap();
-        let result = ping_with_retries(ip, Duration::from_secs(5), 3).await;
-
-        // Localhost ping should succeed on most systems
-        if result.is_ok() {
-            let duration = result.unwrap();
-            // Localhost should respond in < 100ms on average
-            assert!(duration.as_millis() < 100);
-        }
-        // If it fails, that's okay too - some systems don't allow ping to localhost
-    }
-
-    #[tokio::test]
-    async fn test_ping_with_retries_invalid_ip() {
-        // This should fail - non-routable IP
-        let ip: IpAddr = "192.0.2.1".parse().unwrap(); // TEST-NET-1, not routable
-        let result = ping_with_retries(ip, Duration::from_secs(2), 3).await;
-
-        // Should fail with 100% packet loss
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_parse_ping_output_with_packet_loss() {
-        // Test output with some packet loss but not 100%
-        let output = r#"PING 192.168.1.1 (192.168.1.1): 56 data bytes
-64 bytes from 192.168.1.1: icmp_seq=0 ttl=64 time=1.234 ms
-Request timeout for icmp_seq 1
-
---- 192.168.1.1 ping statistics ---
-3 packets transmitted, 1 packets received, 66.7% packet loss
-round-trip min/avg/max/stddev = 1.234/1.234/1.234/0.000 ms"#;
-
-        let result = parse_ping_output(output);
-        assert!(result.is_ok());
-        let duration = result.unwrap();
-        // Should still parse the average from the one successful packet
-        assert!(duration.as_secs_f64() > 0.001);
-        assert!(duration.as_secs_f64() < 0.002);
-    }
-}

src/poller/executor.rs

@@ -0,0 +1,232 @@
+use crate::buffer::Storage;
+use crate::buffer::StorageError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ExecutorError {
+    #[error("Storage error: {0}")]
+    Storage(#[from] StorageError),
+    #[error("SNMP error: {0}")]
+    Snmp(#[from] crate::snmp::SnmpError),
+    #[error("Conversion error: {0}")]
+    Conversion(String),
+}
+
+pub type Result<T> = std::result::Result<T, ExecutorError>;
+
+use crate::config::EquipmentConfig;
+use crate::metrics::{InterfaceStat, Metric, SensorReading};
+use crate::snmp::SnmpClient;
+
+use crate::metrics::Timestamp;
+use log::{error, info, warn};
+
+/// Executor handles polling individual pieces of equipment
+pub struct Executor {
+    snmp_client: SnmpClient,
+    storage: Storage,
+}
+
+impl Executor {
+    pub fn new(snmp_client: SnmpClient, storage: Storage) -> Self {
+        Self {
+            snmp_client,
+            storage,
+        }
+    }
+
+    /// Poll sensors for a piece of equipment
+    pub async fn poll_sensors(&self, equipment: &EquipmentConfig) -> Result<()> {
+        if !equipment.snmp.enabled || equipment.sensors.is_empty() {
+            return Ok(());
+        }
+
+        info!(
+            "Polling {} sensors for equipment: {}",
+            equipment.sensors.len(),
+            equipment.name
+        );
+
+        for sensor in &equipment.sensors {
+            match self
+                .snmp_client
+                .get(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &sensor.oid,
+                )
+                .await
+            {
+                Ok(value) => {
+                    let raw_value = match value.as_f64() {
+                        Some(v) => v,
+                        None => {
+                            warn!(
+                                "Could not convert sensor value to number for sensor {}",
+                                sensor.id
+                            );
+                            continue;
+                        }
+                    };
+
+                    // Apply divisor if specified
+                    let final_value = if let Some(divisor) = sensor.divisor {
+                        if divisor != 0 {
+                            raw_value / divisor as f64
+                        } else {
+                            raw_value
+                        }
+                    } else {
+                        raw_value
+                    };
+
+                    let reading = SensorReading {
+                        sensor_id: sensor.id.clone(),
+                        value: final_value,
+                        status: "ok".to_string(),
+                        timestamp: Timestamp::now(),
+                    };
+
+                    if let Err(e) = self.storage.store_metric(&Metric::SensorReading(reading)) {
+                        error!("Failed to store sensor reading: {}", e);
+                    }
+                }
+                Err(e) => {
+                    warn!(
+                        "Failed to poll sensor {} for {}: {}",
+                        sensor.oid, equipment.name, e
+                    );
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Poll interfaces for a piece of equipment
+    pub async fn poll_interfaces(&self, equipment: &EquipmentConfig) -> Result<()> {
+        if !equipment.snmp.enabled || equipment.interfaces.is_empty() {
+            return Ok(());
+        }
+
+        info!(
+            "Polling {} interfaces for equipment: {}",
+            equipment.interfaces.len(),
+            equipment.name
+        );
+
+        for interface in &equipment.interfaces {
+            // OIDs for interface statistics (from IF-MIB)
+            let if_in_octets_oid = format!("1.3.6.1.2.1.2.2.1.10.{}", interface.if_index);
+            let if_out_octets_oid = format!("1.3.6.1.2.1.2.2.1.16.{}", interface.if_index);
+            let if_in_errors_oid = format!("1.3.6.1.2.1.2.2.1.14.{}", interface.if_index);
+            let if_out_errors_oid = format!("1.3.6.1.2.1.2.2.1.20.{}", interface.if_index);
+            let if_in_discards_oid = format!("1.3.6.1.2.1.2.2.1.13.{}", interface.if_index);
+            let if_out_discards_oid = format!("1.3.6.1.2.1.2.2.1.19.{}", interface.if_index);
+
+            // Poll each counter
+            let in_octets = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_in_octets_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let out_octets = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_out_octets_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let in_errors = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_in_errors_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let out_errors = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_out_errors_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let in_discards = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_in_discards_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let out_discards = self
+                .get_counter(
+                    &equipment.ip_address,
+                    &equipment.snmp.community,
+                    &equipment.snmp.version,
+                    equipment.snmp.port,
+                    &if_out_discards_oid,
+                )
+                .await
+                .unwrap_or(0);
+
+            let stat = InterfaceStat {
+                interface_id: interface.id.clone(),
+                if_in_octets: in_octets,
+                if_out_octets: out_octets,
+                if_in_errors: in_errors,
+                if_out_errors: out_errors,
+                if_in_discards: in_discards,
+                if_out_discards: out_discards,
+                timestamp: Timestamp::now(),
+            };
+
+            if let Err(e) = self.storage.store_metric(&Metric::InterfaceStat(stat)) {
+                error!("Failed to store interface stat: {}", e);
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn get_counter(
+        &self,
+        ip_address: &str,
+        community: &str,
+        version: &str,
+        port: u16,
+        oid: &str,
+    ) -> Result<i64> {
+        let value = self
+            .snmp_client
+            .get(ip_address, community, version, port, oid)
+            .await?;
+
+        value
+            .as_i64()
+            .ok_or_else(|| ExecutorError::Conversion("Could not convert value to i64".into()))
+    }
+}

src/poller/mod.rs

@@ -0,0 +1,5 @@
+mod executor;
+mod scheduler;
+
+pub use executor::Executor;
+pub use scheduler::Scheduler;

src/poller/scheduler.rs

@@ -0,0 +1,221 @@
+use crate::api_client::ApiClient;
+use crate::buffer::StorageError;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum SchedulerError {
+    #[error("API error: {0}")]
+    Api(#[from] crate::api_client::ApiError),
+    #[error("Storage error: {0}")]
+    Storage(#[from] StorageError),
+    #[error("SNMP error: {0}")]
+    Snmp(#[from] crate::snmp::SnmpError),
+    #[error("Executor error: {0}")]
+    Executor(#[from] super::executor::ExecutorError),
+}
+
+pub type Result<T> = std::result::Result<T, SchedulerError>;
+
+use crate::buffer::Storage;
+use crate::config::{AgentConfig, HeartbeatMetadata};
+use crate::poller::Executor;
+use crate::snmp::SnmpClient;
+
+use crate::metrics::Timestamp;
+use log::{error, info, warn};
+use std::time::Duration;
+use tokio::time::interval;
+
+/// Main scheduler that orchestrates polling, config refresh, and metrics submission
+pub struct Scheduler {
+    api_client: ApiClient,
+    storage: Storage,
+    executor: Executor,
+    config_refresh_seconds: u64,
+    current_config: Option<AgentConfig>,
+    start_time: Timestamp,
+}
+
+impl Scheduler {
+    pub fn new(
+        api_client: ApiClient,
+        storage: Storage,
+        snmp_client: SnmpClient,
+        config_refresh_seconds: u64,
+    ) -> Self {
+        let executor = Executor::new(snmp_client, storage.clone());
+
+        Self {
+            api_client,
+            storage,
+            executor,
+            config_refresh_seconds,
+            current_config: None,
+            start_time: Timestamp::now(),
+        }
+    }
+
+    /// Run the main event loop
+    pub async fn run(&mut self) -> Result<()> {
+        info!("Starting Towerops agent scheduler");
+
+        // Fetch initial configuration
+        if let Err(e) = self.refresh_config().await {
+            error!("Failed to fetch initial config: {}", e);
+        }
+
+        let mut config_ticker = interval(Duration::from_secs(self.config_refresh_seconds));
+        let mut metrics_ticker = interval(Duration::from_secs(30));
+        let mut heartbeat_ticker = interval(Duration::from_secs(60));
+        let mut cleanup_ticker = interval(Duration::from_secs(3600)); // Cleanup every hour
+        let mut poll_ticker = interval(Duration::from_secs(5)); // Check if polling needed every 5s
+
+        loop {
+            tokio::select! {
+                _ = config_ticker.tick() => {
+                    if let Err(e) = self.refresh_config().await {
+                        error!("Failed to refresh config: {}", e);
+                    }
+                }
+
+                _ = metrics_ticker.tick() => {
+                    if let Err(e) = self.flush_metrics().await {
+                        error!("Failed to flush metrics: {}", e);
+                    }
+                }
+
+                _ = heartbeat_ticker.tick() => {
+                    if let Err(e) = self.send_heartbeat().await {
+                        warn!("Failed to send heartbeat: {}", e);
+                    }
+                }
+
+                _ = cleanup_ticker.tick() => {
+                    if let Err(e) = self.storage.cleanup_old_metrics() {
+                        error!("Failed to cleanup old metrics: {}", e);
+                    }
+                }
+
+                _ = poll_ticker.tick() => {
+                    if let Err(e) = self.poll_equipment().await {
+                        error!("Polling error: {}", e);
+                    }
+                }
+            }
+        }
+    }
+
+    async fn refresh_config(&mut self) -> Result<()> {
+        info!("Refreshing configuration from API");
+
+        match self.api_client.fetch_config().await {
+            Ok(config) => {
+                info!(
+                    "Configuration updated: {} equipment items",
+                    config.equipment.len()
+                );
+                self.current_config = Some(config);
+                Ok(())
+            }
+            Err(e) => {
+                warn!(
+                    "Failed to fetch config, continuing with cached config: {}",
+                    e
+                );
+                Err(e.into())
+            }
+        }
+    }
+
+    async fn flush_metrics(&self) -> Result<()> {
+        let pending = self.storage.get_pending_metrics(100)?;
+
+        if pending.is_empty() {
+            return Ok(());
+        }
+
+        info!("Flushing {} pending metrics to API", pending.len());
+
+        let ids: Vec<i64> = pending.iter().map(|(id, _)| *id).collect();
+        let metrics: Vec<_> = pending.into_iter().map(|(_, m)| m).collect();
+
+        match self.api_client.submit_metrics(metrics).await {
+            Ok(_) => {
+                self.storage.mark_metrics_sent(&ids)?;
+                info!("Successfully submitted {} metrics", ids.len());
+                Ok(())
+            }
+            Err(e) => {
+                warn!("Failed to submit metrics, will retry later: {}", e);
+                Err(e.into())
+            }
+        }
+    }
+
+    async fn send_heartbeat(&self) -> Result<()> {
+        let uptime = self.start_time.elapsed_secs() as u64;
+
+        let hostname = hostname::get()
+            .ok()
+            .and_then(|h| h.into_string().ok())
+            .unwrap_or_else(|| "unknown".to_string());
+
+        let metadata = HeartbeatMetadata {
+            version: env!("CARGO_PKG_VERSION").to_string(),
+            hostname,
+            uptime_seconds: uptime,
+        };
+
+        self.api_client.heartbeat(metadata).await?;
+        Ok(())
+    }
+
+    async fn poll_equipment(&self) -> Result<()> {
+        let config = match &self.current_config {
+            Some(c) => c,
+            None => return Ok(()),
+        };
+
+        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
+                }
+                None => true, // Never polled before
+            };
+
+            if should_poll {
+                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)
+                );
+
+                if let Err(e) = sensor_result {
+                    error!("Failed to poll sensors for {}: {}", equipment.name, e);
+                }
+
+                if let Err(e) = interface_result {
+                    error!("Failed to poll interfaces for {}: {}", equipment.name, e);
+                }
+
+                // Update last poll time
+                if let Err(e) = self.storage.update_last_poll_time(&equipment.id) {
+                    error!("Failed to update last poll time: {}", e);
+                }
+            }
+        }
+
+        Ok(())
+    }
+}

src/websocket_client.rs

@@ -21,10 +21,7 @@ const CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
-use crate::ping::ping_with_retries;
+use crate::proto::agent::{AgentHeartbeat, AgentJob, AgentJobList, JobType, QueryType, SnmpResult};
-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).
@@ -43,10 +40,6 @@ pub struct AgentClient {
     agent_id: String,
     result_tx: mpsc::UnboundedSender<SnmpResult>,
     result_rx: mpsc::UnboundedReceiver<SnmpResult>,
-    monitoring_check_tx: mpsc::UnboundedSender<MonitoringCheck>,
-    monitoring_check_rx: mpsc::UnboundedReceiver<MonitoringCheck>,
-    task_shutdown_tx: watch::Sender<bool>,
-    task_shutdown_rx: watch::Receiver<bool>,
 }
 
 impl AgentClient {
@@ -94,8 +87,6 @@ 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();
-        let (task_shutdown_tx, task_shutdown_rx) = watch::channel(false);
 
         // Join Phoenix channel with token in payload
         let join_msg = PhoenixMessage {
@@ -117,10 +108,6 @@ impl AgentClient {
             agent_id,
             result_tx,
             result_rx,
-            monitoring_check_tx,
-            monitoring_check_rx,
-            task_shutdown_tx,
-            task_shutdown_rx,
         })
     }
 
@@ -185,13 +172,6 @@ impl AgentClient {
                     }
                 }
 
-                // Receive monitoring checks from ping tasks
-                Some(check) = self.monitoring_check_rx.recv() => {
-                    if let Err(e) = self.send_monitoring_check(check).await {
-                        crate::log_error!("Error sending monitoring check: {}", e);
-                    }
-                }
-
                 // Send periodic heartbeats
                 _ = heartbeat_interval.tick() => {
                     if let Err(e) = self.send_heartbeat().await {
@@ -201,13 +181,6 @@ impl AgentClient {
             }
         };
 
-        // Signal all spawned tasks to shut down
-        crate::log_info!("Shutting down all spawned tasks");
-        let _ = self.task_shutdown_tx.send(true);
-
-        // Give tasks a moment to shut down gracefully
-        tokio::time::sleep(Duration::from_secs(2)).await;
-
         result
     }
 
@@ -248,6 +221,10 @@ impl AgentClient {
     }
 
     /// Process job list from server.
+    ///
+    /// Each job is executed once in the background and results are sent back.
+    /// No long-running tasks are spawned - the agent is stateless.
+    /// Server handles all scheduling and retries via Oban.
     async fn handle_jobs(&self, job_list: AgentJobList) -> Result<()> {
         crate::log_info!("Received {} jobs from server", job_list.jobs.len());
 
@@ -255,57 +232,14 @@ impl AgentClient {
             let job_type = JobType::try_from(job.job_type).unwrap_or(JobType::Poll);
             crate::log_info!("Executing job: {} (type: {:?})", job.job_id, job_type);
 
-            // Extract monitoring configuration before moving job
+            // Spawn task to execute SNMP job (one-off execution)
-            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();
-            let shutdown_rx = self.task_shutdown_rx.clone();
 
             tokio::spawn(async move {
                 if let Err(e) = execute_job(job, result_tx).await {
                     crate::log_error!("Job execution failed: {}", e);
                 }
             });
-
-            // Start monitoring task if enabled
-            if let Some((ip, interval_seconds)) = monitoring_info {
-                crate::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,
-                        shutdown_rx,
-                    )
-                    .await
-                    {
-                        crate::log_error!("Monitoring task failed: {}", e);
-                    }
-                });
-            }
         }
 
         Ok(())
@@ -354,28 +288,6 @@ impl AgentClient {
         crate::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?;
-
-        crate::log_debug!(
-            "Sent monitoring check for device {}: {}",
-            check.device_id,
-            check.status
-        );
-        Ok(())
-    }
 }
 
 /// Execute an SNMP job and collect results.
@@ -629,95 +541,6 @@ fn get_local_ip() -> Option<String> {
     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<MonitoringCheck>,
-    mut shutdown_rx: watch::Receiver<bool>,
-) -> Result<()> {
-    let mut interval_timer = interval(Duration::from_secs(interval_seconds));
-
-    crate::log_info!("Monitoring task started for device {} at {}", device_id, ip);
-
-    loop {
-        tokio::select! {
-            // Check for shutdown signal
-            _ = shutdown_rx.changed() => {
-                if *shutdown_rx.borrow() {
-                    crate::log_info!(
-                        "Monitoring task for device {} shutting down gracefully",
-                        device_id
-                    );
-                    return Ok(());
-                }
-            }
-
-            // Wait for next interval
-            _ = interval_timer.tick() => {
-                let ip_addr: std::net::IpAddr = match ip.parse() {
-                    Ok(addr) => addr,
-                    Err(e) => {
-                        crate::log_error!("Invalid IP address for device {}: {}", device_id, e);
-                        continue;
-                    }
-                };
-
-                let timeout = Duration::from_secs(5);
-
-                // Send 3 pings for reliability - only fail if all 3 fail
-                match ping_with_retries(ip_addr, timeout, 3).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 check_tx.send(check).is_err() {
-                            crate::log_warn!(
-                                "Monitoring task for device {}: channel closed, stopping task (connection may have dropped)",
-                                device_id
-                            );
-                            return Ok(());
-                        }
-
-                        crate::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 check_tx.send(check).is_err() {
-                            crate::log_warn!(
-                                "Monitoring task for device {}: channel closed, stopping task (connection may have dropped)",
-                                device_id
-                            );
-                            return Ok(());
-                        }
-
-                        crate::log_warn!("ICMP ping failed for {}: {}", ip, e);
-                    }
-                }
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;