diff --git a/docker-compose.example.yml b/docker-compose.example.yml new file mode 100644 index 0000000..eb8c505 --- /dev/null +++ b/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 diff --git a/proto/agent.proto b/proto/agent.proto index 428f861..e7ea17d 100644 --- a/proto/agent.proto +++ b/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 { diff --git a/src/api_client.rs b/src/api_client.rs new file mode 100644 index 0000000..5170749 --- /dev/null +++ b/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 = std::result::Result; + +/// 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 { + Ok(Self { base_url, token }) + } + + /// Fetch configuration from the API + pub async fn fetch_config(&self) -> Result { + 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) -> 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 = 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(), + }, + )), + }, + } +} diff --git a/src/buffer/mod.rs b/src/buffer/mod.rs new file mode 100644 index 0000000..7968e16 --- /dev/null +++ b/src/buffer/mod.rs @@ -0,0 +1,3 @@ +mod storage; + +pub use storage::{Storage, StorageError}; diff --git a/src/buffer/storage.rs b/src/buffer/storage.rs new file mode 100644 index 0000000..6ac52f7 --- /dev/null +++ b/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 = std::result::Result; + +/// SQLite storage for buffering metrics when API is unavailable +#[derive(Clone)] +pub struct Storage { + conn: Arc>, +} + +impl Storage { + /// Create a new storage instance + pub fn new>(path: P) -> Result { + 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> { + 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::(&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::>().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> { + let conn = self.conn.lock().unwrap(); + + let result: std::result::Result = 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> { + 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 = 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) + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..a46c8ba --- /dev/null +++ b/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, +} + +/// 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, + pub interfaces: Vec, +} + +/// 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, + pub unit: Option, + pub metadata: Option, +} + +/// 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, +} diff --git a/src/main.rs b/src/main.rs index 318717f..74c4364 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,3 @@ -mod ping; mod proto; mod snmp; mod version; diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs new file mode 100644 index 0000000..8db631f --- /dev/null +++ b/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(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_rfc3339()) + } +} + +impl<'de> Deserialize<'de> for Timestamp { + fn deserialize(deserializer: D) -> Result + 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, +} diff --git a/src/ping.rs b/src/ping.rs deleted file mode 100644 index c3e92be..0000000 --- a/src/ping.rs +++ /dev/null @@ -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 = std::result::Result>; - -/// 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 { - 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 { - // 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::() { - 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::() { - 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::() { - 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); - } -} diff --git a/src/poller/executor.rs b/src/poller/executor.rs new file mode 100644 index 0000000..29541ae --- /dev/null +++ b/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 = std::result::Result; + +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 { + 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())) + } +} diff --git a/src/poller/mod.rs b/src/poller/mod.rs new file mode 100644 index 0000000..8c7e04d --- /dev/null +++ b/src/poller/mod.rs @@ -0,0 +1,5 @@ +mod executor; +mod scheduler; + +pub use executor::Executor; +pub use scheduler::Scheduler; diff --git a/src/poller/scheduler.rs b/src/poller/scheduler.rs new file mode 100644 index 0000000..cd69a5b --- /dev/null +++ b/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 = std::result::Result; + +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, + 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 = 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(()) + } +} diff --git a/src/websocket_client.rs b/src/websocket_client.rs index 3de5d39..04c5a1e 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -21,10 +21,7 @@ const CONNECTION_TIMEOUT: Duration = Duration::from_secs(30); type Result = std::result::Result>; -use crate::ping::ping_with_retries; -use crate::proto::agent::{ - AgentHeartbeat, AgentJob, AgentJobList, JobType, MonitoringCheck, QueryType, SnmpResult, -}; +use crate::proto::agent::{AgentHeartbeat, AgentJob, AgentJobList, JobType, 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, result_rx: mpsc::UnboundedReceiver, - monitoring_check_tx: mpsc::UnboundedSender, - monitoring_check_rx: mpsc::UnboundedReceiver, - task_shutdown_tx: watch::Sender, - task_shutdown_rx: watch::Receiver, } 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 - 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 + // Spawn task to execute SNMP job (one-off execution) 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 { 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, - mut shutdown_rx: watch::Receiver, -) -> 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::*;