init

.dockerignore

@@ -0,0 +1,9 @@
+target/
+.git/
+.gitignore
+Dockerfile
+.dockerignore
+README.md
+*.db
+*.db-shm
+*.db-wal

.gitignore

@@ -0,0 +1,25 @@
+# Created by https://gitignore.org
+# Rust.gitignore
+
+# Generated by Cargo
+# will have compiled files and executables
+debug
+target
+
+# These are backup files generated by rustfmt
+**/*.rs.bk
+
+# MSVC Windows builds of rustc generate these, which store debugging information
+*.pdb
+
+# Generated by cargo mutants
+# Contains mutation testing data
+**/mutants.out*/
+
+# RustRover
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+

Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "towerops-agent"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+snmp = "0.2"
+reqwest = { version = "0.12", features = ["json", "rustls-tls"], default-features = false }
+tokio = { version = "1", features = ["full"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+rusqlite = { version = "0.32", features = ["bundled"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+anyhow = "1.0"
+thiserror = "1.0"
+chrono = { version = "0.4", features = ["serde"] }
+clap = { version = "4.0", features = ["derive", "env"] }
+hostname = "0.4"
+
+[profile.release]
+opt-level = "z"
+lto = true
+codegen-units = 1
+strip = true

Dockerfile

@@ -0,0 +1,53 @@
+# Build stage
+FROM rust:1.75-alpine AS builder
+
+WORKDIR /app
+
+# Install build dependencies
+RUN apk add --no-cache musl-dev
+
+# Copy manifests
+COPY Cargo.toml Cargo.lock ./
+
+# Create a dummy main.rs to build dependencies
+RUN mkdir src && echo "fn main() {}" > src/main.rs
+
+# Build dependencies (cached layer)
+RUN cargo build --release --target x86_64-unknown-linux-musl
+
+# Remove dummy src
+RUN rm -rf src
+
+# Copy actual source code
+COPY src ./src
+
+# Build the actual application
+RUN touch src/main.rs && cargo build --release --target x86_64-unknown-linux-musl
+
+# Runtime stage
+FROM alpine:3.19
+
+# Install runtime dependencies
+RUN apk add --no-cache ca-certificates
+
+# Create data directory
+RUN mkdir -p /data
+
+# Copy binary from builder
+COPY --from=builder /app/target/x86_64-unknown-linux-musl/release/towerops-agent /usr/local/bin/
+
+# Volume for database
+VOLUME ["/data"]
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
+  CMD test -f /data/towerops-agent.db || exit 1
+
+# Run as non-root user
+RUN addgroup -g 1000 towerops && \
+    adduser -D -u 1000 -G towerops towerops && \
+    chown -R towerops:towerops /data
+
+USER towerops
+
+ENTRYPOINT ["towerops-agent"]

README.md

@@ -0,0 +1,142 @@
+# Towerops Agent
+
+A lightweight Rust-based remote polling agent for Towerops SNMP monitoring.
+
+## Overview
+
+The Towerops agent enables customers to deploy SNMP polling infrastructure on their internal networks. The agent polls SNMP devices locally and reports metrics to the Towerops API using token-based authentication.
+
+## Features
+
+- **Local SNMP Polling**: Poll devices on your internal network without exposing them to the internet
+- **Secure Communication**: All communication with Towerops API uses HTTPS with token authentication
+- **Metric Buffering**: Stores up to 24 hours of metrics in SQLite when API is unavailable
+- **Automatic Configuration**: Fetches polling configuration from the Towerops API
+- **Low Resource Usage**: Built in Rust for minimal memory and CPU footprint (< 256 MB RAM typical)
+- **Docker Ready**: Pre-built Docker images for easy deployment
+
+## Quick Start
+
+### Using Docker Compose
+
+1. Create a `docker-compose.yml` file:
+
+```yaml
+version: '3.8'
+services:
+  towerops-agent:
+    image: towerops/agent:latest
+    restart: unless-stopped
+    environment:
+      - TOWEROPS_API_URL=https://app.towerops.com
+      - TOWEROPS_AGENT_TOKEN=your-agent-token-here
+    volumes:
+      - ./data:/data
+```
+
+2. Start the agent:
+
+```bash
+docker-compose up -d
+```
+
+### Configuration
+
+The agent is configured via environment variables:
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `TOWEROPS_API_URL` | Towerops API endpoint | Required |
+| `TOWEROPS_AGENT_TOKEN` | Agent authentication token | Required |
+| `CONFIG_REFRESH_SECONDS` | How often to refresh configuration | 300 (5 minutes) |
+| `DATABASE_PATH` | SQLite database path for buffering | `/data/towerops-agent.db` |
+
+### Obtaining an Agent Token
+
+1. Log in to your Towerops account
+2. Navigate to your organization's Agents page
+3. Click "Create New Agent"
+4. Copy the token (it will only be shown once)
+
+## Architecture
+
+The agent consists of several components:
+
+- **API Client**: Communicates with Towerops API to fetch configuration and submit metrics
+- **SNMP Poller**: Polls configured devices for sensor readings and interface statistics
+- **Storage Buffer**: SQLite database that buffers metrics during API outages
+- **Scheduler**: Coordinates polling intervals and metric submission
+
+### Polling Flow
+
+1. Agent fetches configuration from API every 5 minutes (configurable)
+2. For each equipment item, polls sensors and interfaces at configured intervals
+3. Metrics are stored locally in SQLite
+4. Every 30 seconds, pending metrics are submitted to the API
+5. Successfully submitted metrics are marked as sent and cleaned up after 24 hours
+
+### Buffering
+
+If the API is unreachable, metrics are stored locally for up to 24 hours. When connectivity is restored, the agent automatically submits all buffered metrics.
+
+## Building from Source
+
+### Prerequisites
+
+- Rust 1.75 or later
+- SQLite development libraries
+
+### Build
+
+```bash
+cargo build --release
+```
+
+The binary will be in `target/release/towerops-agent`.
+
+### Docker Build
+
+```bash
+docker build -t towerops-agent .
+```
+
+## Troubleshooting
+
+### Agent Not Connecting
+
+- Verify the API URL is correct
+- Check that the agent token is valid and hasn't been revoked
+- Ensure network connectivity to the Towerops API
+
+### No Metrics Appearing
+
+- Check that equipment is assigned to this agent in Towerops
+- Verify SNMP credentials are correct in Towerops equipment settings
+- Review agent logs for SNMP polling errors
+
+### High Memory Usage
+
+- Check the size of the SQLite database (`/data/towerops-agent.db`)
+- Verify metrics are being submitted successfully (check API connectivity)
+
+## Resource Requirements
+
+- **Memory**: 128-256 MB typical, 512 MB maximum
+- **CPU**: 0.1-0.5 cores typical
+- **Disk**: ~100 MB for 24 hours of buffered metrics
+- **Network**: Minimal (metrics are small JSON payloads)
+
+## Security
+
+- Agent token should be kept secret (treat like a password)
+- All API communication uses HTTPS with certificate verification
+- Agent requires no inbound network connections
+- SNMP community strings are only used locally and never logged
+
+## Development Status
+
+**NOTE**: SNMP library integration is incomplete. The current implementation compiles but requires proper SNMP library integration for production use. This will be completed in the next development phase.
+
+## License
+
+Copyright © 2026 Towerops

src/api_client.rs

@@ -0,0 +1,107 @@
+use crate::config::{AgentConfig, HeartbeatMetadata};
+use crate::metrics::Metric;
+use anyhow::{Context, Result};
+use reqwest::Client;
+use serde_json::json;
+use std::time::Duration;
+
+/// API client for communicating with the Towerops server
+#[derive(Clone)]
+pub struct ApiClient {
+    client: Client,
+    base_url: String,
+    token: String,
+}
+
+impl ApiClient {
+    /// Create a new API client
+    pub fn new(base_url: String, token: String) -> Result<Self> {
+        let client = Client::builder()
+            .timeout(Duration::from_secs(30))
+            .build()
+            .context("Failed to create HTTP client")?;
+
+        Ok(Self {
+            client,
+            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 response = self
+            .client
+            .get(&url)
+            .header("Authorization", format!("Bearer {}", self.token))
+            .send()
+            .await
+            .context("Failed to send config request")?;
+
+        if !response.status().is_success() {
+            anyhow::bail!(
+                "Config request failed with status: {}",
+                response.status()
+            );
+        }
+
+        let config: AgentConfig = response
+            .json()
+            .await
+            .context("Failed to parse config response")?;
+
+        Ok(config)
+    }
+
+    /// Submit metrics to the API
+    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 response = self
+            .client
+            .post(&url)
+            .header("Authorization", format!("Bearer {}", self.token))
+            .json(&json!({ "metrics": metrics }))
+            .send()
+            .await
+            .context("Failed to send metrics request")?;
+
+        if !response.status().is_success() {
+            anyhow::bail!(
+                "Metrics submission failed with status: {}",
+                response.status()
+            );
+        }
+
+        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 response = self
+            .client
+            .post(&url)
+            .header("Authorization", format!("Bearer {}", self.token))
+            .json(&metadata)
+            .send()
+            .await
+            .context("Failed to send heartbeat request")?;
+
+        if !response.status().is_success() {
+            anyhow::bail!(
+                "Heartbeat failed with status: {}",
+                response.status()
+            );
+        }
+
+        Ok(())
+    }
+}

src/buffer/mod.rs

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

src/buffer/storage.rs

@@ -0,0 +1,188 @@
+use crate::metrics::Metric;
+use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
+use rusqlite::{params, Connection};
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+
+/// 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).context("Failed to open database")?;
+
+        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
+            )",
+            [],
+        )
+        .context("Failed to create metrics table")?;
+
+        conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_metrics_sent ON metrics(sent, created_at)",
+            [],
+        )
+        .context("Failed to create index")?;
+
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS last_poll_times (
+                equipment_id TEXT PRIMARY KEY,
+                last_poll_time TEXT NOT NULL
+            )",
+            [],
+        )
+        .context("Failed to create last_poll_times table")?;
+
+        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).context("Failed to serialize 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()
+            ],
+        )
+        .context("Failed to insert metric")?;
+
+        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")
+            .context("Failed to prepare statement")?;
+
+        let metrics = stmt
+            .query_map([limit], |row| {
+                let id: i64 = row.get(0)?;
+                let data: String = row.get(1)?;
+                Ok((id, data))
+            })
+            .context("Failed to query metrics")?
+            .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())
+            .context("Failed to mark metrics as sent")?;
+
+        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')",
+            [],
+        )
+        .context("Failed to cleanup old metrics")?;
+
+        Ok(())
+    }
+
+    /// Get the last poll time for equipment
+    pub fn get_last_poll_time(&self, equipment_id: &str) -> Result<Option<DateTime<Utc>>> {
+        let conn = self.conn.lock().unwrap();
+
+        let 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) => {
+                let dt = DateTime::parse_from_rfc3339(&timestamp_str)
+                    .context("Failed to parse timestamp")?
+                    .with_timezone(&Utc);
+                Ok(Some(dt))
+            }
+            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
+            Err(e) => Err(e).context("Failed to query last poll time"),
+        }
+    }
+
+    /// 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 = Utc::now().to_rfc3339();
+
+        conn.execute(
+            "INSERT OR REPLACE INTO last_poll_times (equipment_id, last_poll_time) VALUES (?1, ?2)",
+            params![equipment_id, now],
+        )
+        .context("Failed to update last poll time")?;
+
+        Ok(())
+    }
+
+    /// Get all last poll times
+    pub fn get_all_last_poll_times(&self) -> Result<HashMap<String, DateTime<Utc>>> {
+        let conn = self.conn.lock().unwrap();
+
+        let mut stmt = conn
+            .prepare("SELECT equipment_id, last_poll_time FROM last_poll_times")
+            .context("Failed to prepare statement")?;
+
+        let times: HashMap<String, DateTime<Utc>> = stmt
+            .query_map([], |row| {
+                let equipment_id: String = row.get(0)?;
+                let timestamp_str: String = row.get(1)?;
+                Ok((equipment_id, timestamp_str))
+            })
+            .context("Failed to query poll times")?
+            .filter_map(|r| r.ok())
+            .filter_map(|(id, ts)| {
+                DateTime::parse_from_rfc3339(&ts)
+                    .ok()
+                    .map(|dt| (id, dt.with_timezone(&Utc)))
+            })
+            .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

@@ -0,0 +1,68 @@
+mod api_client;
+mod buffer;
+mod config;
+mod metrics;
+mod poller;
+mod snmp;
+
+use anyhow::Result;
+use clap::Parser;
+use poller::Scheduler;
+use tracing::info;
+
+#[derive(Parser)]
+#[command(name = "towerops-agent")]
+#[command(about = "Towerops remote SNMP polling agent", long_about = None)]
+struct Args {
+    /// API URL (e.g., https://app.towerops.com)
+    #[arg(long, env = "TOWEROPS_API_URL")]
+    api_url: String,
+
+    /// Agent authentication token
+    #[arg(long, env = "TOWEROPS_AGENT_TOKEN")]
+    token: String,
+
+    /// Configuration refresh interval in seconds
+    #[arg(long, env = "CONFIG_REFRESH_SECONDS", default_value = "300")]
+    config_refresh_seconds: u64,
+
+    /// Database path for metrics buffering
+    #[arg(long, env = "DATABASE_PATH", default_value = "/data/towerops-agent.db")]
+    database_path: String,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Initialize logging
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        .init();
+
+    let args = Args::parse();
+
+    info!("Towerops agent starting");
+    info!("API URL: {}", args.api_url);
+    info!(
+        "Config refresh interval: {} seconds",
+        args.config_refresh_seconds
+    );
+    info!("Database path: {}", args.database_path);
+
+    // Initialize components
+    let api_client = api_client::ApiClient::new(args.api_url, args.token)?;
+    let storage = buffer::Storage::new(args.database_path)?;
+    let snmp_client = snmp::SnmpClient::new();
+
+    // Create and run scheduler
+    let mut scheduler = Scheduler::new(
+        api_client,
+        storage,
+        snmp_client,
+        args.config_refresh_seconds,
+    );
+
+    scheduler.run().await
+}

src/metrics/mod.rs

@@ -0,0 +1,50 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+/// 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) -> &DateTime<Utc> {
+        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: DateTime<Utc>,
+}
+
+/// 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: DateTime<Utc>,
+}

