graham/towerops-agent
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

structured logging

Browse source
This commit is contained in:
Graham McIntire 2026-01-31 15:40:42 -06:00
parent b8aecf0a55
commit 98d1bbbb21
No known key found for this signature in database
7 changed files with 168 additions and 354 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
Cargo.lock generated
View file

@ -534,6 +534,12 @@ version = "1.0.17"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2"
[[package]]
name = "lazy_static"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]] [[package]]
name = "libc" name = "libc"
version = "0.2.180" version = "0.2.180"
@ -558,6 +564,15 @@ version = "0.4.29"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
[[package]]
name = "matchers"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
dependencies = [
"regex-automata",
]
[[package]] [[package]]
name = "memchr" name = "memchr"
version = "2.7.6" version = "2.7.6"
@ -581,6 +596,15 @@ version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
[[package]]
name = "nu-ansi-term"
version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
"windows-sys 0.61.2",
]
[[package]] [[package]]
name = "once_cell" name = "once_cell"
version = "1.21.3" version = "1.21.3"
@ -905,6 +929,15 @@ dependencies = [
"digest", "digest",
] ]
[[package]]
name = "sharded-slab"
version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
dependencies = [
"lazy_static",
]
[[package]] [[package]]
name = "shlex" name = "shlex"
version = "1.3.0" version = "1.3.0"
@ -1022,6 +1055,15 @@ dependencies = [
"syn", "syn",
] ]
[[package]]
name = "thread_local"
version = "1.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185"
dependencies = [
"cfg-if",
]
[[package]] [[package]]
name = "tinystr" name = "tinystr"
version = "0.8.2" version = "0.8.2"
@ -1101,6 +1143,69 @@ dependencies = [
"snmp", "snmp",
"tokio", "tokio",
"tokio-tungstenite", "tokio-tungstenite",
"tracing",
"tracing-subscriber",
]
[[package]]
name = "tracing"
version = "0.1.44"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
dependencies = [
"pin-project-lite",
"tracing-attributes",
"tracing-core",
]
[[package]]
name = "tracing-attributes"
version = "0.1.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "tracing-core"
version = "0.1.36"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
dependencies = [
"once_cell",
"valuable",
]
[[package]]
name = "tracing-log"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
dependencies = [
"log",
"once_cell",
"tracing-core",
]
[[package]]
name = "tracing-subscriber"
version = "0.3.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e"
dependencies = [
"matchers",
"nu-ansi-term",
"once_cell",
"regex-automata",
"sharded-slab",
"smallvec",
"thread_local",
"tracing",
"tracing-core",
"tracing-log",
] ]
[[package]] [[package]]
@ -1172,6 +1277,12 @@ version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "valuable"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
[[package]] [[package]]
name = "version_check" name = "version_check"
version = "0.9.5" version = "0.9.5"

2
Cargo.toml
View file

@ -14,6 +14,8 @@ clap = { version = "4.0", features = ["derive", "env"] }
prost = "0.13" prost = "0.13"
prost-types = "0.13" prost-types = "0.13"
regex-lite = "0.1" regex-lite = "0.1"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
[build-dependencies] [build-dependencies]
prost-build = "0.13" prost-build = "0.13"

16
docker-compose.example.yml
View file

@ -1,5 +1,3 @@
version: '3.8'
services: services:
towerops-agent: towerops-agent:
image: registry.gitlab.com/towerops/towerops-agent:latest image: registry.gitlab.com/towerops/towerops-agent:latest
@ -29,17 +27,3 @@ services:
ports: ports:
# SNMP trap listener (UDP) # SNMP trap listener (UDP)
- "162:162/udp" - "162:162/udp"
volumes:
# Persistent storage for metrics buffering
- ./data:/data
# Optional: Resource limits
deploy:
resources:
limits:
cpus: '0.5'
memory: 512M
reservations:
cpus: '0.1'
memory: 128M

327
src/main.rs
View file

