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)
This commit is contained in:
parent
570a37ffc0
commit
967d317b69
15 changed files with 242 additions and 2001 deletions
165
Cargo.lock
generated
165
Cargo.lock
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<string, string> 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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<std::io::Error> for ApiError {
|
||||
fn from(err: std::io::Error) -> Self {
|
||||
Self::JsonError(err)
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, ApiError>;
|
||||
|
||||
/// API client for communicating with the Towerops server
|
||||
#[derive(Clone)]
|
||||
pub struct ApiClient {
|
||||
base_url: String,
|
||||
token: String,
|
||||
}
|
||||
|
||||
impl ApiClient {
|
||||
/// Create a new API client
|
||||
pub fn new(base_url: String, token: String) -> Result<Self> {
|
||||
// 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<AgentConfig> {
|
||||
let url = format!("{}/api/v1/agent/config", self.base_url);
|
||||
let token = self.token.clone();
|
||||
|
||||
let config = tokio::task::spawn_blocking(move || {
|
||||
let response = ureq::get(&url)
|
||||
.set("Authorization", &format!("Bearer {}", token))
|
||||
.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<Metric>) -> Result<()> {
|
||||
if metrics.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let url = format!("{}/api/v1/agent/metrics", self.base_url);
|
||||
let token = self.token.clone();
|
||||
|
||||
tokio::task::spawn_blocking(move || {
|
||||
// Convert metrics to protobuf
|
||||
let proto_metrics: Vec<agent::Metric> = metrics
|
||||
.into_iter()
|
||||
.map(|m| convert_metric_to_proto(&m))
|
||||
.collect();
|
||||
|
||||
let batch = agent::MetricBatch {
|
||||
metrics: proto_metrics,
|
||||
};
|
||||
|
||||
// Encode to bytes
|
||||
let mut buf = Vec::new();
|
||||
batch
|
||||
.encode(&mut buf)
|
||||
.map_err(|e| ApiError::RequestFailed(format!("Protobuf encoding error: {}", e)))?;
|
||||
|
||||
// Send as protobuf
|
||||
let response = ureq::post(&url)
|
||||
.set("Authorization", &format!("Bearer {}", token))
|
||||
.set("Content-Type", "application/x-protobuf")
|
||||
.timeout(Duration::from_secs(30))
|
||||
.send_bytes(&buf)
|
||||
.map_err(|e| ApiError::RequestFailed(e.to_string()))?;
|
||||
|
||||
let status = response.status();
|
||||
if status != 200 {
|
||||
return Err(ApiError::StatusError(status));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
.map_err(|e| ApiError::JoinError(e.to_string()))??;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Send heartbeat to the API 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<String, serde_json::Value> = 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(),
|
||||
},
|
||||
)),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
mod storage;
|
||||
|
||||
pub use storage::{Storage, StorageError};
|
||||
|
|
@ -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<rusqlite::Error> for StorageError {
|
||||
fn from(err: rusqlite::Error) -> Self {
|
||||
Self::Database(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<serde_json::Error> for StorageError {
|
||||
fn from(err: serde_json::Error) -> Self {
|
||||
Self::Serialization(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<std::io::Error> for StorageError {
|
||||
fn from(err: std::io::Error) -> Self {
|
||||
Self::Io(err)
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, StorageError>;
|
||||
|
||||
/// SQLite storage for buffering metrics when API is unavailable
|
||||
#[derive(Clone)]
|
||||
pub struct Storage {
|
||||
conn: Arc<Mutex<Connection>>,
|
||||
}
|
||||
|
||||
impl Storage {
|
||||
/// Create a new storage instance
|
||||
pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
|
||||
let conn = Connection::open(path)?;
|
||||
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS metrics (
|
||||
id INTEGER PRIMARY KEY,
|
||||
metric_type TEXT NOT NULL,
|
||||
data TEXT NOT NULL,
|
||||
timestamp TEXT NOT NULL,
|
||||
sent INTEGER DEFAULT 0,
|
||||
created_at TEXT DEFAULT CURRENT_TIMESTAMP
|
||||
)",
|
||||
[],
|
||||
)?;
|
||||
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_metrics_sent ON metrics(sent, created_at)",
|
||||
[],
|
||||
)?;
|
||||
|
||||
conn.execute(
|
||||
"CREATE TABLE IF NOT EXISTS last_poll_times (
|
||||
equipment_id TEXT PRIMARY KEY,
|
||||
last_poll_time TEXT NOT NULL
|
||||
)",
|
||||
[],
|
||||
)?;
|
||||
|
||||
Ok(Self {
|
||||
conn: Arc::new(Mutex::new(conn)),
|
||||
})
|
||||
}
|
||||
|
||||
/// Store a metric in the buffer
|
||||
pub fn store_metric(&self, metric: &Metric) -> Result<()> {
|
||||
let data = serde_json::to_string(metric)?;
|
||||
let conn = self
|
||||
.conn
|
||||
.lock()
|
||||
.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<Vec<(i64, Metric)>> {
|
||||
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::<Metric>(&data)
|
||||
.ok()
|
||||
.map(|metric| (id, metric))
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(metrics)
|
||||
}
|
||||
|
||||
/// Mark metrics as sent
|
||||
pub fn mark_metrics_sent(&self, ids: &[i64]) -> Result<()> {
|
||||
if ids.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let conn = self
|
||||
.conn
|
||||
.lock()
|
||||
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||
let placeholders = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
|
||||
let query = format!("UPDATE metrics SET sent = 1 WHERE id IN ({})", placeholders);
|
||||
|
||||
let params: Vec<_> = ids.iter().map(|id| id as &dyn rusqlite::ToSql).collect();
|
||||
conn.execute(&query, params.as_slice())?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Clean up old metrics that have been sent
|
||||
pub fn cleanup_old_metrics(&self) -> Result<()> {
|
||||
let conn = self
|
||||
.conn
|
||||
.lock()
|
||||
.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<Option<Timestamp>> {
|
||||
let conn = self
|
||||
.conn
|
||||
.lock()
|
||||
.map_err(|_| StorageError::Database(rusqlite::Error::InvalidQuery))?;
|
||||
|
||||
let result: std::result::Result<String, _> = conn.query_row(
|
||||
"SELECT last_poll_time FROM last_poll_times WHERE equipment_id = ?1",
|
||||
params![equipment_id],
|
||||
|row| row.get(0),
|
||||
);
|
||||
|
||||
match result {
|
||||
Ok(_timestamp_str) => {
|
||||
// Return a Timestamp - parsing from string not implemented yet
|
||||
// For now just return None since this function isn't used yet
|
||||
Ok(None)
|
||||
}
|
||||
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
|
||||
Err(e) => Err(e.into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Update the last poll time for equipment
|
||||
pub fn update_last_poll_time(&self, equipment_id: &str) -> Result<()> {
|
||||
let conn = self
|
||||
.conn
|
||||
.lock()
|
||||
.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<std::collections::HashMap<String, Timestamp>> {
|
||||
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<String, Timestamp> = stmt
|
||||
.query_map([], |row| {
|
||||
let equipment_id: String = row.get(0)?;
|
||||
let _timestamp_str: String = row.get(1)?;
|
||||
Ok(equipment_id)
|
||||
})?
|
||||
.filter_map(|r| r.ok())
|
||||
.map(|id| (id, Timestamp::now())) // Simplified - not parsing timestamps yet
|
||||
.collect();
|
||||
|
||||
Ok(times)
|
||||
}
|
||||
}
|
||||
|
|
@ -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<EquipmentConfig>,
|
||||
}
|
||||
|
||||
/// Configuration for a single piece of equipment
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct EquipmentConfig {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub ip_address: String,
|
||||
pub snmp: SnmpConfig,
|
||||
pub poll_interval_seconds: u64,
|
||||
pub sensors: Vec<SensorConfig>,
|
||||
pub interfaces: Vec<InterfaceConfig>,
|
||||
}
|
||||
|
||||
/// SNMP configuration
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct SnmpConfig {
|
||||
pub enabled: bool,
|
||||
pub version: String,
|
||||
pub community: String,
|
||||
pub port: u16,
|
||||
}
|
||||
|
||||
/// Sensor configuration
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct SensorConfig {
|
||||
pub id: String,
|
||||
#[serde(rename = "type")]
|
||||
pub sensor_type: String,
|
||||
pub oid: String,
|
||||
pub divisor: Option<i32>,
|
||||
pub unit: Option<String>,
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// Interface configuration
|
||||
#[derive(Debug, Clone, Deserialize, Serialize)]
|
||||
pub struct InterfaceConfig {
|
||||
pub id: String,
|
||||
pub if_index: i32,
|
||||
pub if_name: String,
|
||||
}
|
||||
|
||||
/// Heartbeat metadata sent to the API
|
||||
#[derive(Debug, Serialize)]
|
||||
pub struct HeartbeatMetadata {
|
||||
pub version: String,
|
||||
pub hostname: String,
|
||||
pub uptime_seconds: u64,
|
||||
}
|
||||
139
src/health.rs
139
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<String>,
|
||||
pub metrics_pending: usize,
|
||||
pub last_error: Option<String>,
|
||||
}
|
||||
|
||||
pub struct HealthServer {
|
||||
start_time: Timestamp,
|
||||
storage: Storage,
|
||||
last_error: Arc<Mutex<Option<String>>>,
|
||||
config_last_fetch: Arc<Mutex<Option<Timestamp>>>,
|
||||
}
|
||||
/// 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(())
|
||||
}
|
||||
|
|
|
|||
81
src/main.rs
81
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");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
serializer.serialize_str(&self.to_rfc3339())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for Timestamp {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let _s = String::deserialize(deserializer)?;
|
||||
// Simple RFC3339 parsing - accepts the format we produce
|
||||
// For a production system, you might want more robust parsing
|
||||
Ok(Timestamp { secs: 0 }) // Simplified - we mainly serialize, not deserialize
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert days since Unix epoch to year/month/day
|
||||
fn days_to_ymd(mut days: i64) -> (i32, u8, u8) {
|
||||
let mut year = 1970;
|
||||
loop {
|
||||
let days_in_year = if is_leap_year(year) { 366 } else { 365 };
|
||||
if days < days_in_year {
|
||||
break;
|
||||
}
|
||||
days -= days_in_year;
|
||||
year += 1;
|
||||
}
|
||||
|
||||
let days_in_months = if is_leap_year(year) {
|
||||
[31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
|
||||
} else {
|
||||
[31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
|
||||
};
|
||||
|
||||
let mut month = 1;
|
||||
for &dim in &days_in_months {
|
||||
if days < dim as i64 {
|
||||
break;
|
||||
}
|
||||
days -= dim as i64;
|
||||
month += 1;
|
||||
}
|
||||
|
||||
(year, month, days as u8 + 1)
|
||||
}
|
||||
|
||||
fn is_leap_year(year: i32) -> bool {
|
||||
(year % 4 == 0 && year % 100 != 0) || (year % 400 == 0)
|
||||
}
|
||||
|
||||
/// Metric types that can be submitted to the API
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum Metric {
|
||||
#[serde(rename = "sensor_reading")]
|
||||
SensorReading(SensorReading),
|
||||
#[serde(rename = "interface_stat")]
|
||||
InterfaceStat(InterfaceStat),
|
||||
#[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<String>,
|
||||
pub timestamp: Timestamp,
|
||||
}
|
||||
|
|
@ -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<StorageError> for ExecutorError {
|
||||
fn from(err: StorageError) -> Self {
|
||||
Self::Storage(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<crate::snmp::SnmpError> for ExecutorError {
|
||||
fn from(err: crate::snmp::SnmpError) -> Self {
|
||||
Self::Snmp(err)
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, ExecutorError>;
|
||||
|
||||
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<i64> {
|
||||
let value = self
|
||||
.snmp_client
|
||||
.get(ip_address, community, version, port, oid)
|
||||
.await?;
|
||||
|
||||
value
|
||||
.as_i64()
|
||||
.ok_or_else(|| ExecutorError::Conversion("Could not convert value to i64".into()))
|
||||
}
|
||||
}
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
mod executor;
|
||||
mod scheduler;
|
||||
|
||||
pub use executor::Executor;
|
||||
pub use scheduler::Scheduler;
|
||||
|
|
@ -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<crate::api_client::ApiError> for SchedulerError {
|
||||
fn from(err: crate::api_client::ApiError) -> Self {
|
||||
Self::Api(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<StorageError> for SchedulerError {
|
||||
fn from(err: StorageError) -> Self {
|
||||
Self::Storage(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<crate::snmp::SnmpError> for SchedulerError {
|
||||
fn from(err: crate::snmp::SnmpError) -> Self {
|
||||
Self::Snmp(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<super::executor::ExecutorError> for SchedulerError {
|
||||
fn from(err: super::executor::ExecutorError) -> Self {
|
||||
Self::Executor(err)
|
||||
}
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, SchedulerError>;
|
||||
|
||||
use crate::buffer::Storage;
|
||||
use crate::config::{AgentConfig, HeartbeatMetadata};
|
||||
use crate::poller::Executor;
|
||||
use crate::snmp::SnmpClient;
|
||||
|
||||
use crate::metrics::Timestamp;
|
||||
use log::{error, info, warn};
|
||||
use std::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<AgentConfig>,
|
||||
start_time: Timestamp,
|
||||
}
|
||||
|
||||
impl Scheduler {
|
||||
pub fn new(
|
||||
api_client: ApiClient,
|
||||
storage: Storage,
|
||||
snmp_client: SnmpClient,
|
||||
config_refresh_seconds: u64,
|
||||
) -> Self {
|
||||
let executor = Executor::new(snmp_client, storage.clone());
|
||||
|
||||
Self {
|
||||
api_client,
|
||||
storage,
|
||||
executor,
|
||||
config_refresh_seconds,
|
||||
current_config: None,
|
||||
start_time: Timestamp::now(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Run the main event loop
|
||||
pub async fn run(&mut self) -> Result<()> {
|
||||
info!("Starting Towerops agent scheduler");
|
||||
|
||||
// Fetch initial configuration
|
||||
if let Err(e) = self.refresh_config().await {
|
||||
error!("Failed to fetch initial config: {}", e);
|
||||
}
|
||||
|
||||
let mut config_ticker = interval(Duration::from_secs(self.config_refresh_seconds));
|
||||
let mut metrics_ticker = interval(Duration::from_secs(30));
|
||||
let mut heartbeat_ticker = interval(Duration::from_secs(60));
|
||||
let mut cleanup_ticker = interval(Duration::from_secs(3600)); // Cleanup every hour
|
||||
let mut poll_ticker = interval(Duration::from_secs(5)); // Check if polling needed every 5s
|
||||
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<i64> = pending.iter().map(|(id, _)| *id).collect();
|
||||
let metrics: Vec<_> = pending.into_iter().map(|(_, m)| m).collect();
|
||||
|
||||
match self.api_client.submit_metrics(metrics).await {
|
||||
Ok(_) => {
|
||||
self.storage.mark_metrics_sent(&ids)?;
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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};
|
||||
|
|
|
|||
|
|
@ -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<NeighborDiscovery> {
|
||||
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<Vec<NeighborDiscovery>> {
|
||||
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<String, HashMap<String, String>> = 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<Vec<NeighborDiscovery>> {
|
||||
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<String, HashMap<String, String>> = 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<String> {
|
||||
// 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<String> {
|
||||
// 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<String, String>,
|
||||
interfaces: &[(String, u32)],
|
||||
) -> Option<NeighborDiscovery> {
|
||||
// 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<String, String>,
|
||||
interfaces: &[(String, u32)],
|
||||
) -> Option<NeighborDiscovery> {
|
||||
// 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(),
|
||||
})
|
||||
}
|
||||
|
|
@ -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<MaybeTlsStream<TcpStream>>,
|
||||
token: String,
|
||||
agent_id: String,
|
||||
result_tx: mpsc::UnboundedSender<SnmpResult>,
|
||||
result_rx: mpsc::UnboundedReceiver<SnmpResult>,
|
||||
}
|
||||
|
||||
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<SnmpResult>) -> Result<()> {
|
||||
let device = job.device.context("Job missing device info")?;
|
||||
let mut oid_values = HashMap::new();
|
||||
let mut oid_values: HashMap<String, String> = 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<String> {
|
||||
// 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<HashMap<String, String>> {
|
||||
// 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.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue