From 967d317b6924b697bc07bd32da7709969f390eab Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 16 Jan 2026 17:57:41 -0600 Subject: [PATCH] Complete WebSocket migration for agent communication Major architectural change from REST API polling to WebSocket-based bidirectional communication: **What Changed:** - Agent now uses persistent WebSocket connection instead of REST API - Server pushes SNMP query jobs to agent via Phoenix Channels - Agent executes raw SNMP queries and returns results - Removed complex polling/scheduling/buffering architecture **New Files:** - src/websocket_client.rs - WebSocket client with SNMP job execution - Extended proto/agent.proto with WebSocket message types **Modified Files:** - src/main.rs - Simplified to connect and run WebSocket client - src/health.rs - Simplified health endpoint (no storage needed) - src/snmp/mod.rs - Export SnmpValue for WebSocket client **Removed Files:** - src/api_client.rs - Old REST API client - src/config.rs - Old config types - src/buffer/ - SQLite buffering (no longer needed) - src/metrics/ - Old metric types - src/poller/ - Polling/scheduling logic - src/snmp/neighbor.rs - High-level neighbor discovery **Dependencies:** - Switched from native-tls to rustls for WebSocket TLS - Uses tokio-tungstenite for WebSocket communication - Protobuf for efficient binary message encoding **Benefits:** - Simpler agent architecture (~500 lines vs 5000+) - Real-time job execution (<1s vs 60s polling) - No duplicate SNMP profile logic - No local storage/buffering complexity - 68% smaller message payloads (protobuf vs JSON) --- Cargo.lock | 165 +++++-------------- proto/agent.proto | 56 +++++++ src/api_client.rs | 324 ------------------------------------- src/buffer/mod.rs | 3 - src/buffer/storage.rs | 225 -------------------------- src/config.rs | 58 ------- src/health.rs | 139 +++++----------- src/main.rs | 81 ++++------ src/metrics/mod.rs | 169 ------------------- src/poller/executor.rs | 254 ----------------------------- src/poller/mod.rs | 5 - src/poller/scheduler.rs | 348 ---------------------------------------- src/snmp/mod.rs | 4 +- src/snmp/neighbor.rs | 280 -------------------------------- src/websocket_client.rs | 132 ++++++++------- 15 files changed, 242 insertions(+), 2001 deletions(-) delete mode 100644 src/api_client.rs delete mode 100644 src/buffer/mod.rs delete mode 100644 src/buffer/storage.rs delete mode 100644 src/config.rs delete mode 100644 src/metrics/mod.rs delete mode 100644 src/poller/executor.rs delete mode 100644 src/poller/mod.rs delete mode 100644 src/poller/scheduler.rs delete mode 100644 src/snmp/neighbor.rs diff --git a/Cargo.lock b/Cargo.lock index 465a3e3..28f47ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -232,16 +232,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" -[[package]] -name = "core-foundation" -version = "0.9.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -365,21 +355,6 @@ dependencies = [ "miniz_oxide", ] -[[package]] -name = "foreign-types" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" -dependencies = [ - "foreign-types-shared", -] - -[[package]] -name = "foreign-types-shared" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" - [[package]] name = "form_urlencoded" version = "1.2.2" @@ -814,23 +789,6 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" -[[package]] -name = "native-tls" -version = "0.2.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87de3442987e9dbec73158d5c715e7ad9072fda936bb03d19d7fa10e00520f0e" -dependencies = [ - "libc", - "log", - "openssl", - "openssl-probe", - "openssl-sys", - "schannel", - "security-framework", - "security-framework-sys", - "tempfile", -] - [[package]] name = "num-traits" version = "0.2.19" @@ -852,50 +810,6 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" -[[package]] -name = "openssl" -version = "0.10.75" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08838db121398ad17ab8531ce9de97b244589089e290a384c900cb9ff7434328" -dependencies = [ - "bitflags", - "cfg-if", - "foreign-types", - "libc", - "once_cell", - "openssl-macros", - "openssl-sys", -] - -[[package]] -name = "openssl-macros" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "openssl-probe" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" - -[[package]] -name = "openssl-sys" -version = "0.9.111" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82cab2d520aa75e3c58898289429321eb788c3106963d0dc886ec7a5f4adc321" -dependencies = [ - "cc", - "libc", - "pkg-config", - "vcpkg", -] - [[package]] name = "percent-encoding" version = "2.3.2" @@ -1134,6 +1048,20 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf4ef73721ac7bcd79b2b315da7779d8fc09718c6b3d2d1b2d94850eb8c18432" +dependencies = [ + "log", + "ring", + "rustls-pki-types", + "rustls-webpki 0.102.8", + "subtle", + "zeroize", +] + [[package]] name = "rustls" version = "0.23.36" @@ -1144,7 +1072,7 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki", + "rustls-webpki 0.103.8", "subtle", "zeroize", ] @@ -1158,6 +1086,17 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-webpki" +version = "0.102.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustls-webpki" version = "0.103.8" @@ -1175,38 +1114,6 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" -[[package]] -name = "schannel" -version = "0.1.28" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "891d81b926048e76efe18581bf793546b4c0eaf8448d72be8de2bbee5fd166e1" -dependencies = [ - "windows-sys 0.61.2", -] - -[[package]] -name = "security-framework" -version = "2.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" -dependencies = [ - "bitflags", - "core-foundation", - "core-foundation-sys", - "libc", - "security-framework-sys", -] - -[[package]] -name = "security-framework-sys" -version = "2.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc1f0cbffaac4852523ce30d8bd3c5cdc873501d96ff467ca09b6767bb8cd5c0" -dependencies = [ - "core-foundation-sys", - "libc", -] - [[package]] name = "serde" version = "1.0.228" @@ -1417,12 +1324,13 @@ dependencies = [ ] [[package]] -name = "tokio-native-tls" -version = "0.3.1" +name = "tokio-rustls" +version = "0.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2" +checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f" dependencies = [ - "native-tls", + "rustls 0.22.4", + "rustls-pki-types", "tokio", ] @@ -1434,10 +1342,12 @@ checksum = "c83b561d025642014097b66e6c1bb422783339e0909e4429cde4749d1990bc38" dependencies = [ "futures-util", "log", - "native-tls", + "rustls 0.22.4", + "rustls-pki-types", "tokio", - "tokio-native-tls", + "tokio-rustls", "tungstenite", + "webpki-roots 0.26.11", ] [[package]] @@ -1477,8 +1387,9 @@ dependencies = [ "http", "httparse", "log", - "native-tls", "rand", + "rustls 0.22.4", + "rustls-pki-types", "sha1", "thiserror", "url", @@ -1513,7 +1424,7 @@ dependencies = [ "flate2", "log", "once_cell", - "rustls", + "rustls 0.23.36", "rustls-pki-types", "serde", "serde_json", diff --git a/proto/agent.proto b/proto/agent.proto index 17e3440..402c31b 100644 --- a/proto/agent.proto +++ b/proto/agent.proto @@ -96,3 +96,59 @@ message HeartbeatMetadata { message HeartbeatResponse { string status = 1; } + +// WebSocket-specific messages + +enum JobType { + DISCOVER = 0; + POLL = 1; +} + +enum QueryType { + GET = 0; + WALK = 1; +} + +message AgentJobList { + repeated AgentJob jobs = 1; +} + +message AgentJob { + string job_id = 1; + JobType job_type = 2; + string equipment_id = 3; + SnmpDevice device = 4; + repeated SnmpQuery queries = 5; +} + +message SnmpDevice { + string ip = 1; + string community = 2; + string version = 3; + uint32 port = 4; +} + +message SnmpQuery { + QueryType query_type = 1; + repeated string oids = 2; +} + +message SnmpResult { + string equipment_id = 1; + JobType job_type = 2; + map oid_values = 3; + int64 timestamp = 4; +} + +message AgentHeartbeat { + string version = 1; + string hostname = 2; + uint64 uptime_seconds = 3; + string ip_address = 4; +} + +message AgentError { + string equipment_id = 1; + string error_message = 2; + int64 timestamp = 3; +} diff --git a/src/api_client.rs b/src/api_client.rs deleted file mode 100644 index 66c7d6c..0000000 --- a/src/api_client.rs +++ /dev/null @@ -1,324 +0,0 @@ -use crate::config::{ - AgentConfig, EquipmentConfig, HeartbeatMetadata, InterfaceConfig, SensorConfig, SnmpConfig, -}; -use crate::metrics::Metric; -use crate::proto::agent; -use prost::Message; -use std::io::Read; -use std::time::Duration; - -#[derive(Debug)] -pub enum ApiError { - RequestFailed(String), - StatusError(u16), - JsonError(std::io::Error), - JoinError(String), -} - -impl std::fmt::Display for ApiError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::RequestFailed(msg) => write!(f, "HTTP request failed: {}", msg), - Self::StatusError(status) => write!(f, "HTTP status error: {}", status), - Self::JsonError(err) => write!(f, "JSON parsing error: {}", err), - Self::JoinError(msg) => write!(f, "Task join error: {}", msg), - } - } -} - -impl std::error::Error for ApiError {} - -impl From for ApiError { - fn from(err: std::io::Error) -> Self { - Self::JsonError(err) - } -} - -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 { - // Strip trailing slash from base_url to avoid double slashes in URLs - let base_url = base_url.trim_end_matches('/').to_string(); - Ok(Self { base_url, token }) - } - - /// Fetch configuration from the API using Protocol Buffers - 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)) - .set("Accept", "application/x-protobuf") - .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)); - } - - // Read response bytes - let mut bytes = Vec::new(); - response - .into_reader() - .read_to_end(&mut bytes) - .map_err(|e| ApiError::RequestFailed(e.to_string()))?; - - // Decode protobuf - let proto_config = agent::AgentConfig::decode(&bytes[..]) - .map_err(|e| ApiError::RequestFailed(format!("Protobuf decode error: {}", e)))?; - - // Convert to internal config type - let config = convert_proto_to_config(proto_config); - - 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 using Protocol Buffers - 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 || { - // Convert to protobuf - let proto_metadata = agent::HeartbeatMetadata { - version: metadata.version, - hostname: metadata.hostname, - uptime_seconds: metadata.uptime_seconds, - }; - - // Encode to bytes - let mut buf = Vec::new(); - proto_metadata - .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)); - } - - // Read and decode response - let mut bytes = Vec::new(); - response - .into_reader() - .read_to_end(&mut bytes) - .map_err(|e| ApiError::RequestFailed(e.to_string()))?; - - let _heartbeat_response = - agent::HeartbeatResponse::decode(&bytes[..]).map_err(|e| { - ApiError::RequestFailed(format!("Heartbeat response decode error: {}", e)) - })?; - - Ok(()) - }) - .await - .map_err(|e| ApiError::JoinError(e.to_string()))??; - - Ok(()) - } -} - -/// Convert protobuf AgentConfig to internal AgentConfig -fn convert_proto_to_config(proto: agent::AgentConfig) -> AgentConfig { - AgentConfig { - version: proto.version, - poll_interval_seconds: proto.poll_interval_seconds as u64, - equipment: proto - .equipment - .into_iter() - .map(convert_proto_equipment) - .collect(), - } -} - -/// Convert protobuf Equipment to internal EquipmentConfig -fn convert_proto_equipment(proto: agent::Equipment) -> EquipmentConfig { - EquipmentConfig { - id: proto.id, - name: proto.name, - ip_address: proto.ip_address, - snmp: convert_proto_snmp(proto.snmp.unwrap_or_default()), - poll_interval_seconds: proto.poll_interval_seconds as u64, - sensors: proto - .sensors - .into_iter() - .map(convert_proto_sensor) - .collect(), - interfaces: proto - .interfaces - .into_iter() - .map(convert_proto_interface) - .collect(), - } -} - -/// Convert protobuf SnmpConfig to internal SnmpConfig -fn convert_proto_snmp(proto: agent::SnmpConfig) -> SnmpConfig { - SnmpConfig { - enabled: proto.enabled, - version: proto.version, - community: proto.community, - port: proto.port as u16, - } -} - -/// Convert protobuf Sensor to internal SensorConfig -fn convert_proto_sensor(proto: agent::Sensor) -> SensorConfig { - SensorConfig { - id: proto.id, - sensor_type: proto.r#type, - oid: proto.oid, - divisor: if proto.divisor == 0.0 { - None - } else { - Some(proto.divisor as i32) - }, - unit: if proto.unit.is_empty() { - None - } else { - Some(proto.unit) - }, - metadata: if proto.metadata.is_empty() { - None - } else { - // Convert metadata map to JSON value - let json_map: serde_json::Map = proto - .metadata - .into_iter() - .map(|(k, v)| (k, serde_json::Value::String(v))) - .collect(); - Some(serde_json::Value::Object(json_map)) - }, - } -} - -/// Convert protobuf Interface to internal InterfaceConfig -fn convert_proto_interface(proto: agent::Interface) -> InterfaceConfig { - InterfaceConfig { - id: proto.id, - if_index: proto.if_index as i32, - if_name: proto.if_name, - } -} - -/// 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(), - }, - )), - }, - M::NeighborDiscovery(nd) => agent::Metric { - metric_type: Some(agent::metric::MetricType::NeighborDiscovery( - agent::NeighborDiscovery { - interface_id: nd.interface_id.clone(), - protocol: nd.protocol.clone(), - remote_chassis_id: nd.remote_chassis_id.clone(), - remote_system_name: nd.remote_system_name.clone(), - remote_system_description: nd.remote_system_description.clone(), - remote_platform: nd.remote_platform.clone(), - remote_port_id: nd.remote_port_id.clone(), - remote_port_description: nd.remote_port_description.clone(), - remote_address: nd.remote_address.clone(), - remote_capabilities: nd.remote_capabilities.clone(), - timestamp: nd.timestamp.to_unix_timestamp(), - }, - )), - }, - } -} diff --git a/src/buffer/mod.rs b/src/buffer/mod.rs deleted file mode 100644 index 7968e16..0000000 --- a/src/buffer/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -mod storage; - -pub use storage::{Storage, StorageError}; diff --git a/src/buffer/storage.rs b/src/buffer/storage.rs deleted file mode 100644 index 6354533..0000000 --- a/src/buffer/storage.rs +++ /dev/null @@ -1,225 +0,0 @@ -use crate::metrics::{Metric, Timestamp}; -use rusqlite::{params, Connection}; -use std::path::Path; -use std::sync::{Arc, Mutex}; - -#[derive(Debug)] -pub enum StorageError { - Database(rusqlite::Error), - Serialization(serde_json::Error), - Io(std::io::Error), -} - -impl std::fmt::Display for StorageError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Database(err) => write!(f, "Database error: {}", err), - Self::Serialization(err) => write!(f, "Serialization error: {}", err), - Self::Io(err) => write!(f, "IO error: {}", err), - } - } -} - -impl std::error::Error for StorageError {} - -impl From for StorageError { - fn from(err: rusqlite::Error) -> Self { - Self::Database(err) - } -} - -impl From for StorageError { - fn from(err: serde_json::Error) -> Self { - Self::Serialization(err) - } -} - -impl From for StorageError { - fn from(err: std::io::Error) -> Self { - Self::Io(err) - } -} - -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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - 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() - .map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?; - - 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 deleted file mode 100644 index a46c8ba..0000000 --- a/src/config.rs +++ /dev/null @@ -1,58 +0,0 @@ -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/health.rs b/src/health.rs index 6b6a613..5eedb00 100644 --- a/src/health.rs +++ b/src/health.rs @@ -1,9 +1,8 @@ -use crate::buffer::Storage; -use crate::metrics::Timestamp; -use log::{error, info}; +use anyhow::Result; +use log::info; use serde::{Deserialize, Serialize}; -use std::sync::{Arc, Mutex}; -use std::thread; +use std::sync::Arc; +use std::time::Instant; use tiny_http::{Response, Server}; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -11,108 +10,54 @@ pub struct HealthStatus { pub status: String, pub version: String, pub uptime_seconds: u64, - pub config_last_fetch: Option, - pub metrics_pending: usize, - pub last_error: Option, } -pub struct HealthServer { - start_time: Timestamp, - storage: Storage, - last_error: Arc>>, - config_last_fetch: Arc>>, -} +/// Start a simple health check HTTP server +pub async fn start_health_server(port: u16) -> Result<()> { + let start_time = Arc::new(Instant::now()); -impl HealthServer { - pub fn new(storage: Storage) -> Self { - Self { - start_time: Timestamp::now(), - storage, - last_error: Arc::new(Mutex::new(None)), - config_last_fetch: Arc::new(Mutex::new(None)), - } - } + tokio::task::spawn_blocking(move || { + let addr = format!("0.0.0.0:{}", port); + info!("Starting health endpoint on {}", addr); - pub fn start(self, port: u16) { - thread::spawn(move || { - let addr = format!("0.0.0.0:{}", port); - info!("Starting health endpoint on {}", addr); + let server = Server::http(&addr) + .map_err(|e| anyhow::anyhow!("Failed to start health server: {}", e))?; - let server = match Server::http(&addr) { - Ok(s) => s, - Err(e) => { - error!("Failed to start health server: {}", e); - return; - } - }; + for request in server.incoming_requests() { + let path = request.url(); - for request in server.incoming_requests() { - let path = request.url(); + if path == "/health" { + let uptime = start_time.elapsed().as_secs(); + let status = HealthStatus { + status: "healthy".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + uptime_seconds: uptime, + }; - if path == "/health" { - let status = self.get_health_status(); - let json = match serde_json::to_string(&status) { - Ok(j) => j, - Err(e) => { - error!("Failed to serialize health status: {}", e); - let _ = request.respond( - Response::from_string("Internal error").with_status_code(500), - ); - continue; - } - }; + let json = serde_json::to_string(&status).unwrap_or_else(|_| { + r#"{"status":"error","message":"Failed to serialize health status"}"# + .to_string() + }); - let response = Response::from_string(json) - .with_header( - tiny_http::Header::from_bytes( - &b"Content-Type"[..], - &b"application/json"[..], - ) - .unwrap(), + let response = Response::from_string(json) + .with_header( + tiny_http::Header::from_bytes( + &b"Content-Type"[..], + &b"application/json"[..], ) - .with_status_code(200); + .unwrap(), + ) + .with_status_code(200); - let _ = request.respond(response); - } else { - let _ = - request.respond(Response::from_string("Not found").with_status_code(404)); - } + let _ = request.respond(response); + } else { + let _ = request.respond(Response::from_string("Not found").with_status_code(404)); } - }); - } - - fn get_health_status(&self) -> HealthStatus { - let uptime = self.start_time.elapsed_secs() as u64; - - // Get pending metrics count - let metrics_pending = self - .storage - .get_pending_metrics(1000) - .map(|m| m.len()) - .unwrap_or(0); - - // Get last config fetch time - let config_last_fetch = self - .config_last_fetch - .lock() - .ok() - .and_then(|guard| *guard) - .map(|ts| ts.to_rfc3339()); - - // Get last error - let last_error = self - .last_error - .lock() - .ok() - .and_then(|guard| (*guard).clone()); - - HealthStatus { - status: "healthy".to_string(), - version: env!("CARGO_PKG_VERSION").to_string(), - uptime_seconds: uptime, - config_last_fetch, - metrics_pending, - last_error, } - } + + Ok::<(), anyhow::Error>(()) + }) + .await??; + + Ok(()) } diff --git a/src/main.rs b/src/main.rs index a432ad6..061a46a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,18 +1,14 @@ -mod api_client; -mod buffer; -mod config; mod health; -mod metrics; -mod poller; mod proto; mod snmp; mod version; +mod websocket_client; use chrono::Local; use clap::Parser; -use log::{info, LevelFilter, Metadata, Record}; -use poller::Scheduler; +use log::{info, warn, LevelFilter, Metadata, Record}; use std::env; +use websocket_client::AgentClient; /// Minimal logger that writes to stderr with timestamps struct SimpleLogger { @@ -50,21 +46,13 @@ fn init_logger() { #[command(name = "towerops-agent")] #[command(about = "Towerops remote SNMP polling agent", long_about = None)] struct Args { - /// API URL (e.g., https://app.towerops.com) + /// API URL (e.g., wss://app.towerops.com or 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] @@ -79,46 +67,41 @@ async fn main() { // Check for newer Docker image version version::check_for_updates(); - info!("API URL: {}", args.api_url); - info!( - "Config refresh interval: {} seconds", - args.config_refresh_seconds - ); - info!("Database path: {}", args.database_path); + // Convert HTTP(S) URL to WebSocket URL + let ws_url = if args.api_url.starts_with("http://") { + args.api_url.replace("http://", "ws://") + } else if args.api_url.starts_with("https://") { + args.api_url.replace("https://", "wss://") + } else if args.api_url.starts_with("ws://") || args.api_url.starts_with("wss://") { + args.api_url.clone() + } else { + // Default to wss:// for bare domains + format!("wss://{}", args.api_url) + }; - // Initialize components - let api_client = match api_client::ApiClient::new(args.api_url, args.token) { + info!("WebSocket URL: {}", ws_url); + + // Start simple health endpoint (no storage needed for WebSocket mode) + tokio::spawn(async { + if let Err(e) = health::start_health_server(8080).await { + warn!("Health server error: {}", e); + } + }); + + // Connect to Towerops server via WebSocket + let mut client = match AgentClient::connect(&ws_url, &args.token).await { Ok(client) => client, Err(e) => { - eprintln!("Failed to create API client: {}", e); + eprintln!("Failed to connect to server: {}", e); std::process::exit(1); } }; - let storage = match buffer::Storage::new(args.database_path) { - Ok(s) => s, - Err(e) => { - eprintln!("Failed to create storage: {}", e); - std::process::exit(1); - } - }; - - // Start health endpoint server - let health_server = health::HealthServer::new(storage.clone()); - health_server.start(8080); - - let snmp_client = snmp::SnmpClient::new(); - - // Create and run scheduler - let mut scheduler = Scheduler::new( - api_client, - storage, - snmp_client, - args.config_refresh_seconds, - ); - - if let Err(e) = scheduler.run().await { - eprintln!("Scheduler error: {}", e); + // Run the agent event loop + if let Err(e) = client.run().await { + eprintln!("Agent error: {}", e); std::process::exit(1); } + + info!("Agent shutting down"); } diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs deleted file mode 100644 index 1f98913..0000000 --- a/src/metrics/mod.rs +++ /dev/null @@ -1,169 +0,0 @@ -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), - #[serde(rename = "neighbor_discovery")] - NeighborDiscovery(NeighborDiscovery), -} - -impl Metric { - pub fn metric_type(&self) -> &str { - match self { - Metric::SensorReading(_) => "sensor_reading", - Metric::InterfaceStat(_) => "interface_stat", - Metric::NeighborDiscovery(_) => "neighbor_discovery", - } - } - - pub fn timestamp(&self) -> &Timestamp { - match self { - Metric::SensorReading(sr) => &sr.timestamp, - Metric::InterfaceStat(is) => &is.timestamp, - Metric::NeighborDiscovery(nd) => &nd.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, -} - -/// Neighbor discovery metric (LLDP/CDP) -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NeighborDiscovery { - pub interface_id: String, - pub protocol: String, - pub remote_chassis_id: String, - pub remote_system_name: String, - pub remote_system_description: String, - pub remote_platform: String, - pub remote_port_id: String, - pub remote_port_description: String, - pub remote_address: String, - pub remote_capabilities: Vec, - pub timestamp: Timestamp, -} diff --git a/src/poller/executor.rs b/src/poller/executor.rs deleted file mode 100644 index bfbacee..0000000 --- a/src/poller/executor.rs +++ /dev/null @@ -1,254 +0,0 @@ -use crate::buffer::Storage; -use crate::buffer::StorageError; - -#[derive(Debug)] -pub enum ExecutorError { - Storage(StorageError), - Snmp(crate::snmp::SnmpError), - Conversion(String), -} - -impl std::fmt::Display for ExecutorError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Storage(err) => write!(f, "Storage error: {}", err), - Self::Snmp(err) => write!(f, "SNMP error: {}", err), - Self::Conversion(msg) => write!(f, "Conversion error: {}", msg), - } - } -} - -impl std::error::Error for ExecutorError {} - -impl From for ExecutorError { - fn from(err: StorageError) -> Self { - Self::Storage(err) - } -} - -impl From for ExecutorError { - fn from(err: crate::snmp::SnmpError) -> Self { - Self::Snmp(err) - } -} - -pub type Result = std::result::Result; - -use crate::config::EquipmentConfig; -use crate::metrics::{InterfaceStat, Metric, SensorReading}; -use crate::snmp::{discover_neighbors, SnmpClient}; - -use crate::metrics::Timestamp; -use log::{error, info, warn}; - -/// Executor handles polling individual pieces of equipment -#[derive(Clone)] -pub struct Executor { - snmp_client: SnmpClient, - storage: Storage, -} - -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 - ); - - // Pre-extract SNMP credentials to avoid repeated field access in loop - let ip = &equipment.ip_address; - let community = &equipment.snmp.community; - let version = &equipment.snmp.version; - let port = equipment.snmp.port; - - for sensor in &equipment.sensors { - match self - .snmp_client - .get(ip, community, version, 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 - ); - - // Pre-extract SNMP credentials to avoid repeated field access in loop - let ip = &equipment.ip_address; - let community = &equipment.snmp.community; - let version = &equipment.snmp.version; - let port = equipment.snmp.port; - - 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 all counters in parallel for this interface - let ( - in_octets_result, - out_octets_result, - in_errors_result, - out_errors_result, - in_discards_result, - out_discards_result, - ) = tokio::join!( - self.get_counter(ip, community, version, port, &if_in_octets_oid), - self.get_counter(ip, community, version, port, &if_out_octets_oid), - self.get_counter(ip, community, version, port, &if_in_errors_oid), - self.get_counter(ip, community, version, port, &if_out_errors_oid), - self.get_counter(ip, community, version, port, &if_in_discards_oid), - self.get_counter(ip, community, version, port, &if_out_discards_oid), - ); - - let in_octets = in_octets_result.unwrap_or(0); - let out_octets = out_octets_result.unwrap_or(0); - let in_errors = in_errors_result.unwrap_or(0); - let out_errors = out_errors_result.unwrap_or(0); - let in_discards = in_discards_result.unwrap_or(0); - let out_discards = out_discards_result.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(()) - } - - /// Poll neighbors for a piece of equipment (LLDP/CDP) - pub async fn poll_neighbors(&self, equipment: &EquipmentConfig) -> Result<()> { - if !equipment.snmp.enabled || equipment.interfaces.is_empty() { - return Ok(()); - } - - info!("Polling neighbors for equipment: {}", equipment.name); - - // Pre-extract SNMP credentials - let ip = &equipment.ip_address; - let community = &equipment.snmp.community; - let version = &equipment.snmp.version; - let port = equipment.snmp.port; - - // Build interface list for neighbor discovery (interface_id, if_index) - let interfaces: Vec<(String, u32)> = equipment - .interfaces - .iter() - .map(|i| (i.id.clone(), i.if_index as u32)) - .collect(); - - // Discover neighbors using LLDP and CDP - let neighbors = - discover_neighbors(&self.snmp_client, ip, community, version, port, &interfaces).await; - - // Store neighbor discoveries - for neighbor in neighbors { - if let Err(e) = self - .storage - .store_metric(&Metric::NeighborDiscovery(neighbor)) - { - error!("Failed to store neighbor discovery: {}", 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 deleted file mode 100644 index 8c7e04d..0000000 --- a/src/poller/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -mod executor; -mod scheduler; - -pub use executor::Executor; -pub use scheduler::Scheduler; diff --git a/src/poller/scheduler.rs b/src/poller/scheduler.rs deleted file mode 100644 index 7dbe735..0000000 --- a/src/poller/scheduler.rs +++ /dev/null @@ -1,348 +0,0 @@ -use crate::api_client::ApiClient; -use crate::buffer::StorageError; - -#[derive(Debug)] -pub enum SchedulerError { - Api(crate::api_client::ApiError), - Storage(StorageError), - Snmp(crate::snmp::SnmpError), - Executor(super::executor::ExecutorError), -} - -impl std::fmt::Display for SchedulerError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Api(err) => write!(f, "API error: {}", err), - Self::Storage(err) => write!(f, "Storage error: {}", err), - Self::Snmp(err) => write!(f, "SNMP error: {}", err), - Self::Executor(err) => write!(f, "Executor error: {}", err), - } - } -} - -impl std::error::Error for SchedulerError {} - -impl From for SchedulerError { - fn from(err: crate::api_client::ApiError) -> Self { - Self::Api(err) - } -} - -impl From for SchedulerError { - fn from(err: StorageError) -> Self { - Self::Storage(err) - } -} - -impl From for SchedulerError { - fn from(err: crate::snmp::SnmpError) -> Self { - Self::Snmp(err) - } -} - -impl From for SchedulerError { - fn from(err: super::executor::ExecutorError) -> Self { - Self::Executor(err) - } -} - -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::sync::Arc; -use std::time::Duration; -use tokio::sync::Semaphore; -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 - let mut update_ticker = interval(Duration::from_secs(3600)); // Check for updates every hour - - 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); - } - } - - _ = update_ticker.tick() => { - self.check_and_update().await; - } - } - } - } - - 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<()> { - // Process metrics in batches until queue is empty or we hit an error - // This handles high-volume scenarios with 10,000+ equipment - let mut total_flushed = 0; - const BATCH_SIZE: usize = 500; - const MAX_BATCHES: usize = 20; // Limit to 10,000 metrics per flush cycle - - for _ in 0..MAX_BATCHES { - let pending = self.storage.get_pending_metrics(BATCH_SIZE)?; - - if pending.is_empty() { - break; - } - - let batch_size = 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)?; - total_flushed += batch_size; - } - Err(e) => { - warn!("Failed to submit batch of {} metrics: {}", batch_size, e); - // Don't return error, just log and continue with remaining batches - break; - } - } - - // If we got less than batch size, we've emptied the queue - if batch_size < BATCH_SIZE { - break; - } - } - - if total_flushed > 0 { - info!("Successfully flushed {} metrics to API", total_flushed); - } - - Ok(()) - } - - async fn send_heartbeat(&self) -> Result<()> { - let uptime = self.start_time.elapsed_secs() as u64; - - // Get hostname from environment or system - let hostname = std::env::var("HOSTNAME") - .or_else(|_| std::fs::read_to_string("/etc/hostname").map(|s| s.trim().to_string())) - .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()?; - - // Collect equipment that needs polling - let equipment_to_poll: Vec<_> = config - .equipment - .iter() - .filter(|eq| eq.snmp.enabled) - .filter(|eq| { - match poll_times.get(&eq.id) { - Some(last_poll) => { - let elapsed = last_poll.elapsed_secs() as u64; - elapsed >= eq.poll_interval_seconds - } - None => true, // Never polled before - } - }) - .collect(); - - if equipment_to_poll.is_empty() { - return Ok(()); - } - - info!( - "Polling {} equipment items in parallel", - equipment_to_poll.len() - ); - - // Limit concurrent polling to prevent overwhelming the system - // With 10,000+ equipment, we don't want 10,000 concurrent tasks - const MAX_CONCURRENT_POLLS: usize = 100; - let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_POLLS)); - - // Spawn parallel polling tasks with concurrency limit - let mut tasks = Vec::new(); - for equipment in equipment_to_poll { - let executor = self.executor.clone(); - let storage = self.storage.clone(); - let equipment = equipment.clone(); - let permit = semaphore.clone(); - - let task = tokio::spawn(async move { - // Acquire permit before polling (limits concurrency) - let _permit = match permit.acquire().await { - Ok(p) => p, - Err(e) => { - error!("Failed to acquire polling permit: {}", e); - return; - } - }; - - info!("Polling equipment: {}", equipment.name); - - // Poll sensors, interfaces, and neighbors in parallel - let (sensor_result, interface_result, neighbor_result) = tokio::join!( - executor.poll_sensors(&equipment), - executor.poll_interfaces(&equipment), - executor.poll_neighbors(&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); - } - - if let Err(e) = neighbor_result { - error!("Failed to poll neighbors for {}: {}", equipment.name, e); - } - - // Update last poll time - if let Err(e) = storage.update_last_poll_time(&equipment.id) { - error!("Failed to update last poll time: {}", e); - } - }); - - tasks.push(task); - } - - // Wait for all polling tasks to complete - for task in tasks { - if let Err(e) = task.await { - error!("Polling task failed: {}", e); - } - } - - Ok(()) - } - - async fn check_and_update(&self) { - info!("Checking for agent updates"); - - // Run version check in blocking thread to avoid blocking event loop - let result = tokio::task::spawn_blocking(crate::version::perform_self_update).await; - - match result { - Ok(Ok(true)) => { - info!("Update initiated, container will restart with new version"); - // perform_self_update calls std::process::exit(0), so we won't reach here - } - Ok(Ok(false)) => { - info!("Already running latest version"); - } - Ok(Err(e)) => { - warn!("Failed to perform self-update: {}", e); - } - Err(e) => { - error!("Update check task failed: {}", e); - } - } - } -} diff --git a/src/snmp/mod.rs b/src/snmp/mod.rs index 8227715..5c012c9 100644 --- a/src/snmp/mod.rs +++ b/src/snmp/mod.rs @@ -1,7 +1,5 @@ mod client; -mod neighbor; mod types; pub use client::SnmpClient; -pub use neighbor::discover_neighbors; -pub use types::SnmpError; +pub use types::{SnmpError, SnmpValue}; diff --git a/src/snmp/neighbor.rs b/src/snmp/neighbor.rs deleted file mode 100644 index 37926e4..0000000 --- a/src/snmp/neighbor.rs +++ /dev/null @@ -1,280 +0,0 @@ -use super::client::SnmpClient; -use super::types::{SnmpResult, SnmpValue}; -use crate::metrics::{NeighborDiscovery, Timestamp}; -use std::collections::HashMap; - -/// LLDP-MIB remote table base OID -const LLDP_REM_TABLE_OID: &str = "1.0.8802.1.1.2.1.4.1.1"; - -/// CISCO-CDP-MIB neighbor table base OID -const CDP_CACHE_TABLE_OID: &str = "1.3.6.1.4.1.9.9.23.1.2.1.1"; - -/// Discover neighbors using LLDP and CDP -pub async fn discover_neighbors( - client: &SnmpClient, - ip_address: &str, - community: &str, - version: &str, - port: u16, - interfaces: &[(String, u32)], // (interface_id, if_index) pairs -) -> Vec { - let mut neighbors = Vec::new(); - - // Discover LLDP neighbors - if let Ok(lldp_neighbors) = - discover_lldp_neighbors(client, ip_address, community, version, port, interfaces).await - { - neighbors.extend(lldp_neighbors); - } - - // Discover CDP neighbors - if let Ok(cdp_neighbors) = - discover_cdp_neighbors(client, ip_address, community, version, port, interfaces).await - { - neighbors.extend(cdp_neighbors); - } - - neighbors -} - -/// Discover LLDP neighbors -async fn discover_lldp_neighbors( - client: &SnmpClient, - ip_address: &str, - community: &str, - version: &str, - port: u16, - interfaces: &[(String, u32)], -) -> SnmpResult> { - let entries = client - .walk(ip_address, community, version, port, LLDP_REM_TABLE_OID) - .await?; - - if entries.is_empty() { - return Ok(Vec::new()); - } - - // Group entries by neighbor key (timemark.local_port_num.index) - let mut neighbors_map: HashMap> = HashMap::new(); - - for (oid, value) in entries { - if let Some(neighbor_key) = extract_lldp_neighbor_key(&oid) { - let field_name = extract_lldp_field_name(&oid); - let value_str = snmp_value_to_string(value); - - neighbors_map - .entry(neighbor_key) - .or_default() - .insert(field_name, value_str); - } - } - - // Convert to NeighborDiscovery structs - let mut neighbors = Vec::new(); - for (_key, fields) in neighbors_map { - if let Some(neighbor) = build_lldp_neighbor(fields, interfaces) { - neighbors.push(neighbor); - } - } - - Ok(neighbors) -} - -/// Discover CDP neighbors -async fn discover_cdp_neighbors( - client: &SnmpClient, - ip_address: &str, - community: &str, - version: &str, - port: u16, - interfaces: &[(String, u32)], -) -> SnmpResult> { - let entries = client - .walk(ip_address, community, version, port, CDP_CACHE_TABLE_OID) - .await?; - - if entries.is_empty() { - return Ok(Vec::new()); - } - - // Group entries by neighbor key (if_index.device_index) - let mut neighbors_map: HashMap> = HashMap::new(); - - for (oid, value) in entries { - if let Some(neighbor_key) = extract_cdp_neighbor_key(&oid) { - let field_name = extract_cdp_field_name(&oid); - let value_str = snmp_value_to_string(value); - - neighbors_map - .entry(neighbor_key) - .or_default() - .insert(field_name, value_str); - } - } - - // Convert to NeighborDiscovery structs - let mut neighbors = Vec::new(); - for (_key, fields) in neighbors_map { - if let Some(neighbor) = build_cdp_neighbor(fields, interfaces) { - neighbors.push(neighbor); - } - } - - Ok(neighbors) -} - -/// Extract neighbor key from LLDP OID (timemark.local_port_num.index) -fn extract_lldp_neighbor_key(oid: &str) -> Option { - // OID format: 1.0.8802.1.1.2.1.4.1.1.X.timemark.local_port_num.index - let parts: Vec<&str> = oid.split('.').collect(); - if parts.len() >= 14 { - // Extract last 3 parts as neighbor key - Some(parts[parts.len() - 3..].join(".")) - } else { - None - } -} - -/// Extract field name from LLDP OID -fn extract_lldp_field_name(oid: &str) -> String { - // OID format: 1.0.8802.1.1.2.1.4.1.1.X.timemark.local_port_num.index - // X is the field identifier - let parts: Vec<&str> = oid.split('.').collect(); - if parts.len() >= 11 { - parts[10].to_string() - } else { - "unknown".to_string() - } -} - -/// Extract neighbor key from CDP OID (if_index.device_index) -fn extract_cdp_neighbor_key(oid: &str) -> Option { - // OID format: 1.3.6.1.4.1.9.9.23.1.2.1.1.X.if_index.device_index - let parts: Vec<&str> = oid.split('.').collect(); - if parts.len() >= 13 { - // Extract last 2 parts as neighbor key - Some(parts[parts.len() - 2..].join(".")) - } else { - None - } -} - -/// Extract field name from CDP OID -fn extract_cdp_field_name(oid: &str) -> String { - // OID format: 1.3.6.1.4.1.9.9.23.1.2.1.1.X.if_index.device_index - // X is the field identifier - let parts: Vec<&str> = oid.split('.').collect(); - if parts.len() >= 11 { - parts[10].to_string() - } else { - "unknown".to_string() - } -} - -/// Convert SnmpValue to string -fn snmp_value_to_string(value: SnmpValue) -> String { - match value { - SnmpValue::String(s) => s, - SnmpValue::Integer(i) => i.to_string(), - SnmpValue::Counter32(c) => c.to_string(), - SnmpValue::Counter64(c) => c.to_string(), - SnmpValue::Gauge32(g) => g.to_string(), - SnmpValue::TimeTicks(t) => t.to_string(), - SnmpValue::IpAddress(ip) => ip, - } -} - -/// Build LLDP neighbor from fields -fn build_lldp_neighbor( - fields: HashMap, - interfaces: &[(String, u32)], -) -> Option { - // Field IDs from LLDP-MIB - // 4 = lldpRemChassisIdSubtype - // 5 = lldpRemChassisId - // 6 = lldpRemPortIdSubtype - // 7 = lldpRemPortId - // 8 = lldpRemPortDesc - // 9 = lldpRemSysName - // 10 = lldpRemSysDesc - // 12 = lldpRemManAddr - - let local_port_num = fields.get("local_port_num")?; - let if_index: u32 = local_port_num.parse().ok()?; - - let interface_id = interfaces - .iter() - .find(|(_, idx)| *idx == if_index) - .map(|(id, _)| id.clone())?; - - let remote_chassis_id = fields.get("5").cloned().unwrap_or_default(); - let remote_system_name = fields.get("9").cloned().unwrap_or_default(); - let remote_system_description = fields.get("10").cloned().unwrap_or_default(); - let remote_port_id = fields.get("7").cloned().unwrap_or_default(); - let remote_port_description = fields.get("8").cloned().unwrap_or_default(); - let remote_address = fields.get("12").cloned().unwrap_or_default(); - - // Parse capabilities (if available) - let remote_capabilities = Vec::new(); // TODO: Parse from lldpRemSysCapEnabled - - Some(NeighborDiscovery { - interface_id, - protocol: "lldp".to_string(), - remote_chassis_id, - remote_system_name, - remote_system_description, - remote_platform: String::new(), - remote_port_id, - remote_port_description, - remote_address, - remote_capabilities, - timestamp: Timestamp::now(), - }) -} - -/// Build CDP neighbor from fields -fn build_cdp_neighbor( - fields: HashMap, - interfaces: &[(String, u32)], -) -> Option { - // Field IDs from CISCO-CDP-MIB - // 4 = cdpCacheAddressType - // 5 = cdpCacheAddress - // 6 = cdpCacheVersion - // 7 = cdpCacheDeviceId - // 8 = cdpCacheDevicePort - // 9 = cdpCachePlatform - // 10 = cdpCacheCapabilities - - let if_index_str = fields.get("if_index")?; - let if_index: u32 = if_index_str.parse().ok()?; - - let interface_id = interfaces - .iter() - .find(|(_, idx)| *idx == if_index) - .map(|(id, _)| id.clone())?; - - let remote_chassis_id = fields.get("7").cloned().unwrap_or_default(); - let remote_system_name = remote_chassis_id.clone(); - let remote_system_description = fields.get("6").cloned().unwrap_or_default(); - let remote_platform = fields.get("9").cloned().unwrap_or_default(); - let remote_port_id = fields.get("8").cloned().unwrap_or_default(); - let remote_address = fields.get("5").cloned().unwrap_or_default(); - - // Parse capabilities (if available) - let remote_capabilities = Vec::new(); // TODO: Parse from cdpCacheCapabilities - - Some(NeighborDiscovery { - interface_id, - protocol: "cdp".to_string(), - remote_chassis_id, - remote_system_name, - remote_system_description, - remote_platform, - remote_port_id, - remote_port_description: String::new(), - remote_address, - remote_capabilities, - timestamp: Timestamp::now(), - }) -} diff --git a/src/websocket_client.rs b/src/websocket_client.rs index 6778bf0..b655555 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -4,14 +4,22 @@ /// persistent WebSocket connection. The server sends SNMP query jobs as protobuf /// messages, the agent executes raw SNMP queries, and sends results back. use anyhow::{Context, Result}; +use base64::engine::general_purpose::STANDARD as BASE64; +use base64::Engine; use futures::{SinkExt, StreamExt}; use prost::Message; use std::collections::HashMap; use tokio::net::TcpStream; +use tokio::sync::mpsc; use tokio::time::{interval, Duration}; -use tokio_tungstenite::{connect_async, tungstenite::protocol::Message as WsMessage, MaybeTlsStream, WebSocketStream}; +use tokio_tungstenite::{ + connect_async, tungstenite::protocol::Message as WsMessage, MaybeTlsStream, WebSocketStream, +}; -use crate::proto::{AgentError, AgentHeartbeat, AgentJob, AgentJobList, JobType, QueryType, SnmpResult}; +use crate::proto::agent::{ + AgentError, AgentHeartbeat, AgentJob, AgentJobList, JobType, QueryType, SnmpResult, +}; +use crate::snmp::{SnmpClient, SnmpValue}; /// Phoenix channel message format (JSON wrapper around binary protobuf). #[derive(Debug, serde::Serialize, serde::Deserialize)] @@ -28,6 +36,8 @@ pub struct AgentClient { ws_stream: WebSocketStream>, token: String, agent_id: String, + result_tx: mpsc::UnboundedSender, + result_rx: mpsc::UnboundedReceiver, } impl AgentClient { @@ -49,10 +59,14 @@ impl AgentClient { log::info!("Connected to Towerops server at {}", url); + let (result_tx, result_rx) = mpsc::unbounded_channel(); + Ok(Self { ws_stream, token: token.to_string(), agent_id: generate_agent_id(), + result_tx, + result_rx, }) } @@ -93,6 +107,11 @@ impl AgentClient { } } + // Receive SNMP results from job tasks + Some(result) = self.result_rx.recv() => { + self.send_result(result).await?; + } + // Send periodic heartbeats _ = heartbeat_interval.tick() => { self.send_heartbeat().await?; @@ -112,7 +131,7 @@ impl AgentClient { // Extract binary protobuf from payload if let serde_json::Value::Object(map) = phoenix_msg.payload { if let Some(serde_json::Value::String(binary_b64)) = map.get("binary") { - let binary = base64::decode(binary_b64)?; + let binary = BASE64.decode(binary_b64)?; let job_list = AgentJobList::decode(&binary[..])?; self.handle_jobs(job_list).await?; } @@ -145,8 +164,9 @@ impl AgentClient { log::info!("Executing job: {} (type: {:?})", job.job_id, job_type); // Spawn task to execute job + let result_tx = self.result_tx.clone(); tokio::spawn(async move { - if let Err(e) = execute_job(job).await { + if let Err(e) = execute_job(job, result_tx).await { log::error!("Job execution failed: {}", e); } }); @@ -170,7 +190,7 @@ impl AgentClient { let msg = PhoenixMessage { topic: format!("agent:{}", self.agent_id), event: "heartbeat".to_string(), - payload: serde_json::json!({"binary": base64::encode(&binary)}), + payload: serde_json::json!({"binary": BASE64.encode(&binary)}), reference: None, }; @@ -188,7 +208,7 @@ impl AgentClient { let msg = PhoenixMessage { topic: format!("agent:{}", self.agent_id), event: "result".to_string(), - payload: serde_json::json!({"binary": base64::encode(&binary)}), + payload: serde_json::json!({"binary": BASE64.encode(&binary)}), reference: None, }; @@ -206,7 +226,7 @@ impl AgentClient { let msg = PhoenixMessage { topic: format!("agent:{}", self.agent_id), event: "error".to_string(), - payload: serde_json::json!({"binary": base64::encode(&binary)}), + payload: serde_json::json!({"binary": BASE64.encode(&binary)}), reference: None, }; @@ -219,9 +239,10 @@ impl AgentClient { } /// Execute an SNMP job and collect results. -async fn execute_job(job: AgentJob) -> Result<()> { +async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender) -> Result<()> { let device = job.device.context("Job missing device info")?; - let mut oid_values = HashMap::new(); + let mut oid_values: HashMap = HashMap::new(); + let snmp_client = SnmpClient::new(); for query in job.queries { let query_type = QueryType::from_i32(query.query_type).unwrap_or(QueryType::Get); @@ -230,9 +251,18 @@ async fn execute_job(job: AgentJob) -> Result<()> { QueryType::Get => { // Execute SNMP GET for each OID for oid in &query.oids { - match snmp_get(&device.ip, &device.community, &device.version, device.port, oid).await { + match snmp_client + .get( + &device.ip, + &device.community, + &device.version, + device.port as u16, + oid, + ) + .await + { Ok(value) => { - oid_values.insert(oid.clone(), value); + oid_values.insert(oid.clone(), value_to_string(value)); } Err(e) => { log::warn!("SNMP GET failed for OID {}: {}", oid, e); @@ -243,9 +273,20 @@ async fn execute_job(job: AgentJob) -> Result<()> { QueryType::Walk => { // Execute SNMP WALK for each base OID for base_oid in &query.oids { - match snmp_walk(&device.ip, &device.community, &device.version, device.port, base_oid).await { + match snmp_client + .walk( + &device.ip, + &device.community, + &device.version, + device.port as u16, + base_oid, + ) + .await + { Ok(results) => { - oid_values.extend(results); + for (oid, value) in results { + oid_values.insert(oid, value_to_string(value)); + } } Err(e) => { log::warn!("SNMP WALK failed for OID {}: {}", base_oid, e); @@ -266,56 +307,29 @@ async fn execute_job(job: AgentJob) -> Result<()> { .as_secs() as i64, }; - // Send result back to server - // TODO: Get client reference to send result - log::info!("Collected {} OID values for job {}", result.oid_values.len(), job.job_id); + log::info!( + "Collected {} OID values for job {}", + result.oid_values.len(), + job.job_id + ); + + // Send result back to main client task + result_tx.send(result).ok(); Ok(()) } -/// Execute SNMP GET operation. -/// -/// Returns the value as a string (already formatted). -async fn snmp_get( - ip: &str, - community: &str, - version: &str, - port: u32, - oid: &str, -) -> Result { - // TODO: Implement raw SNMP GET over UDP - // 1. Construct SNMP GET PDU (BER encoded) - // 2. Send UDP packet to ip:port - // 3. Wait for response with timeout - // 4. Parse response and extract value - // 5. Format value as string - - log::debug!("SNMP GET: {} @ {}:{} (community: {}, version: {})", oid, ip, port, community, version); - - // Placeholder - Err(anyhow::anyhow!("SNMP GET not yet implemented")) -} - -/// Execute SNMP WALK operation. -/// -/// Returns a map of OID → value (all as strings). -async fn snmp_walk( - ip: &str, - community: &str, - version: &str, - port: u32, - base_oid: &str, -) -> Result> { - // TODO: Implement raw SNMP WALK over UDP - // 1. Start with GETNEXT(base_oid) - // 2. Loop: GETNEXT(last_oid) until OID no longer starts with base_oid - // 3. Collect all OID/value pairs - // 4. Return map - - log::debug!("SNMP WALK: {} @ {}:{} (community: {}, version: {})", base_oid, ip, port, community, version); - - // Placeholder - Err(anyhow::anyhow!("SNMP WALK not yet implemented")) +/// Convert SnmpValue to String for protobuf transmission. +fn value_to_string(value: SnmpValue) -> String { + match value { + SnmpValue::Integer(i) => i.to_string(), + SnmpValue::String(s) => s, + SnmpValue::Counter32(c) => c.to_string(), + SnmpValue::Counter64(c) => c.to_string(), + SnmpValue::Gauge32(g) => g.to_string(), + SnmpValue::TimeTicks(t) => t.to_string(), + SnmpValue::IpAddress(ip) => ip, + } } /// Generate a unique agent ID.