@ -7,151 +7,22 @@ use clap::Parser;
use std::env; use std::env;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::Duration;
use tokio::sync::watch; use tokio::sync::watch;
use tokio::time::sleep; use tokio::time::sleep;
use tracing_subscriber::EnvFilter;
use websocket_client::AgentClient; use websocket_client::AgentClient;
// Log levels
pub(crate) static LOG_LEVEL: std::sync::OnceLock<LogLevel> = std::sync::OnceLock::new();
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum LogLevel {
Error = 1,
Warn = 2,
Info = 3,
Debug = 4,
}
impl LogLevel {
fn from_str(s: &str) -> Self {
match s.to_lowercase().as_str() {
"error" => LogLevel::Error,
"warn" => LogLevel::Warn,
"info" => LogLevel::Info,
"debug" => LogLevel::Debug,
_ => LogLevel::Info,
}
}
}
#[macro_export]
macro_rules! log_error {
($($arg:tt)*) => {{
let level = $crate::LOG_LEVEL.get().copied().unwrap_or($crate::LogLevel::Info);
if level >= $crate::LogLevel::Error {
let ts = $crate::format_timestamp();
eprintln!("[{}] [ERROR] {}", ts, format!($($arg)*));
}
}};
}
#[macro_export]
macro_rules! log_warn {
($($arg:tt)*) => {{
let level = $crate::LOG_LEVEL.get().copied().unwrap_or($crate::LogLevel::Info);
if level >= $crate::LogLevel::Warn {
let ts = $crate::format_timestamp();
eprintln!("[{}] [WARN] {}", ts, format!($($arg)*));
}
}};
}
#[macro_export]
macro_rules! log_info {
($($arg:tt)*) => {{
let level = $crate::LOG_LEVEL.get().copied().unwrap_or($crate::LogLevel::Info);
if level >= $crate::LogLevel::Info {
let ts = $crate::format_timestamp();
eprintln!("[{}] [INFO] {}", ts, format!($($arg)*));
}
}};
}
#[macro_export]
macro_rules! log_debug {
($($arg:tt)*) => {{
let level = $crate::LOG_LEVEL.get().copied().unwrap_or($crate::LogLevel::Info);
if level >= $crate::LogLevel::Debug {
let ts = $crate::format_timestamp();
eprintln!("[{}] [DEBUG] {}", ts, format!($($arg)*));
}
}};
}
/// Format current timestamp as "YYYY-MM-DD HH:MM:SS.mmm"
pub fn format_timestamp() -> String {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default();
let secs = now.as_secs();
let millis = now.subsec_millis();
// Calculate date/time components
const SECS_PER_DAY: u64 = 86400;
const SECS_PER_HOUR: u64 = 3600;
const SECS_PER_MIN: u64 = 60;
let days_since_epoch = secs / SECS_PER_DAY;
let secs_today = secs % SECS_PER_DAY;
let hour = (secs_today / SECS_PER_HOUR) as u8;
let min = ((secs_today % SECS_PER_HOUR) / SECS_PER_MIN) as u8;
let sec = (secs_today % SECS_PER_MIN) as u8;
// Simple epoch to date conversion (good enough for logging)
// Days since 1970-01-01
let mut year = 1970;
let mut days_left = days_since_epoch;
loop {
let days_in_year = if is_leap_year(year) { 366 } else { 365 };
if days_left >= days_in_year {
days_left -= days_in_year;
year += 1;
} else {
break;
}
}
let (month, day) = days_to_month_day(days_left as u16, is_leap_year(year));
format!(
"{:04}-{:02}-{:02} {:02}:{:02}:{:02}.{:03}",
year, month, day, hour, min, sec, millis
)
}
fn is_leap_year(year: i32) -> bool {
(year % 4 == 0 && year % 100 != 0) || (year % 400 == 0)
}
fn days_to_month_day(days: u16, is_leap: bool) -> (u8, u8) {
let days_in_month = if is_leap {
[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 days_left = days;
for (month_idx, &days_in_month) in days_in_month.iter().enumerate() {
if days_left < days_in_month as u16 {
return ((month_idx + 1) as u8, (days_left + 1) as u8);
}
days_left -= days_in_month as u16;
}
(12, 31) // Fallback
}
fn init_logger() { fn init_logger() {
// Use LOG_LEVEL env var (fall back to RUST_LOG for backwards compatibility) // Use LOG_LEVEL env var (fall back to RUST_LOG for backwards compatibility)
let level_str = env::var("LOG_LEVEL") let filter = env::var("LOG_LEVEL")
.or_else(|_| env::var("RUST_LOG")) .or_else(|_| env::var("RUST_LOG"))
.unwrap_or_else(|_| "info".to_string()); .unwrap_or_else(|_| "info".to_string());
let level = LogLevel::from_str(&level_str);
LOG_LEVEL.set(level).ok(); tracing_subscriber::fmt()
.with_env_filter(EnvFilter::new(&filter))
.with_target(false)
.init();
} }
/// Convert HTTP(S) URL to WebSocket URL /// Convert HTTP(S) URL to WebSocket URL
@ -196,7 +67,7 @@ async fn main() {
let args = Args::parse(); let args = Args::parse();
log_info!("Towerops agent starting"); tracing::info!("Towerops agent starting");
// Check for newer Docker image version // Check for newer Docker image version
version::check_for_updates(); version::check_for_updates();
@ -215,7 +86,7 @@ async fn main() {
// Log received traps // Log received traps
while let Some(trap) = trap_rx.recv().await { while let Some(trap) = trap_rx.recv().await {
log_info!("{}", trap); tracing::info!("{}", trap);
} }
}); });
} }
@ -223,7 +94,7 @@ async fn main() {
// Convert HTTP(S) URL to WebSocket URL // Convert HTTP(S) URL to WebSocket URL
let ws_url = convert_to_websocket_url(&args.api_url); let ws_url = convert_to_websocket_url(&args.api_url);
log_info!("WebSocket URL: {}", ws_url); tracing::info!("WebSocket URL: {}", ws_url);
// Shared connection state // Shared connection state
// Starts as false (not connected), updated when WebSocket connects/disconnects // Starts as false (not connected), updated when WebSocket connects/disconnects
@ -235,7 +106,7 @@ async fn main() {
// Spawn signal handler for graceful shutdown // Spawn signal handler for graceful shutdown
tokio::spawn(async move { tokio::spawn(async move {
wait_for_shutdown_signal().await; wait_for_shutdown_signal().await;
log_info!("Shutdown signal received, initiating graceful shutdown..."); tracing::info!("Shutdown signal received, initiating graceful shutdown...");
let _ = shutdown_tx.send(true); let _ = shutdown_tx.send(true);
}); });
@ -247,14 +118,14 @@ async fn main() {
loop { loop {
// Check if shutdown was requested // Check if shutdown was requested
if *shutdown_rx.borrow() { if *shutdown_rx.borrow() {
log_info!("Shutdown requested, exiting main loop"); tracing::info!("Shutdown requested, exiting main loop");
break; break;
} }
attempt += 1; attempt += 1;
if attempt > 1 { if attempt > 1 {
log_info!( tracing::info!(
"Retry attempt {} - waiting {} seconds before reconnecting", "Retry attempt {} - waiting {} seconds before reconnecting",
attempt, attempt,
retry_delay.as_secs() retry_delay.as_secs()
@ -268,7 +139,7 @@ async fn main() {
// Connect to Towerops server via WebSocket // Connect to Towerops server via WebSocket
let mut client = match AgentClient::connect(&ws_url, &args.token).await { let mut client = match AgentClient::connect(&ws_url, &args.token).await {
Ok(client) => { Ok(client) => {
log_info!("Successfully connected to server"); tracing::info!("Successfully connected to server");
// Mark as connected for health check // Mark as connected for health check
connected.store(true, Ordering::Relaxed); connected.store(true, Ordering::Relaxed);
// Reset retry delay on successful connection // Reset retry delay on successful connection
@ -277,7 +148,7 @@ async fn main() {
client client
} }
Err(e) => { Err(e) => {
log_error!("Failed to connect to server: {}", e); tracing::error!("Failed to connect to server: {}", e);
// Mark as disconnected for health check // Mark as disconnected for health check
connected.store(false, Ordering::Relaxed); connected.store(false, Ordering::Relaxed);
continue; continue;
@ -289,12 +160,12 @@ async fn main() {
Ok(()) => { Ok(()) => {
// Clean shutdown requested // Clean shutdown requested
if *shutdown_rx.borrow() { if *shutdown_rx.borrow() {
log_info!("Agent shutdown complete"); tracing::info!("Agent shutdown complete");
break; break;
} }
} }
Err(e) => { Err(e) => {
log_error!("Agent disconnected: {}", e); tracing::error!("Agent disconnected: {}", e);
} }
} }
@ -303,7 +174,7 @@ async fn main() {
// Loop will retry with backoff (unless shutdown was requested) // Loop will retry with backoff (unless shutdown was requested)
} }
log_info!("Towerops agent stopped"); tracing::info!("Towerops agent stopped");
} }
/// Wait for SIGTERM or SIGINT shutdown signal. /// Wait for SIGTERM or SIGINT shutdown signal.
@ -319,10 +190,10 @@ async fn wait_for_shutdown_signal() {
tokio::select! { tokio::select! {
_ = sigterm.recv() => { _ = sigterm.recv() => {
log_info!("Received SIGTERM"); tracing::info!("Received SIGTERM");
} }
_ = sigint.recv() => { _ = sigint.recv() => {
log_info!("Received SIGINT"); tracing::info!("Received SIGINT");
} }
} }
} }
@ -333,7 +204,7 @@ async fn wait_for_shutdown_signal() {
tokio::signal::ctrl_c() tokio::signal::ctrl_c()
.await .await
.expect("Failed to register Ctrl+C handler"); .expect("Failed to register Ctrl+C handler");
log_info!("Received Ctrl+C"); tracing::info!("Received Ctrl+C");
} }
} }
@ -380,158 +251,4 @@ mod tests {
"wss://localhost:4000" "wss://localhost:4000"
); );
} }
#[test]
fn test_format_timestamp() {
let timestamp = format_timestamp();
// Should be in format "YYYY-MM-DD HH:MM:SS.mmm"
assert!(timestamp.len() >= 23); // Minimum length
assert!(timestamp.contains('-'));
assert!(timestamp.contains(':'));
assert!(timestamp.contains('.'));
}
#[test]
fn test_is_leap_year() {
assert!(is_leap_year(2000)); // Divisible by 400
assert!(is_leap_year(2024)); // Divisible by 4, not by 100
assert!(!is_leap_year(1900)); // Divisible by 100, not by 400
assert!(!is_leap_year(2023)); // Not divisible by 4
}
#[test]
fn test_days_to_month_day() {
// January 1st (day 0)
assert_eq!(days_to_month_day(0, false), (1, 1));
// January 31st (day 30)
assert_eq!(days_to_month_day(30, false), (1, 31));
// February 1st (day 31)
assert_eq!(days_to_month_day(31, false), (2, 1));
// March 1st in non-leap year (day 59)
assert_eq!(days_to_month_day(59, false), (3, 1));
// March 1st in leap year (day 60)
assert_eq!(days_to_month_day(60, true), (3, 1));
// December 31st in non-leap year (day 364)
assert_eq!(days_to_month_day(364, false), (12, 31));
}
#[test]
fn test_log_level_from_str_error() {
assert_eq!(LogLevel::from_str("error"), LogLevel::Error);
assert_eq!(LogLevel::from_str("ERROR"), LogLevel::Error);
assert_eq!(LogLevel::from_str("Error"), LogLevel::Error);
}
#[test]
fn test_log_level_from_str_warn() {
assert_eq!(LogLevel::from_str("warn"), LogLevel::Warn);
assert_eq!(LogLevel::from_str("WARN"), LogLevel::Warn);
assert_eq!(LogLevel::from_str("Warn"), LogLevel::Warn);
}
#[test]
fn test_log_level_from_str_info() {
assert_eq!(LogLevel::from_str("info"), LogLevel::Info);
assert_eq!(LogLevel::from_str("INFO"), LogLevel::Info);
assert_eq!(LogLevel::from_str("Info"), LogLevel::Info);
}
#[test]
fn test_log_level_from_str_debug() {
assert_eq!(LogLevel::from_str("debug"), LogLevel::Debug);
assert_eq!(LogLevel::from_str("DEBUG"), LogLevel::Debug);
assert_eq!(LogLevel::from_str("Debug"), LogLevel::Debug);
}
#[test]
fn test_log_level_from_str_unknown() {
// Unknown values default to Info
assert_eq!(LogLevel::from_str("unknown"), LogLevel::Info);
assert_eq!(LogLevel::from_str("trace"), LogLevel::Info);
assert_eq!(LogLevel::from_str(""), LogLevel::Info);
}
#[test]
fn test_log_level_ordering() {
assert!(LogLevel::Error < LogLevel::Warn);
assert!(LogLevel::Warn < LogLevel::Info);
assert!(LogLevel::Info < LogLevel::Debug);
}
#[test]
fn test_log_level_eq() {
assert_eq!(LogLevel::Error, LogLevel::Error);
assert_eq!(LogLevel::Warn, LogLevel::Warn);
assert_eq!(LogLevel::Info, LogLevel::Info);
assert_eq!(LogLevel::Debug, LogLevel::Debug);
}
#[test]
fn test_log_level_copy() {
let level = LogLevel::Debug;
let copied = level;
assert_eq!(copied, LogLevel::Debug);
}
#[test]
fn test_log_level_clone() {
let level = LogLevel::Warn;
let cloned = level.clone();
assert_eq!(cloned, LogLevel::Warn);
}
#[test]
fn test_days_to_month_day_february() {
// February 28th in non-leap year (day 58)
assert_eq!(days_to_month_day(58, false), (2, 28));
// February 29th in leap year (day 59)
assert_eq!(days_to_month_day(59, true), (2, 29));
}
#[test]
fn test_days_to_month_day_later_months() {
// April 15th in non-leap year (day 105)
// Jan=31, Feb=28, Mar=31, Apr 1-15 = 31+28+31+14 = 104 (0-indexed day 104)
assert_eq!(days_to_month_day(104, false), (4, 15));
// July 4th in non-leap year
// Jan=31, Feb=28, Mar=31, Apr=30, May=31, Jun=30, Jul 1-4 = 31+28+31+30+31+30+3 = 184 (day 184)
assert_eq!(days_to_month_day(184, false), (7, 4));
}
#[test]
fn test_days_to_month_day_end_of_year() {
// December 31st in leap year (day 365)
assert_eq!(days_to_month_day(365, true), (12, 31));
// December 25th in non-leap year
// 31+28+31+30+31+30+31+31+30+31+30+24 = 358 (day 358)
assert_eq!(days_to_month_day(358, false), (12, 25));
}
#[test]
fn test_days_to_month_day_overflow_fallback() {
// Day 400 is beyond year - should fallback to Dec 31
assert_eq!(days_to_month_day(400, false), (12, 31));
}
#[test]
fn test_format_timestamp_structure() {
let timestamp = format_timestamp();
// Should be in format "YYYY-MM-DD HH:MM:SS.mmm"
let parts: Vec<&str> = timestamp.split(' ').collect();
assert_eq!(parts.len(), 2, "Expected date and time parts");
let date_parts: Vec<&str> = parts[0].split('-').collect();
assert_eq!(date_parts.len(), 3, "Expected year-month-day");
let time_parts: Vec<&str> = parts[1].split(':').collect();
assert_eq!(time_parts.len(), 3, "Expected hour:min:sec.ms");
}
} }

10
src/snmp/trap.rs
View file

@ -137,11 +137,11 @@ impl TrapListener {
let socket = match UdpSocket::bind(&bind_addr).await { let socket = match UdpSocket::bind(&bind_addr).await {
Ok(s) => { Ok(s) => {
crate::log_info!("SNMP trap listener started on UDP port {}", self.port); tracing::info!("SNMP trap listener started on UDP port {}", self.port);
s s
} }
Err(e) => { Err(e) => {
crate::log_error!("Failed to bind trap listener to {}: {}", bind_addr, e); tracing::error!("Failed to bind trap listener to {}: {}", bind_addr, e);
return; return;
} }
}; };
@ -156,17 +156,17 @@ impl TrapListener {
match parse_trap(packet, src_addr) { match parse_trap(packet, src_addr) {
Ok(trap) => { Ok(trap) => {
if trap_tx.send(trap).await.is_err() { if trap_tx.send(trap).await.is_err() {
crate::log_warn!("Trap channel closed, stopping listener"); tracing::warn!("Trap channel closed, stopping listener");
break; break;
} }
} }
Err(e) => { Err(e) => {
crate::log_warn!("Failed to parse SNMP trap from {}: {}", src_addr, e); tracing::warn!("Failed to parse SNMP trap from {}: {}", src_addr, e);
} }
} }
} }
Err(e) => { Err(e) => {
crate::log_warn!("Error receiving trap packet: {}", e); tracing::warn!("Error receiving trap packet: {}", e);
} }
} }
} }