src/poller/executor.rs

@@ -0,0 +1,217 @@
+use crate::buffer::Storage;
+use crate::config::EquipmentConfig;
+use crate::metrics::{InterfaceStat, Metric, SensorReading};
+use crate::snmp::SnmpClient;
+use anyhow::Result;
+use chrono::Utc;
+use tracing::{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: Utc::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: Utc::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(|| anyhow::anyhow!("Could not convert value to i64"))
+    }
+}

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,205 @@
+use crate::api_client::ApiClient;
+use crate::buffer::Storage;
+use crate::config::{AgentConfig, HeartbeatMetadata};
+use crate::poller::Executor;
+use crate::snmp::SnmpClient;
+use anyhow::Result;
+use chrono::Utc;
+use std::time::Duration;
+use tokio::time::interval;
+use tracing::{error, info, warn};
+
+/// 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: chrono::DateTime<Utc>,
+}
+
+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: Utc::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)
+            }
+        }
+    }
+
+    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)
+            }
+        }
+    }
+
+    async fn send_heartbeat(&self) -> Result<()> {
+        let uptime = Utc::now()
+            .signed_duration_since(self.start_time)
+            .num_seconds() 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 = Utc::now()
+                        .signed_duration_since(*last_poll)
+                        .num_seconds() 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/snmp/client.rs

@@ -0,0 +1,64 @@
+use super::types::{SnmpError, SnmpResult, SnmpValue};
+use std::net::IpAddr;
+use std::str::FromStr;
+
+/// SNMP client for polling devices
+///
+/// NOTE: This is a simplified implementation. Full SNMP integration
+/// requires proper SNMP library integration with correct error handling.
+#[derive(Debug)]
+pub struct SnmpClient;
+
+impl SnmpClient {
+    pub fn new() -> Self {
+        Self
+    }
+
+    /// Perform an SNMP GET operation
+    ///
+    /// TODO: Complete SNMP library integration with proper session management
+    pub async fn get(
+        &self,
+        ip_address: &str,
+        _community: &str,
+        _version: &str,
+        _port: u16,
+        _oid: &str,
+    ) -> SnmpResult<SnmpValue> {
+        // Validate IP address
+        IpAddr::from_str(ip_address)
+            .map_err(|e| SnmpError::InvalidOid(format!("Invalid IP: {}", e)))?;
+
+        // TODO: Implement actual SNMP GET using snmp library
+        // For now, return an error indicating incomplete implementation
+        Err(SnmpError::RequestFailed(
+            "SNMP implementation incomplete - requires library integration".into(),
+        ))
+    }
+
+    /// Perform an SNMP WALK operation to get multiple values
+    ///
+    /// TODO: Complete SNMP library integration with proper walk implementation
+    pub async fn walk(
+        &self,
+        ip_address: &str,
+        _community: &str,
+        _version: &str,
+        _port: u16,
+        _base_oid: &str,
+    ) -> SnmpResult<Vec<(String, SnmpValue)>> {
+        // Validate IP address
+        IpAddr::from_str(ip_address)
+            .map_err(|e| SnmpError::InvalidOid(format!("Invalid IP: {}", e)))?;
+
+        // TODO: Implement actual SNMP WALK using snmp library
+        // For now, return an empty result
+        Ok(Vec::new())
+    }
+}
+
+impl Default for SnmpClient {
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/snmp/mod.rs

@@ -0,0 +1,5 @@
+mod client;
+mod types;
+
+pub use client::SnmpClient;
+pub use types::*;

src/snmp/types.rs

@@ -0,0 +1,50 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum SnmpError {
+    #[error("SNMP request failed: {0}")]
+    RequestFailed(String),
+
+    #[error("Invalid OID: {0}")]
+    InvalidOid(String),
+
+    #[error("Timeout")]
+    Timeout,
+
+    #[error("Authentication failure")]
+    AuthFailure,
+
+    #[error("Network unreachable")]
+    NetworkUnreachable,
+}
+
+pub type SnmpResult<T> = Result<T, SnmpError>;
+
+/// SNMP value returned from a GET operation
+#[derive(Debug, Clone)]
+pub enum SnmpValue {
+    Integer(i64),
+    String(String),
+    Counter32(u32),
+    Counter64(u64),
+    Gauge32(u32),
+    TimeTicks(u32),
+    IpAddress(String),
+}
+
+impl SnmpValue {
+    pub fn as_i64(&self) -> Option<i64> {
+        match self {
+            SnmpValue::Integer(v) => Some(*v),
+            SnmpValue::Counter32(v) => Some(*v as i64),
+            SnmpValue::Counter64(v) => Some(*v as i64),
+            SnmpValue::Gauge32(v) => Some(*v as i64),
+            SnmpValue::TimeTicks(v) => Some(*v as i64),
+            _ => None,
+        }
+    }
+
+    pub fn as_f64(&self) -> Option<f64> {
+        self.as_i64().map(|v| v as f64)
+    }
+}