4
src/version.rs
View file

@ -6,8 +6,8 @@ pub fn current_version() -> &'static str {
/// Startup check - logs current version /// Startup check - logs current version
pub fn check_for_updates() { pub fn check_for_updates() {
let current_ver = current_version(); let current_ver = current_version();
crate::log_info!("Current version: {}", current_ver); tracing::info!("Current version: {}", current_ver);
crate::log_info!("Watchtower will automatically update to new versions"); tracing::info!("Watchtower will automatically update to new versions");
} }
#[cfg(test)] #[cfg(test)]

52
src/websocket_client.rs
View file

@ -57,7 +57,7 @@ impl AgentClient {
// Strip trailing slash from base URL to avoid double slashes // Strip trailing slash from base URL to avoid double slashes
let base_url = url.trim_end_matches('/'); let base_url = url.trim_end_matches('/');
let ws_url = format!("{}/socket/agent/websocket", base_url); let ws_url = format!("{}/socket/agent/websocket", base_url);
crate::log_info!( tracing::info!(
"Connecting to WebSocket: {} (timeout: {}s)", "Connecting to WebSocket: {} (timeout: {}s)",
ws_url, ws_url,
CONNECTION_TIMEOUT.as_secs() CONNECTION_TIMEOUT.as_secs()
@ -67,11 +67,11 @@ impl AgentClient {
let (mut ws_stream, _) = match timeout(CONNECTION_TIMEOUT, connect_async(&ws_url)).await { let (mut ws_stream, _) = match timeout(CONNECTION_TIMEOUT, connect_async(&ws_url)).await {
Ok(Ok(result)) => result, Ok(Ok(result)) => result,
Ok(Err(e)) => { Ok(Err(e)) => {
crate::log_error!("WebSocket connection failed: {}", e); tracing::error!("WebSocket connection failed: {}", e);
return Err(format!("Failed to connect to WebSocket: {}", e).into()); return Err(format!("Failed to connect to WebSocket: {}", e).into());
} }
Err(_) => { Err(_) => {
crate::log_error!( tracing::error!(
"WebSocket connection timed out after {}s", "WebSocket connection timed out after {}s",
CONNECTION_TIMEOUT.as_secs() CONNECTION_TIMEOUT.as_secs()
); );
@ -83,7 +83,7 @@ impl AgentClient {
} }
}; };
crate::log_info!("Connected to Towerops server at {}", url); tracing::info!("Connected to Towerops server at {}", url);
let agent_id = generate_agent_id(); let agent_id = generate_agent_id();
let (result_tx, result_rx) = mpsc::unbounded_channel(); let (result_tx, result_rx) = mpsc::unbounded_channel();
@ -98,7 +98,7 @@ impl AgentClient {
let join_text = serde_json::to_string(&join_msg)?; let join_text = serde_json::to_string(&join_msg)?;
ws_stream.send(WsMessage::Text(join_text)).await?; ws_stream.send(WsMessage::Text(join_text)).await?;
crate::log_info!( tracing::info!(
"Sent channel join request with token for agent:{}", "Sent channel join request with token for agent:{}",
agent_id agent_id
); );
@ -127,10 +127,10 @@ impl AgentClient {
// Check for shutdown signal (highest priority) // Check for shutdown signal (highest priority)
_ = shutdown_rx.changed() => { _ = shutdown_rx.changed() => {
if *shutdown_rx.borrow() { if *shutdown_rx.borrow() {
crate::log_info!("Shutdown signal received, closing WebSocket connection gracefully"); tracing::info!("Shutdown signal received, closing WebSocket connection gracefully");
// Send close frame to server // Send close frame to server
if let Err(e) = self.ws_stream.close(None).await { if let Err(e) = self.ws_stream.close(None).await {
crate::log_warn!("Error sending WebSocket close frame: {}", e); tracing::warn!("Error sending WebSocket close frame: {}", e);
} }
break Ok(()); break Ok(());
} }
@ -141,24 +141,24 @@ impl AgentClient {
match msg { match msg {
Some(Ok(WsMessage::Binary(data))) => { Some(Ok(WsMessage::Binary(data))) => {
if let Err(e) = self.handle_message(&data).await { if let Err(e) = self.handle_message(&data).await {
crate::log_error!("Error handling binary message: {}", e); tracing::error!("Error handling binary message: {}", e);
} }
} }
Some(Ok(WsMessage::Text(text))) => { Some(Ok(WsMessage::Text(text))) => {
if let Err(e) = self.handle_text_message(&text).await { if let Err(e) = self.handle_text_message(&text).await {
crate::log_error!("Error handling text message: {}", e); tracing::error!("Error handling text message: {}", e);
} }
} }
Some(Ok(WsMessage::Close(_))) => { Some(Ok(WsMessage::Close(_))) => {
crate::log_info!("Server closed connection"); tracing::info!("Server closed connection");
break Ok(()); break Ok(());
} }
Some(Err(e)) => { Some(Err(e)) => {
crate::log_error!("WebSocket error: {}", e); tracing::error!("WebSocket error: {}", e);
break Err(e.into()); break Err(e.into());
} }
None => { None => {
crate::log_info!("Connection closed"); tracing::info!("Connection closed");
break Ok(()); break Ok(());
} }
_ => {} _ => {}
@ -168,14 +168,14 @@ impl AgentClient {
// Receive SNMP results from job tasks // Receive SNMP results from job tasks
Some(snmp_result) = self.result_rx.recv() => { Some(snmp_result) = self.result_rx.recv() => {
if let Err(e) = self.send_result(snmp_result).await { if let Err(e) = self.send_result(snmp_result).await {
crate::log_error!("Error sending SNMP result: {}", e); tracing::error!("Error sending SNMP result: {}", e);
} }
} }
// Send periodic heartbeats // Send periodic heartbeats
_ = heartbeat_interval.tick() => { _ = heartbeat_interval.tick() => {
if let Err(e) = self.send_heartbeat().await { if let Err(e) = self.send_heartbeat().await {
crate::log_error!("Error sending heartbeat: {}", e); tracing::error!("Error sending heartbeat: {}", e);
} }
} }
} }
@ -188,7 +188,7 @@ impl AgentClient {
match phoenix_msg.event.as_str() { match phoenix_msg.event.as_str() {
"phx_reply" => { "phx_reply" => {
crate::log_info!("Channel join reply: {:?}", phoenix_msg.payload); tracing::info!("Channel join reply: {:?}", phoenix_msg.payload);
} }
"jobs" => { "jobs" => {
// Extract binary protobuf from payload // Extract binary protobuf from payload
@ -201,7 +201,7 @@ impl AgentClient {
} }
} }
_ => { _ => {
crate::log_debug!("Ignoring unknown event: {}", phoenix_msg.event); tracing::debug!("Ignoring unknown event: {}", phoenix_msg.event);
} }
} }
@ -224,18 +224,18 @@ impl AgentClient {
/// No long-running tasks are spawned - the agent is stateless. /// No long-running tasks are spawned - the agent is stateless.
/// Server handles all scheduling and retries via Oban. /// Server handles all scheduling and retries via Oban.
async fn handle_jobs(&self, job_list: AgentJobList) -> Result<()> { async fn handle_jobs(&self, job_list: AgentJobList) -> Result<()> {
crate::log_info!("Received {} jobs from server", job_list.jobs.len()); tracing::info!("Received {} jobs from server", job_list.jobs.len());
for job in job_list.jobs { for job in job_list.jobs {
let job_type = JobType::try_from(job.job_type).unwrap_or(JobType::Poll); let job_type = JobType::try_from(job.job_type).unwrap_or(JobType::Poll);
crate::log_info!("Executing job: {} (type: {:?})", job.job_id, job_type); tracing::info!("Executing job: {} (type: {:?})", job.job_id, job_type);
// Spawn task to execute SNMP job (one-off execution) // Spawn task to execute SNMP job (one-off execution)
let result_tx = self.result_tx.clone(); let result_tx = self.result_tx.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = execute_job(job, result_tx).await { if let Err(e) = execute_job(job, result_tx).await {
crate::log_error!("Job execution failed: {}", e); tracing::error!("Job execution failed: {}", e);
} }
}); });
} }
@ -265,7 +265,7 @@ impl AgentClient {
let text = serde_json::to_string(&msg)?; let text = serde_json::to_string(&msg)?;
self.ws_stream.send(WsMessage::Text(text)).await?; self.ws_stream.send(WsMessage::Text(text)).await?;
crate::log_debug!("Sent heartbeat"); tracing::debug!("Sent heartbeat");
Ok(()) Ok(())
} }
@ -283,7 +283,7 @@ impl AgentClient {
let text = serde_json::to_string(&msg)?; let text = serde_json::to_string(&msg)?;
self.ws_stream.send(WsMessage::Text(text)).await?; self.ws_stream.send(WsMessage::Text(text)).await?;
crate::log_debug!("Sent SNMP result for device {}", result.device_id); tracing::debug!("Sent SNMP result for device {}", result.device_id);
Ok(()) Ok(())
} }
} }
@ -299,7 +299,7 @@ async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender<SnmpResult>
"***".to_string() "***".to_string()
}; };
crate::log_info!( tracing::info!(
"Executing SNMP job for device {} at {}:{} (community: {}, version: {})", "Executing SNMP job for device {} at {}:{} (community: {}, version: {})",
job.device_id, job.device_id,
snmp_device.ip, snmp_device.ip,
@ -332,7 +332,7 @@ async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender<SnmpResult>
oid_values.insert(oid.clone(), value_to_string(value)); oid_values.insert(oid.clone(), value_to_string(value));
} }
Err(e) => { Err(e) => {
crate::log_warn!( tracing::warn!(
"SNMP GET failed for device {} at {}:{} (version: {}, community: {}), OID {}: {}", "SNMP GET failed for device {} at {}:{} (version: {}, community: {}), OID {}: {}",
job.device_id, job.device_id,
snmp_device.ip, snmp_device.ip,
@ -365,7 +365,7 @@ async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender<SnmpResult>
} }
} }
Err(e) => { Err(e) => {
crate::log_warn!( tracing::warn!(
"SNMP WALK failed for device {} at {}:{} (version: {}, community: {}), OID {}: {}", "SNMP WALK failed for device {} at {}:{} (version: {}, community: {}), OID {}: {}",
job.device_id, job.device_id,
snmp_device.ip, snmp_device.ip,
@ -392,7 +392,7 @@ async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender<SnmpResult>
.as_secs() as i64, .as_secs() as i64,
}; };
crate::log_info!( tracing::info!(
"Collected {} OID values for job {}", "Collected {} OID values for job {}",
result.oid_values.len(), result.oid_values.len(),
job.job_id job.job_id
@ -400,7 +400,7 @@ async fn execute_job(job: AgentJob, result_tx: mpsc::UnboundedSender<SnmpResult>
// Send result back to main client task // Send result back to main client task
if let Err(e) = result_tx.send(result) { if let Err(e) = result_tx.send(result) {
crate::log_warn!( tracing::warn!(
"Failed to send SNMP result for job {}: channel closed (connection may have dropped)", "Failed to send SNMP result for job {}: channel closed (connection may have dropped)",
job.job_id job.job_id
); );