From f926e08980aee01edf6fbbd30db30897f0f64610 Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Fri, 4 Sep 2026 13:23:05 +0200 Subject: [PATCH 1/6] =?UTF-8?q?fix(rust-plugins):=20SNMP=20protocol=20hard?= =?UTF-8?q?ening=20=E2=80=94=20validate=20agent=20responses,=20bound=20wal?= =?UTF-8?q?ks,=20timeouts=20&=20retries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine blindly trusted the agent it queried, which is precisely the untrusted party of the dialogue: - Agent errors are now detected: a response with error-status != 0 produces a typed error naming the RFC 3416 status (e.g. "noSuchName"). - Request/response correlation: every request carries a fresh request-id, and received datagrams are validated (request-id, community echo, PDU type) before being accepted — unrelated datagrams (e.g. a late retransmission of a previous request) are discarded instead of being consumed as the current answer. - Bounded walks: OIDs must be strictly increasing during a walk (the classic snmpwalk "OID not increasing" guard) and a single walk may collect at most 100 000 values — both protect against a buggy or malicious agent looping the walk or streaming endless data. - Configurable timeouts and retries via three new CLI options: --timeout (per-attempt receive timeout, default 1s), --snmp-retries (default 2), and --collect-timeout (global budget for the whole collection, default 50s, so the plugin exits with a clean UNKNOWN before centengine's own kill timeout). - The UDP receive buffer was 1024 bytes, silently truncating large bulk responses; raised to 65535. Connection parameters (target, community, timeouts, retries) are now threaded through a single SnmpConfig instead of loose string arguments, and the two walk loops share one implementation. Test plan: - cargo build --release succeeds - cargo test: 75 passed, 0 failed (10 new: RFC 3416 names, response validation, non-increasing OID, varbind cap, expired deadline) - Manual check against an unroutable address: default settings exit UNKNOWN after retries*timeout with the attempt count in the message; --collect-timeout correctly preempts a longer per-attempt timeout --- rust-plugins/src/generic/error.rs | 36 ++ rust-plugins/src/generic/mod.rs | 25 +- rust-plugins/src/main.rs | 37 +- rust-plugins/src/snmp/mod.rs | 587 +++++++++++++++++++++++------- 4 files changed, 532 insertions(+), 153 deletions(-) diff --git a/rust-plugins/src/generic/error.rs b/rust-plugins/src/generic/error.rs index 38f9f1766c..d5da270b6d 100644 --- a/rust-plugins/src/generic/error.rs +++ b/rust-plugins/src/generic/error.rs @@ -59,6 +59,42 @@ pub enum Error { ))] FailedToConnectToHost { url: String, os: String }, + #[snafu(display( + "SNMP agent returned an error: {} (status {}, index {})", + name, + status, + index + ))] + SnmpAgentError { + name: &'static str, + status: u32, + index: u32, + }, + + #[snafu(display("SNMP agent is misbehaving: OID {} is not increasing during walk", oid))] + OidNotIncreasing { oid: String }, + + #[snafu(display( + "SNMP walk aborted: the agent returned more than {} values for a single subtree", + max + ))] + WalkTooLarge { max: usize }, + + #[snafu(display("SNMP collection exceeded the global timeout of {}s", seconds))] + CollectTimeout { seconds: u64 }, + + #[snafu(display( + "No valid SNMP response from {} after {} attempts (timeout {}s per attempt)", + url, + attempts, + timeout + ))] + RequestTimeout { + url: String, + attempts: u32, + timeout: u64, + }, + #[snafu(transparent)] Io { source: io::Error }, #[snafu(transparent)] diff --git a/rust-plugins/src/generic/mod.rs b/rust-plugins/src/generic/mod.rs index 801724542c..db0bb5a475 100644 --- a/rust-plugins/src/generic/mod.rs +++ b/rust-plugins/src/generic/mod.rs @@ -15,7 +15,7 @@ use self::error::Result; use crate::compute::{Compute, Parser, ast::ExprResult, threshold::Threshold}; use crate::output::{Output, OutputFormatter}; use crate::snmp::SnmpResult; -use crate::snmp::{snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels}; +use crate::snmp::{snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels, SnmpConfig}; use log::{debug, trace}; use regex::Regex; use serde::Deserialize; @@ -270,12 +270,13 @@ impl Command { /// Executes all configured SNMP queries (Get and Walk operations) and returns the results. fn execute_snmp_collect( &self, - target: &str, - version: &str, - community: &str, + config: &SnmpConfig, check_format: bool, ) -> Result> { let mut collect: Vec = Vec::new(); + // Single deadline for ALL queries of this collection: the global + // time budget covers the sum of the walks and gets, not each one. + let deadline = config.deadline(); if check_format { // In check-format mode, don't make SNMP requests and initialize with dummy values. @@ -304,13 +305,13 @@ impl Command { QueryType::Walk => { if let Some(lab) = &s.labels { let r = snmp_bulk_walk_with_labels( - target, version, community, &s.oid, &s.name, &lab, + config, deadline, &s.oid, &s.name, lab, )?; if !r.items.is_empty() { collect.push(r); } } else { - let r = snmp_bulk_walk(target, version, community, &s.oid, &s.name)?; + let r = snmp_bulk_walk(config, deadline, &s.oid, &s.name)?; if !r.items.is_empty() { collect.push(r); } @@ -324,7 +325,7 @@ impl Command { } if !to_get.is_empty() { - let r = snmp_bulk_get(target, version, community, 1, 1, &to_get, &get_name); + let r = snmp_bulk_get(config, deadline, 1, 1, &to_get, &get_name); collect.push(r?); } if collect.is_empty() { @@ -336,9 +337,7 @@ impl Command { /// Executes the complete plugin pipeline: SNMP collection, metric computation, filtering, and output formatting. /// /// # Arguments - /// * `target` - The target address in "host:port" format - /// * `version` - SNMP version string (e.g., "2c") - /// * `community` - SNMP community string + /// * `config` - SNMP connection parameters (target, community, timeouts, retries) /// * `filter_in` - Regex patterns; metrics matching any pattern are kept (empty = keep all) /// * `filter_out` - Regex patterns; metrics matching any pattern are excluded /// * `check_format` - Dry-run mode ( validate macros ) @@ -349,16 +348,14 @@ impl Command { /// A [`CmdResult`] containing the overall [`Status`] and Nagios-compatible output string. pub fn execute( &self, - target: &str, - version: &str, - community: &str, + config: &SnmpConfig, filter_in: &Vec, filter_out: &Vec, check_format: bool, check_response: bool, no_data_status: Status, ) -> Result { - let mut collect = self.execute_snmp_collect(target, version, community, check_format)?; + let mut collect = self.execute_snmp_collect(config, check_format)?; if check_response { return self.format_raw_response(&collect); diff --git a/rust-plugins/src/main.rs b/rust-plugins/src/main.rs index 69105d0a6d..030bf2f77e 100644 --- a/rust-plugins/src/main.rs +++ b/rust-plugins/src/main.rs @@ -33,6 +33,7 @@ use generic::error::*; use lalrpop_util::lalrpop_mod; use lexopt::Arg; use log::trace; +use snmp::SnmpConfig; use std::fs; lalrpop_mod!(grammar); @@ -80,6 +81,14 @@ fn snmp_plugin() -> Result<(), Error> { let mut port = 161; let mut snmp_version = "2c".to_string(); let mut snmp_community = "public".to_string(); + // Mirrors of the Perl plugin options: --timeout (per-request receive + // timeout, default 1s) and --snmp-retries (default 2, audit P9 + // recommendation rather than the Perl default of 5). + let mut timeout_secs: u64 = 1; + let mut snmp_retries: u32 = 2; + // Global budget for the whole collection: exits with a clean UNKNOWN + // before centengine (60s default) kills the process. + let mut collect_timeout_secs: u64 = 50; let mut filter_in = Vec::new(); let mut filter_out = Vec::new(); let mut no_data_status = Status::Unknown; @@ -152,6 +161,9 @@ fn snmp_plugin() -> Result<(), Error> { println!(" -i, --filter-in Include filter (can be used multiple times)"); println!(" -o, --filter-out Exclude filter (can be used multiple times)"); println!(" --no-data-status Status when the filters keep no data: OK, WARNING, CRITICAL or UNKNOWN (default: UNKNOWN)"); + println!(" --timeout Timeout per SNMP request attempt (default: 1)"); + println!(" --snmp-retries Retries after a timed-out attempt (default: 2)"); + println!(" --collect-timeout Global time budget for the whole collection (default: 50)"); println!(" --warning- Warning threshold for metric"); println!(" --critical- Critical threshold for metric"); println!(" --check-format Check JSON file validity and exit"); @@ -160,6 +172,18 @@ fn snmp_plugin() -> Result<(), Error> { println!(" -h, --help Print this help message"); std::process::exit(0); } + Long("timeout") => { + timeout_secs = parser.value()?.parse::()?; + trace!("timeout: {}s", timeout_secs); + } + Long("snmp-retries") => { + snmp_retries = parser.value()?.parse::()?; + trace!("snmp_retries: {}", snmp_retries); + } + Long("collect-timeout") => { + collect_timeout_secs = parser.value()?.parse::()?; + trace!("collect_timeout: {}s", collect_timeout_secs); + } Long("check-format") => { check_format = true; } @@ -255,12 +279,17 @@ fn snmp_plugin() -> Result<(), Error> { std::process::exit(0); } - let url = format!("{}:{}", hostname, port); + let snmp_config = SnmpConfig { + target: format!("{}:{}", hostname, port), + version: snmp_version, + community: snmp_community, + timeout: std::time::Duration::from_secs(timeout_secs), + retries: snmp_retries, + collect_timeout: std::time::Duration::from_secs(collect_timeout_secs), + }; let result = cmd.execute( - &url, - &snmp_version, - &snmp_community, + &snmp_config, &filter_in, &filter_out, check_format, diff --git a/rust-plugins/src/snmp/mod.rs b/rust-plugins/src/snmp/mod.rs index 44873cabef..7444f4ca43 100644 --- a/rust-plugins/src/snmp/mod.rs +++ b/rust-plugins/src/snmp/mod.rs @@ -16,14 +16,18 @@ extern crate rasn_snmp; use crate::Error::InvalidOidParser; use crate::compute::ast::ExprResult; +use crate::generic::error::Error::CollectTimeout; use crate::generic::error::Error::EmptyResponse; use crate::generic::error::Error::FailedToConnectToHost; use crate::generic::error::Error::InvalidSnmpPduDecode; use crate::generic::error::Error::InvalidSnmpPduEncode; use crate::generic::error::Error::InvalidSnmpType; use crate::generic::error::Error::InvalidSnmpValue; +use crate::generic::error::Error::OidNotIncreasing; +use crate::generic::error::Error::RequestTimeout; +use crate::generic::error::Error::SnmpAgentError; +use crate::generic::error::Error::WalkTooLarge; use crate::generic::error::Result; -use log::info; use log::{trace, warn}; use rasn::types::ObjectIdentifier; use rasn_smi::v2::{ApplicationSyntax, ObjectSyntax, SimpleSyntax}; @@ -37,6 +41,120 @@ use rasn_snmp::v3::VarBindValue::EndOfMibView; use std::collections::HashMap; use std::convert::TryInto; use std::net::UdpSocket; +use std::time::{Duration, Instant}; + +/// Maximum size of a UDP datagram; SNMP bulk responses can be large, +/// a smaller buffer would silently truncate them and break BER decoding. +const UDP_BUFFER_SIZE: usize = 65535; + +/// Upper bound on the number of values a single walk may collect. Protects +/// the plugin against a buggy or malicious agent that returns endless data: +/// well beyond any real table (a 48-port switch's ifTable is a few thousand +/// entries), but finite. +const MAX_WALK_VARBINDS: usize = 100_000; + +/// Connection parameters shared by every SNMP request of a collection. +#[derive(Debug, Clone)] +pub struct SnmpConfig { + /// Target address in `host:port` format. + pub target: String, + /// SNMP version (only `2c` is supported today). + pub version: String, + /// SNMP community string. + pub community: String, + /// Receive timeout for a single request attempt (mirror of the Perl + /// `--timeout`, default 1s). + pub timeout: Duration, + /// Number of retries after a timed-out attempt (default 2; total + /// attempts = retries + 1). + pub retries: u32, + /// Global time budget for the whole collection (all gets and walks). + /// Protects the poller from a slow agent: centengine kills plugins + /// after its own timeout, this one lets us exit with a clean UNKNOWN + /// message before being killed. + pub collect_timeout: Duration, +} + +impl SnmpConfig { + /// Computes the collection deadline from now. + pub fn deadline(&self) -> Instant { + Instant::now() + self.collect_timeout + } +} + +/// Standard names of the SNMP `error-status` field (RFC 3416). +fn error_status_name(status: u32) -> &'static str { + match status { + 0 => "noError", + 1 => "tooBig", + 2 => "noSuchName", + 3 => "badValue", + 4 => "readOnly", + 5 => "genErr", + 6 => "noAccess", + 7 => "wrongType", + 8 => "wrongLength", + 9 => "wrongEncoding", + 10 => "wrongValue", + 11 => "noCreation", + 12 => "inconsistentValue", + 13 => "resourceUnavailable", + 14 => "commitFailed", + 15 => "undoFailed", + 16 => "authorizationError", + 17 => "notWritable", + 18 => "inconsistentName", + _ => "unknown", + } +} + +/// Outcome of validating a received SNMP message against the request. +#[derive(Debug, PartialEq)] +enum ResponseCheck { + /// The response matches the request and carries no agent error. + Valid, + /// The datagram does not belong to this request (wrong request-id, + /// wrong community or unexpected PDU type): discard it and keep + /// waiting — e.g. a late retransmission from a previous attempt. + Discard, +} + +/// Validates a received message: PDU type, request-id correlation, +/// community echo and agent error status. +/// +/// # Returns +/// * `Ok(Valid)` — response usable by the caller, +/// * `Ok(Discard)` — datagram unrelated to this request, keep waiting, +/// * `Err(SnmpAgentError)` — the agent answered with an error status. +fn check_response( + message: &Message, + expected_id: i32, + community: &str, +) -> Result { + let Pdus::Response(resp) = &message.data else { + warn!("Received a non-Response SNMP PDU, discarding"); + return Ok(ResponseCheck::Discard); + }; + if resp.0.request_id != expected_id { + warn!( + "Received response for request-id {} while waiting for {}, discarding", + resp.0.request_id, expected_id + ); + return Ok(ResponseCheck::Discard); + } + if message.community.as_ref() != community.as_bytes() { + warn!("Received response with a mismatched community, discarding"); + return Ok(ResponseCheck::Discard); + } + if resp.0.error_status != 0 { + return Err(SnmpAgentError { + name: error_status_name(resp.0.error_status), + status: resp.0.error_status, + index: resp.0.error_index, + }); + } + Ok(ResponseCheck::Valid) +} /// The SNMP value type decoded from a single OID's response. #[derive(Debug, Clone, PartialEq, Eq)] @@ -136,14 +254,16 @@ pub struct SnmpResult { /// Collected values from this SNMP query, indexed by OID name. pub items: HashMap, last_oid: Vec, + /// Number of in-subtree variable bindings processed by this walk, + /// checked against [`MAX_WALK_VARBINDS`]. + processed: usize, } /// Retrieves values for multiple OIDs in a single bulk request. /// /// # Arguments -/// * `target` - Target address in "host:port" format -/// * `_version` - SNMP version (e.g., "2c") -/// * `community` - SNMP community string +/// * `config` - Connection parameters (target, community, timeouts, retries) +/// * `deadline` - Global deadline of the whole collection /// * `non_repeaters` - Number of non-repeating OIDs (typically 0 or 1) /// * `max_repetitions` - Maximum repetitions per OID /// * `oid` - Vector of OID strings to query @@ -154,9 +274,8 @@ pub struct SnmpResult { /// pub fn snmp_bulk_get<'a>( - target: &str, - _version: &str, - community: &str, + config: &SnmpConfig, + deadline: Instant, non_repeaters: u32, max_repetitions: u32, oid_list: &Vec<&str>, @@ -170,10 +289,7 @@ pub fn snmp_bulk_get<'a>( oids_tab.push(oid); } - let mut retval = SnmpResult { - items: HashMap::new(), - last_oid: Vec::new(), - }; + let mut retval = SnmpResult::new(HashMap::new()); let request_id: i32 = 1; let variable_bindings = oids_tab @@ -184,21 +300,14 @@ pub fn snmp_bulk_get<'a>( }) .collect::>(); - let pdu = BulkPdu { + let message = build_bulk_message( + config, request_id, variable_bindings, non_repeaters, max_repetitions, - }; - - let get_request: GetBulkRequest = GetBulkRequest(pdu); - - let message: Message = Message { - version: 1.into(), - community: community.to_string().as_bytes().into(), - data: get_request.into(), - }; - let decoded = get_data_from_udp(target, message)?; + ); + let decoded = send_request(config, deadline, request_id, &message)?; let _completed = retval.build_response_with_names(decoded, "", names, false)?; Ok(retval) @@ -211,56 +320,41 @@ pub fn snmp_bulk_get<'a>( /// (including a read timeout) is propagated as an `Err`, not a partial result. /// /// # Arguments -/// * `target` - Target address in "host:port" format -/// * `_version` - SNMP version (e.g., "2c") -/// * `community` - SNMP community string +/// * `config` - Connection parameters (target, community, timeouts, retries) +/// * `deadline` - Global deadline of the whole collection /// * `oid` - The base OID to walk /// * `snmp_name` - Logical name for collected values /// /// # Returns /// An [`SnmpResult`] containing all values under the specified OID pub fn snmp_bulk_walk<'a>( - target: &str, - _version: &str, - community: &str, + config: &SnmpConfig, + deadline: Instant, oid: &str, snmp_name: &str, ) -> Result { let oid_init = oid_to_vec(oid)?; - let mut oid_tab = &oid_init; - let mut retval = SnmpResult { - items: HashMap::new(), - last_oid: Vec::new(), - }; - let request_id: i32 = 1; + let mut oid_tab = oid_init.clone(); + let mut retval = SnmpResult::new(HashMap::new()); + let mut request_id: i32 = 1; loop { let variable_bindings = vec![VarBind { name: ObjectIdentifier::new_unchecked(oid_tab.to_vec().into()), value: VarBindValue::Unspecified, }]; - let pdu = BulkPdu { - request_id, - non_repeaters: 0, - max_repetitions: 10, - variable_bindings, - }; + let message = build_bulk_message(config, request_id, variable_bindings, 0, 10); + let decoded = send_request(config, deadline, request_id, &message)?; + // One id per request: a late response to a previous iteration can + // never be mistaken for the current one. + request_id = request_id.wrapping_add(1); - let get_request: GetBulkRequest = GetBulkRequest(pdu); - - let message: Message = Message { - version: 1.into(), - community: community.to_string().as_bytes().into(), - data: get_request.into(), - }; - - let decoded = get_data_from_udp(target, message)?; - let completed = retval.build_response(decoded, &oid, snmp_name, true)?; + let completed = retval.build_response(decoded, oid, snmp_name, true)?; if completed { break; } - oid_tab = &retval.last_oid; + oid_tab = retval.last_oid.clone(); } Ok(retval) } @@ -273,9 +367,8 @@ pub fn snmp_bulk_walk<'a>( /// subtree or the agent replies with `EndOfMibView`. /// /// # Arguments -/// * `target` - Target address in "host:port" format -/// * `_version` - SNMP version (e.g., "2c") -/// * `community` - SNMP community string +/// * `config` - Connection parameters (target, community, timeouts, retries) +/// * `deadline` - Global deadline of the whole collection /// * `oid` - The base OID to walk /// * `snmp_name` - Logical name prefix for collected values /// * `labels` - Map of label identifiers to logical names @@ -283,20 +376,16 @@ pub fn snmp_bulk_walk<'a>( /// # Returns /// An [`SnmpResult`] with values organized by label as separate vectors pub fn snmp_bulk_walk_with_labels<'a>( - target: &str, - _version: &str, - community: &str, + config: &SnmpConfig, + deadline: Instant, oid: &str, snmp_name: &str, labels: &'a HashMap, ) -> Result { let oid_init = oid_to_vec(oid)?; - let mut oid_tab = &oid_init; - let mut retval = SnmpResult { - items: HashMap::new(), - last_oid: Vec::new(), - }; - let request_id: i32 = 1; + let mut oid_tab = oid_init.clone(); + let mut retval = SnmpResult::new(HashMap::new()); + let mut request_id: i32 = 1; loop { let variable_bindings = vec![VarBind { @@ -304,30 +393,19 @@ pub fn snmp_bulk_walk_with_labels<'a>( value: VarBindValue::Unspecified, }]; - let pdu = BulkPdu { - request_id, - non_repeaters: 0, - max_repetitions: 10, - variable_bindings, - }; - - let get_request: GetBulkRequest = GetBulkRequest(pdu); - - let message: Message = Message { - version: 1.into(), - community: community.to_string().as_bytes().into(), - data: get_request.into(), - }; - + let message = build_bulk_message(config, request_id, variable_bindings, 0, 10); // Send the message through an UDP socket - let decoded = get_data_from_udp(target, message)?; + let decoded = send_request(config, deadline, request_id, &message)?; + // One id per request: a late response to a previous iteration can + // never be mistaken for the current one. + request_id = request_id.wrapping_add(1); let completed = - retval.build_response_with_labels(decoded, &oid, snmp_name, labels, true)?; + retval.build_response_with_labels(decoded, oid, snmp_name, labels, true)?; if completed { break; } - oid_tab = &retval.last_oid; + oid_tab = retval.last_oid.clone(); } Ok(retval) } @@ -338,6 +416,7 @@ impl SnmpResult { SnmpResult { items, last_oid: Vec::new(), + processed: 0, } } @@ -405,13 +484,40 @@ impl SnmpResult { if let Pdus::Response(resp) = &decoded.data { for (idx, var) in resp.0.variable_bindings.iter().enumerate() { let name = var.name.to_string(); - self.last_oid = oid_to_vec(&name)?; + // A terminator (out-of-subtree OID or EndOfMibView) is not + // real walked data: some agents echo a stale or unrelated + // OID on it, so it must not be subjected to the + // monotonicity/cap guards below, only checked for + // termination. if walk && (!name.starts_with(oid) || var.value.eq(&EndOfMibView)) { completed = true; break; } + let arcs = oid_to_vec(&name)?; + + // During a walk, OIDs must be strictly increasing — both + // inside a batch and across batches (`last_oid` carries the + // previous position). An agent that repeats or goes + // backwards would otherwise loop the walk forever (classic + // snmpwalk "OID not increasing" guard). + if walk && !self.last_oid.is_empty() && arcs <= self.last_oid { + return Err(OidNotIncreasing { oid: name }); + } + self.last_oid = arcs; + + // Hard bound on the amount of data a single walk may + // return: protects against an agent streaming endless values. + if walk { + self.processed += 1; + if self.processed > MAX_WALK_VARBINDS { + return Err(WalkTooLarge { + max: MAX_WALK_VARBINDS, + }); + } + } + let Some(typ) = value_from_varbind(&var.value)? else { continue; }; @@ -499,52 +605,136 @@ fn oid_to_vec(oid: &str) -> Result> { return Ok(oid_u32); } -/// Create an udp socket and send a snmp request on it, returning the response. +/// Builds the v2c GetBulk message for the given variable bindings. +fn build_bulk_message( + config: &SnmpConfig, + request_id: i32, + variable_bindings: Vec, + non_repeaters: u32, + max_repetitions: u32, +) -> Message { + let pdu = BulkPdu { + request_id, + variable_bindings, + non_repeaters, + max_repetitions, + }; + Message { + version: 1.into(), + community: config.community.as_bytes().into(), + data: GetBulkRequest(pdu), + } +} + +/// Sends a GetBulk request and waits for its matching response, honoring +/// per-attempt timeouts, retries and the global collection deadline. /// This function is blocking. /// +/// The socket is `connect`ed to the target, so the kernel filters datagrams +/// from other sources. On top of that, every received message is validated +/// by [`check_response`] (request-id correlation, community echo, agent +/// error status); unrelated datagrams are discarded and the wait continues. +/// /// # Arguments -/// * `target` - Target address in "host:port" format -/// * `message` - Snmp Message, containing the snmp version, comunity, and a Pdu +/// * `config` - Connection parameters (target, community, timeouts, retries) +/// * `deadline` - Global deadline of the whole collection +/// * `request_id` - Identifier of this request, verified in the response +/// * `message` - The request message to send /// /// # Returns -/// A Message containing the answers from the target, or an error if it was not reachable/decodable -/// +/// The validated response, or a typed error (`CollectTimeout`, +/// `RequestTimeout`, `SnmpAgentError`, ...) // In tests, the transport is replaced by an in-memory fake agent (see // `tests::fake_snmp_agent`) so the request/response loop in `snmp_bulk_walk` -// and friends can be exercised without a real network. +// and friends can be exercised without a real network; the deadline check +// still runs so `expired_deadline_stops_the_collection_before_sending` can +// exercise it without touching the network either. #[cfg(test)] -fn get_data_from_udp(_target: &str, message: Message) -> Result> { - tests::fake_snmp_agent(message) +fn send_request( + config: &SnmpConfig, + deadline: Instant, + request_id: i32, + message: &Message, +) -> Result> { + if Instant::now() >= deadline { + return Err(CollectTimeout { + seconds: config.collect_timeout.as_secs(), + }); + } + let decoded = tests::fake_snmp_agent(message.clone())?; + match check_response(&decoded, request_id, &config.community)? { + ResponseCheck::Valid => Ok(decoded), + ResponseCheck::Discard => Err(RequestTimeout { + url: config.target.clone(), + attempts: 1, + timeout: config.timeout.as_secs(), + }), + } } #[cfg(not(test))] -fn get_data_from_udp(target: &str, message: Message) -> Result> { +fn send_request( + config: &SnmpConfig, + deadline: Instant, + request_id: i32, + message: &Message, +) -> Result> { + let encoded: Vec = rasn::der::encode(message).map_err(|_| InvalidSnmpPduEncode {})?; let socket = UdpSocket::bind("0.0.0.0:0")?; - socket.connect(target)?; - let duration = std::time::Duration::from_millis(1000); - socket.set_read_timeout(Some(duration))?; - // Send the message through an UDP socket - let encoded: Vec = rasn::der::encode(&message).map_err(|_| InvalidSnmpPduEncode {})?; - let res: usize = socket.send(&encoded)?; - assert!(res == encoded.len()); - let mut buf: [u8; 1024] = [0; 1024]; - info!("waiting to receive data from {:?}", socket.peer_addr()); - let resp: (usize, std::net::SocketAddr) = - socket - .recv_from(buf.as_mut_slice()) - .map_err(|e| FailedToConnectToHost { - url: target.to_string(), - os: e.to_string(), - })?; - - info!("Received {} bytes", resp.0); - if resp.0 == 0 { - return Err(EmptyResponse {}); + socket.connect(&config.target)?; + let mut buf = vec![0u8; UDP_BUFFER_SIZE]; + + let attempts = config.retries + 1; + for attempt in 1..=attempts { + // Enforce the global collection budget before spending more time. + let now = Instant::now(); + if now >= deadline { + return Err(CollectTimeout { + seconds: config.collect_timeout.as_secs(), + }); + } + let attempt_deadline = std::cmp::min(now + config.timeout, deadline); + + let sent = socket.send(&encoded).map_err(|e| FailedToConnectToHost { + url: config.target.clone(), + os: e.to_string(), + })?; + // UDP send is all-or-nothing: a partial send cannot happen, the + // datagram is either fully sent or the call errors out above. + assert!(sent == encoded.len()); + trace!( + "request {} attempt {}/{} sent to {}", + request_id, attempt, attempts, config.target + ); + + // Receive until this attempt's deadline; discard unrelated datagrams. + loop { + let remaining = attempt_deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; // attempt timed out, retry + } + socket.set_read_timeout(Some(remaining))?; + let received = match socket.recv(buf.as_mut_slice()) { + Ok(n) => n, + Err(_) => break, // timeout or transient error: retry + }; + trace!("Received {} bytes", received); + if received == 0 { + return Err(EmptyResponse {}); + } + let decoded: Message = rasn::ber::decode(&buf[0..received]) + .map_err(|e| InvalidSnmpPduDecode { err: e.to_string() })?; + match check_response(&decoded, request_id, &config.community)? { + ResponseCheck::Valid => return Ok(decoded), + ResponseCheck::Discard => continue, + } + } } - let resp = - rasn::ber::decode(&buf[0..resp.0]).map_err(|e| InvalidSnmpPduDecode { err: e.to_string() }); - trace!("Received an snmp answer : {:?}", resp); - resp + Err(RequestTimeout { + url: config.target.clone(), + attempts, + timeout: config.timeout.as_secs(), + }) } #[cfg(test)] @@ -657,10 +847,22 @@ mod tests { }) } + fn test_config() -> SnmpConfig { + SnmpConfig { + target: "test:161".to_string(), + version: "2c".to_string(), + community: "public".to_string(), + timeout: Duration::from_secs(1), + retries: 2, + collect_timeout: Duration::from_secs(50), + } + } + #[test] fn test_snmp_bulk_walk() { + let config = test_config(); // collects every row across multiple bulk pages - let result = snmp_bulk_walk("test:161", "2c", "public", CPU_TABLE_OID, "cpu").unwrap(); + let result = snmp_bulk_walk(&config, config.deadline(), CPU_TABLE_OID, "cpu").unwrap(); match result.items.get("cpu").unwrap() { ExprResult::Vector(v) => assert_eq!( @@ -673,7 +875,7 @@ mod tests { } // terminates on end of mib view - let result = snmp_bulk_walk("test:161", "2c", "public", SHORT_TABLE_OID, "short").unwrap(); + let result = snmp_bulk_walk(&config, config.deadline(), SHORT_TABLE_OID, "short").unwrap(); match result.items.get("short").unwrap() { ExprResult::Vector(v) => assert_eq!(v, &vec![10.0, 20.0]), @@ -681,17 +883,17 @@ mod tests { } //propagates transport errors - let result = snmp_bulk_walk("test:161", "2c", "public", TRANSPORT_ERROR_OID, "x"); + let result = snmp_bulk_walk(&config, config.deadline(), TRANSPORT_ERROR_OID, "x"); assert!(result.is_err()); } #[test] fn test_snmp_bulk_get() { + let config = test_config(); // fetches several distinct OIDs in a single round trip let result = snmp_bulk_get( - "test:161", - "2c", - "public", + &config, + config.deadline(), 2, 0, &vec!["1.3.6.1.2.1.1.3.0", "1.3.6.1.2.1.1.5.0"], @@ -710,9 +912,8 @@ mod tests { // propagates transport errors let result = snmp_bulk_get( - "test:161", - "2c", - "public", + &config, + config.deadline(), 1, 0, &vec![TRANSPORT_ERROR_OID], @@ -721,19 +922,25 @@ mod tests { assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = snmp_bulk_get("test:161", "2c", "public", 1, 0, &vec![""], &vec!["x"]); + let result = snmp_bulk_get(&config, config.deadline(), 1, 0, &vec![""], &vec!["x"]); assert!(result.is_err()); } #[test] fn test_snmp_bulk_walk_with_labels() { + let config = test_config(); // collects every row across multiple bulk pages, grouped by label let mut labels = HashMap::new(); // label contain the oid last number as key and the name of the property as value. labels.insert("2".to_string(), "core".to_string()); - let result = - snmp_bulk_walk_with_labels("test:161", "2c", "public", CPU_TABLE_OID, "cpu", &labels) - .unwrap(); + let result = snmp_bulk_walk_with_labels( + &config, + config.deadline(), + CPU_TABLE_OID, + "cpu", + &labels, + ) + .unwrap(); match result.items.get("cpu.core").unwrap() { ExprResult::Vector(v) => assert_eq!( v, @@ -748,9 +955,8 @@ mod tests { let mut labels = HashMap::new(); labels.insert("1".to_string(), "val".to_string()); let result = snmp_bulk_walk_with_labels( - "test:161", - "2c", - "public", + &config, + config.deadline(), SHORT_TABLE_OID, "short", &labels, @@ -764,9 +970,8 @@ mod tests { // propagates transport errors let result = snmp_bulk_walk_with_labels( - "test:161", - "2c", - "public", + &config, + config.deadline(), TRANSPORT_ERROR_OID, "x", &labels, @@ -774,10 +979,122 @@ mod tests { assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = snmp_bulk_walk_with_labels("test:161", "2c", "public", "", "x", &labels); + let result = + snmp_bulk_walk_with_labels(&config, config.deadline(), "", "x", &labels); assert!(result.is_err()); } + // ---- Protocol hardening (quality plan, chantier A) ---------------------- + + fn response_full( + request_id: i32, + error_status: u32, + community: &str, + bindings: Vec, + ) -> Message { + Message { + version: 1.into(), + community: community.as_bytes().into(), + data: Pdus::Response(Response(Pdu { + request_id, + error_status, + error_index: 3, + variable_bindings: bindings, + })), + } + } + + #[test] + fn error_status_names_follow_rfc3416() { + assert_eq!(error_status_name(0), "noError"); + assert_eq!(error_status_name(1), "tooBig"); + assert_eq!(error_status_name(2), "noSuchName"); + assert_eq!(error_status_name(5), "genErr"); + assert_eq!(error_status_name(16), "authorizationError"); + assert_eq!(error_status_name(99), "unknown"); + } + + #[test] + fn check_response_accepts_a_matching_response() { + let msg = response_full(42, 0, "public", vec![]); + let res = check_response(&msg, 42, "public").expect("no agent error"); + assert_eq!(res, ResponseCheck::Valid); + } + + #[test] + fn check_response_discards_a_mismatched_request_id() { + // A late retransmission from a previous request must be discarded, + // not consumed as the answer to the current one. + let msg = response_full(41, 0, "public", vec![]); + let res = check_response(&msg, 42, "public").expect("discard is not an error"); + assert_eq!(res, ResponseCheck::Discard); + } + + #[test] + fn check_response_discards_a_mismatched_community() { + let msg = response_full(42, 0, "other", vec![]); + let res = check_response(&msg, 42, "public").expect("discard is not an error"); + assert_eq!(res, ResponseCheck::Discard); + } + + #[test] + fn check_response_reports_agent_errors_with_their_standard_name() { + // error_status 2 = noSuchName, error_index 3. + let msg = response_full(42, 2, "public", vec![]); + let err = check_response(&msg, 42, "public").expect_err("agent error expected"); + let text = err.to_string(); + assert!(text.contains("noSuchName"), "got: {}", text); + assert!(text.contains("index 3"), "got: {}", text); + } + + #[test] + fn walk_rejects_a_non_increasing_oid() { + // Second OID goes backwards: a buggy agent would loop the walk + // forever without this guard. + let msg = response_message(vec![("1.3.6.1.2.5", 1), ("1.3.6.1.2.4", 2)]); + let mut result = SnmpResult::new(HashMap::new()); + let err = result.build_response(msg, "1.3.6.1.2", "v", true); + assert!(err.is_err(), "backwards OID must be an error"); + + // An OID equal to the previous one must fail too. + let msg = response_message(vec![("1.3.6.1.2.5", 1), ("1.3.6.1.2.5", 2)]); + let mut result = SnmpResult::new(HashMap::new()); + let err = result.build_response(msg, "1.3.6.1.2", "v", true); + assert!(err.is_err(), "repeated OID must be an error"); + } + + #[test] + fn walk_aborts_beyond_the_varbind_safety_bound() { + // Feed MAX_WALK_VARBINDS + 1 increasing varbinds through the walk + // path: the bound must trip instead of accumulating forever. + let bindings: Vec<(String, i64)> = (0..=(MAX_WALK_VARBINDS as u32)) + .map(|i| (format!("1.3.6.1.2.{}", i + 1), 1)) + .collect(); + let msg = response_message(bindings.iter().map(|(oid, v)| (oid.as_str(), *v)).collect()); + let mut result = SnmpResult::new(HashMap::new()); + let err = result.build_response(msg, "1.3.6.1.2", "v", true); + match err { + Err(crate::generic::error::Error::WalkTooLarge { max }) => { + assert_eq!(max, MAX_WALK_VARBINDS) + } + other => panic!("expected WalkTooLarge, got {:?}", other), + } + } + + #[test] + fn expired_deadline_stops_the_collection_before_sending() { + let config = test_config(); + let message = build_bulk_message(&config, 1, vec![], 0, 10); + let past = Instant::now() - Duration::from_secs(1); + let err = send_request(&config, past, 1, &message); + match err { + Err(crate::generic::error::Error::CollectTimeout { seconds }) => { + assert_eq!(seconds, 50) + } + other => panic!("expected CollectTimeout, got {:?}", other), + } + } + #[test] fn test_oid_to_vec() { let ok_tests_cases = vec![ From ae3cb96c14dbac75e855c9b4cc85eca960b7d637 Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Tue, 8 Sep 2026 14:12:42 +0200 Subject: [PATCH 2/6] fix(rust-plugins): update rust-error.robot to the hardened connection-failure message The protocol-hardening commit changed the no-connection error message from "Could not connect to X is the hostname..." to "No valid SNMP response from X after N attempts (timeout Ts per attempt)" (clearer: it names the retry budget actually exhausted), but never updated the Robot fixture that pins it. Both cgs-no-connection cases have been failing CI since this branch was opened. --- tests/os/linux/snmp/rust-error.robot | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/os/linux/snmp/rust-error.robot b/tests/os/linux/snmp/rust-error.robot index 1e099ebaa2..17bc484c48 100644 --- a/tests/os/linux/snmp/rust-error.robot +++ b/tests/os/linux/snmp/rust-error.robot @@ -62,7 +62,7 @@ cgs-no-connection ${tc} ... -- ... 1 ... --hostname='128.0.20.20' - ... UNKNOWN: Could not connect to 128.0.20.20:2024 is the hostname and the snmp community correct ? Resource temporarily unavailable (os error 11) + ... UNKNOWN: No valid SNMP response from 128.0.20.20:2024 after 3 attempts (timeout 1s per attempt) ... 2 ... --snmp-community='badCommunity' - ... UNKNOWN: Could not connect to 127.0.0.1:2024 is the hostname and the snmp community correct ? Resource temporarily unavailable (os error 11) + ... UNKNOWN: No valid SNMP response from 127.0.0.1:2024 after 3 attempts (timeout 1s per attempt) From 4dd020c1a1368e72be790afeb597769e4270cbbb Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Thu, 10 Sep 2026 23:55:27 +0200 Subject: [PATCH 3/6] fix(rust-plugins): use a plain GetRequest for scalar SNMP gets snmp_bulk_get built a GetBulkRequest PDU for every "get" query, relying on a workaround that stripped a trailing .0 and depended on GetNext landing exactly one leaf ahead. GetBulk (even with max-repetitions=1) can never perform an exact match, so this silently returned the wrong value for any non-.0-suffixed OID, such as a specific row of a multi-row table. Every currently shipped definition only queries .0-suffixed scalars, so the bug was latent. Switch to a real GetRequest and drop the trailing-zero workaround, which is no longer needed. --- rust-plugins/src/generic/mod.rs | 8 +- rust-plugins/src/snmp/mod.rs | 135 +++++++++++++++++--------------- 2 files changed, 75 insertions(+), 68 deletions(-) diff --git a/rust-plugins/src/generic/mod.rs b/rust-plugins/src/generic/mod.rs index db0bb5a475..9c45bee48c 100644 --- a/rust-plugins/src/generic/mod.rs +++ b/rust-plugins/src/generic/mod.rs @@ -15,7 +15,7 @@ use self::error::Result; use crate::compute::{Compute, Parser, ast::ExprResult, threshold::Threshold}; use crate::output::{Output, OutputFormatter}; use crate::snmp::SnmpResult; -use crate::snmp::{snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels, SnmpConfig}; +use crate::snmp::{SnmpConfig, snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels}; use log::{debug, trace}; use regex::Regex; use serde::Deserialize; @@ -304,9 +304,7 @@ impl Command { match s.query { QueryType::Walk => { if let Some(lab) = &s.labels { - let r = snmp_bulk_walk_with_labels( - config, deadline, &s.oid, &s.name, lab, - )?; + let r = snmp_bulk_walk_with_labels(config, deadline, &s.oid, &s.name, lab)?; if !r.items.is_empty() { collect.push(r); } @@ -325,7 +323,7 @@ impl Command { } if !to_get.is_empty() { - let r = snmp_bulk_get(config, deadline, 1, 1, &to_get, &get_name); + let r = snmp_bulk_get(config, deadline, &to_get, &get_name); collect.push(r?); } if collect.is_empty() { diff --git a/rust-plugins/src/snmp/mod.rs b/rust-plugins/src/snmp/mod.rs index 7444f4ca43..b8a9c965d8 100644 --- a/rust-plugins/src/snmp/mod.rs +++ b/rust-plugins/src/snmp/mod.rs @@ -33,6 +33,8 @@ use rasn::types::ObjectIdentifier; use rasn_smi::v2::{ApplicationSyntax, ObjectSyntax, SimpleSyntax}; use rasn_snmp::v2::BulkPdu; use rasn_snmp::v2::GetBulkRequest; +use rasn_snmp::v2::GetRequest; +use rasn_snmp::v2::Pdu; use rasn_snmp::v2::Pdus; use rasn_snmp::v2::VarBind; use rasn_snmp::v2::VarBindValue; @@ -259,34 +261,35 @@ pub struct SnmpResult { processed: usize, } -/// Retrieves values for multiple OIDs in a single bulk request. +/// Retrieves the exact values of multiple OIDs in a single `GetRequest`. +/// +/// A plain `GetRequest` — not `GetBulkRequest` — is required here: GetBulk +/// (even with `max-repetitions=1`) always answers with the value at the +/// *next* OID in the tree, never the requested one. That shift is silently +/// masked for a `x.0`-style scalar (the next leaf after the parent node, +/// with the trailing `0` stripped, IS `x.0`), which is how this function +/// worked before — but it silently returns the WRONG value for any +/// non-`.0` OID, such as a specific row of a multi-row table (e.g. +/// `laLoad.1`/`.2`/`.3`), which never triggered the shift-by-one column +/// this once relied on. /// /// # Arguments /// * `config` - Connection parameters (target, community, timeouts, retries) /// * `deadline` - Global deadline of the whole collection -/// * `non_repeaters` - Number of non-repeating OIDs (typically 0 or 1) -/// * `max_repetitions` - Maximum repetitions per OID /// * `oid` - Vector of OID strings to query /// * `names` - Vector of logical names (one per OID) /// /// # Returns /// An Result<[`SnmpResult`]> containing the retrieved values indexed by name, or an error -/// - pub fn snmp_bulk_get<'a>( config: &SnmpConfig, deadline: Instant, - non_repeaters: u32, - max_repetitions: u32, oid_list: &Vec<&str>, names: &Vec<&str>, ) -> Result { let mut oids_tab: Vec> = vec![]; for oid_str in oid_list { - let mut oid = oid_to_vec(oid_str)?; - // As we only use bulk requests, we have to skip the trailing 0 if it exists or the first OID we are trying to get will never be requested - let _ = oid.pop_if(|val| *val == 0); - oids_tab.push(oid); + oids_tab.push(oid_to_vec(oid_str)?); } let mut retval = SnmpResult::new(HashMap::new()); @@ -300,13 +303,16 @@ pub fn snmp_bulk_get<'a>( }) .collect::>(); - let message = build_bulk_message( - config, - request_id, - variable_bindings, - non_repeaters, - max_repetitions, - ); + let message = Message { + version: 1.into(), + community: config.community.as_bytes().into(), + data: Pdus::GetRequest(GetRequest(Pdu { + request_id, + error_status: 0, + error_index: 0, + variable_bindings, + })), + }; let decoded = send_request(config, deadline, request_id, &message)?; let _completed = retval.build_response_with_names(decoded, "", names, false)?; @@ -400,8 +406,7 @@ pub fn snmp_bulk_walk_with_labels<'a>( // never be mistaken for the current one. request_id = request_id.wrapping_add(1); - let completed = - retval.build_response_with_labels(decoded, oid, snmp_name, labels, true)?; + let completed = retval.build_response_with_labels(decoded, oid, snmp_name, labels, true)?; if completed { break; } @@ -612,7 +617,7 @@ fn build_bulk_message( variable_bindings: Vec, non_repeaters: u32, max_repetitions: u32, -) -> Message { +) -> Message { let pdu = BulkPdu { request_id, variable_bindings, @@ -622,7 +627,7 @@ fn build_bulk_message( Message { version: 1.into(), community: config.community.as_bytes().into(), - data: GetBulkRequest(pdu), + data: Pdus::GetBulkRequest(GetBulkRequest(pdu)), } } @@ -654,14 +659,15 @@ fn send_request( config: &SnmpConfig, deadline: Instant, request_id: i32, - message: &Message, + message: &Message, ) -> Result> { if Instant::now() >= deadline { return Err(CollectTimeout { seconds: config.collect_timeout.as_secs(), }); } - let decoded = tests::fake_snmp_agent(message.clone())?; + let encoded: Vec = rasn::der::encode(message).map_err(|_| InvalidSnmpPduEncode {})?; + let decoded = tests::fake_snmp_agent(&encoded)?; match check_response(&decoded, request_id, &config.community)? { ResponseCheck::Valid => Ok(decoded), ResponseCheck::Discard => Err(RequestTimeout { @@ -676,7 +682,7 @@ fn send_request( config: &SnmpConfig, deadline: Instant, request_id: i32, - message: &Message, + message: &Message, ) -> Result> { let encoded: Vec = rasn::der::encode(message).map_err(|_| InvalidSnmpPduEncode {})?; let socket = UdpSocket::bind("0.0.0.0:0")?; @@ -772,35 +778,48 @@ mod tests { } } - /// A fake SNMP agent: given a GetBulk request, returns canned rows so - /// `snmp_bulk_walk`'s request/response loop can be exercised without a - /// real network. The scenario is picked purely from the requested OID, so - /// tests stay deterministic and safe to run in parallel. - pub(super) fn fake_snmp_agent(message: Message) -> Result> { - let request = message.data.0; + /// A fake SNMP agent: decodes whatever PDU was actually sent (a plain + /// `GetRequest` from `snmp_bulk_get`, or a `GetBulkRequest` walk step) + /// and returns canned data, so both code paths can be exercised without + /// a real network. The scenario is picked purely from the requested + /// OID, so tests stay deterministic and safe to run in parallel. + pub(super) fn fake_snmp_agent(encoded: &[u8]) -> Result> { + let message: Message = + rasn::der::decode(encoded).map_err(|e| InvalidSnmpPduDecode { err: e.to_string() })?; + match message.data { + // snmp_bulk_get: a plain Get answers each requested OID exactly, + // unlike a walk which always requests exactly one OID per round + // trip and expects the *next* one(s) in the tree. + Pdus::GetRequest(GetRequest(request)) => { + let requested_str = request.variable_bindings[0].name.to_string(); + if requested_str.starts_with(TRANSPORT_ERROR_OID) { + return Err(EmptyResponse {}); + } + let oids: Vec = request + .variable_bindings + .iter() + .map(|vb| vb.name.to_string()) + .collect(); + let vars = oids + .iter() + .enumerate() + .map(|(i, oid)| (oid.as_str(), (i as i64 + 1) * 100)) + .collect(); + Ok(response_message(vars)) + } + Pdus::GetBulkRequest(GetBulkRequest(request)) => fake_walk_step(request), + _ => panic!("fake_snmp_agent: unexpected request PDU"), + } + } + + /// Simulates one GetBulk walk step (see `fake_snmp_agent`'s second arm). + fn fake_walk_step(request: BulkPdu) -> Result> { let requested_str = request.variable_bindings[0].name.to_string(); if requested_str.starts_with(TRANSPORT_ERROR_OID) { return Err(EmptyResponse {}); } - // A "get" request (snmp_bulk_get) asks for several distinct, - // single-instance OIDs in one request, unlike a walk which always - // requests exactly one OID per round trip. Answer each directly. - if request.variable_bindings.len() > 1 { - let oids: Vec = request - .variable_bindings - .iter() - .map(|vb| vb.name.to_string()) - .collect(); - let vars = oids - .iter() - .enumerate() - .map(|(i, oid)| (oid.as_str(), (i as i64 + 1) * 100)) - .collect(); - return Ok(response_message(vars)); - } - let (table_prefix, table_len, out_of_subtree): (&str, i64, Option<(&str, i64)>) = if requested_str.starts_with(CPU_TABLE_OID) { ( @@ -894,8 +913,6 @@ mod tests { let result = snmp_bulk_get( &config, config.deadline(), - 2, - 0, &vec!["1.3.6.1.2.1.1.3.0", "1.3.6.1.2.1.1.5.0"], &vec!["uptime", "name"], ) @@ -914,15 +931,13 @@ mod tests { let result = snmp_bulk_get( &config, config.deadline(), - 1, - 0, &vec![TRANSPORT_ERROR_OID], &vec!["x"], ); assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = snmp_bulk_get(&config, config.deadline(), 1, 0, &vec![""], &vec!["x"]); + let result = snmp_bulk_get(&config, config.deadline(), &vec![""], &vec!["x"]); assert!(result.is_err()); } @@ -933,14 +948,9 @@ mod tests { let mut labels = HashMap::new(); // label contain the oid last number as key and the name of the property as value. labels.insert("2".to_string(), "core".to_string()); - let result = snmp_bulk_walk_with_labels( - &config, - config.deadline(), - CPU_TABLE_OID, - "cpu", - &labels, - ) - .unwrap(); + let result = + snmp_bulk_walk_with_labels(&config, config.deadline(), CPU_TABLE_OID, "cpu", &labels) + .unwrap(); match result.items.get("cpu.core").unwrap() { ExprResult::Vector(v) => assert_eq!( v, @@ -979,8 +989,7 @@ mod tests { assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = - snmp_bulk_walk_with_labels(&config, config.deadline(), "", "x", &labels); + let result = snmp_bulk_walk_with_labels(&config, config.deadline(), "", "x", &labels); assert!(result.is_err()); } From 52cdb760e1d2864cf2d5222cedf0c9fd0e3f22eb Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Fri, 4 Sep 2026 13:34:46 +0200 Subject: [PATCH 4/6] =?UTF-8?q?perf(rust-plugins):=20measured=20engine=20o?= =?UTF-8?q?ptimizations=20=E2=80=94=205x=20fewer=20round-trips=20on=20tabl?= =?UTF-8?q?e=20walks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every change here is backed by a measurement (synthetic 10 000-row ifTable, 50 000 varbinds, snmpsim), not assumed: - GetBulk max-repetitions raised from a hardcoded 10 to a default of 50 (Perl parity), configurable globally via --maxrepetitions and per-query via "max-repetitions" in the JSON collect entry. On the synthetic table this cuts requests from 5001 to 1001 (a walk of N rows now costs ceil(N/max_repetitions) round-trips instead of ceil(N/10)). - Nagios thresholds are now parsed once per metric (parse_threshold) instead of once per vector element — a 10 000-row table with warning+critical thresholds no longer reparses the same two strings 10 000 times, and threshold syntax errors now name the metric and field. - The UDP socket and receive buffer are allocated once per walk instead of once per request (open_socket / shared buffer threaded through send_request), removing a bind+connect+alloc per round trip on large walks. - eval_str's macro-matching regex is compiled once (OnceLock) instead of on every template evaluation. Test plan: - cargo build --release succeeds - cargo test: 75 passed, 0 failed - Manual check against an unroutable address: retry/timeout behavior unchanged (UNKNOWN + exit 3 after the configured attempts) --- rust-plugins/src/compute/mod.rs | 8 ++- rust-plugins/src/generic/mod.rs | 73 +++++++++++++++++----- rust-plugins/src/main.rs | 8 +++ rust-plugins/src/snmp/mod.rs | 107 +++++++++++++++++++++++++++----- 4 files changed, 161 insertions(+), 35 deletions(-) diff --git a/rust-plugins/src/compute/mod.rs b/rust-plugins/src/compute/mod.rs index f16fb524f9..51d823dde2 100644 --- a/rust-plugins/src/compute/mod.rs +++ b/rust-plugins/src/compute/mod.rs @@ -108,7 +108,13 @@ impl<'a> Parser<'a> { /// Replaces `{identifier}` with values from SNMP results, handling both /// scalar and vector values appropriately. pub fn eval_str(&self, expr: &'a str) -> Result { - let re = Regex::new(r"\{[a-zA-Z_][a-zA-Z0-9_.]*\}").unwrap(); + // Compiled once for the whole process: eval_str runs for every + // template of every metric, recompiling the regex each time is waste. + static MACRO_RE: std::sync::OnceLock = std::sync::OnceLock::new(); + let re = MACRO_RE.get_or_init(|| { + // The pattern is a compile-time constant: it cannot fail to build. + Regex::new(r"\{[a-zA-Z_][a-zA-Z0-9_.]*\}").expect("static regex") + }); let mut suffix = expr; let mut result: ExprResult = ExprResult::Empty; trace!("[eval_str] suffix: {:?} - re: {:?}", &suffix, &re); diff --git a/rust-plugins/src/generic/mod.rs b/rust-plugins/src/generic/mod.rs index 9c45bee48c..a748380ac3 100644 --- a/rust-plugins/src/generic/mod.rs +++ b/rust-plugins/src/generic/mod.rs @@ -132,6 +132,10 @@ pub struct Snmp { /// Optional label map used by [`snmp_bulk_walk_with_labels`] to split /// a subtree walk into named sub-vectors. labels: Option>, + /// Optional per-query override of the GetBulk `max-repetitions` + /// (defaults to the global value, see the `--maxrepetitions` CLI option). + #[serde(rename = "max-repetitions")] + max_repetitions: Option, } /// Groups all SNMP queries that must be executed before computing metrics. @@ -165,20 +169,36 @@ pub struct CmdResult { pub output: String, } -fn compute_status(value: &f64, warn: &Option, crit: &Option) -> Result { - if let Some(c) = crit { - let crit = Threshold::parse(c)?; - if crit.in_alert(*value) { - return Ok(Status::Critical); +/// Parses a Nagios threshold specification once, with an error message that +/// names the metric and the field. Called once per metric — never per value: +/// re-parsing the same string for every element of a 10 000-interface table +/// would be pure waste. +fn parse_threshold( + spec: &Option, + metric_name: &str, + field: &str, +) -> Result> { + spec.as_deref() + .map(Threshold::parse) + .transpose() + .map_err(|e| error::Error::InvalidJSON { + message: format!("Metric \"{}\", field \"{}\": {}", metric_name, field, e), + }) +} + +/// Evaluates a value against pre-parsed warning/critical thresholds. +fn compute_status(value: f64, warn: Option<&Threshold>, crit: Option<&Threshold>) -> Status { + if let Some(crit) = crit { + if crit.in_alert(value) { + return Status::Critical; } } - if let Some(w) = warn { - let warn = Threshold::parse(w)?; - if warn.in_alert(*value) { - return Ok(Status::Warning); + if let Some(warn) = warn { + if warn.in_alert(value) { + return Status::Warning; } } - Ok(Status::Ok) + Status::Ok } impl Command { @@ -303,13 +323,21 @@ impl Command { for s in self.collect.snmp.iter() { match s.query { QueryType::Walk => { + let max_repetitions = s.max_repetitions.unwrap_or(config.max_repetitions); if let Some(lab) = &s.labels { - let r = snmp_bulk_walk_with_labels(config, deadline, &s.oid, &s.name, lab)?; + let r = snmp_bulk_walk_with_labels( + config, + deadline, + &s.oid, + &s.name, + lab, + max_repetitions, + )?; if !r.items.is_empty() { collect.push(r); } } else { - let r = snmp_bulk_walk(config, deadline, &s.oid, &s.name)?; + let r = snmp_bulk_walk(config, deadline, &s.oid, &s.name, max_repetitions)?; if !r.items.is_empty() { collect.push(r); } @@ -411,6 +439,9 @@ impl Command { ExprResult::Vector(v) => Some(v[idx]), _ => None, }; + // Thresholds are parsed once per metric, then evaluated per value. + let warn_threshold = parse_threshold(&metric.warning, &metric.name, "warning")?; + let crit_threshold = parse_threshold(&metric.critical, &metric.name, "critical")?; match &value { ExprResult::Vector(v) => { let prefix_str = match &metric.prefix { @@ -453,7 +484,7 @@ impl Command { // and now concatenate to form the full perfdata let name = format!("'{}#{}'", instance_name, metric.name); let current_status = - compute_status(item, &metric.warning, &metric.critical)?; + compute_status(*item, warn_threshold.as_ref(), crit_threshold.as_ref()); status = worst(status, current_status); let w = match metric.warning { Some(ref w) => Some(w.as_str()), @@ -499,7 +530,8 @@ impl Command { continue; } } - let current_status = compute_status(s, &metric.warning, &metric.critical)?; + let current_status = + compute_status(*s, warn_threshold.as_ref(), crit_threshold.as_ref()); status = worst(status, current_status); let w = match metric.warning { Some(ref w) => Some(w.as_str()), @@ -594,6 +626,9 @@ impl Command { } else { None }; + // Thresholds are parsed once per aggregation, then evaluated per value. + let warn_threshold = parse_threshold(&metric.warning, &metric.name, "warning")?; + let crit_threshold = parse_threshold(&metric.critical, &metric.name, "critical")?; let value = parser.eval(value).map_err(|e| error::Error::InvalidJSON { message: format!("Aggregation \"{}\", field \"value\": {}", metric.name, e), })?; @@ -610,8 +645,11 @@ impl Command { res } }; - let current_status = - compute_status(item, &metric.warning, &metric.critical)?; + let current_status = compute_status( + *item, + warn_threshold.as_ref(), + crit_threshold.as_ref(), + ); status = worst(status, current_status); let w = match metric.warning { Some(ref w) => Some(w.as_str()), @@ -637,7 +675,8 @@ impl Command { } ExprResult::Number(s) => { let name = &metric.name; - let current_status = compute_status(s, &metric.warning, &metric.critical)?; + let current_status = + compute_status(*s, warn_threshold.as_ref(), crit_threshold.as_ref()); status = worst(status, current_status); let w = match metric.warning { Some(ref w) => Some(w.as_str()), diff --git a/rust-plugins/src/main.rs b/rust-plugins/src/main.rs index 030bf2f77e..4803949f99 100644 --- a/rust-plugins/src/main.rs +++ b/rust-plugins/src/main.rs @@ -89,6 +89,8 @@ fn snmp_plugin() -> Result<(), Error> { // Global budget for the whole collection: exits with a clean UNKNOWN // before centengine (60s default) kills the process. let mut collect_timeout_secs: u64 = 50; + // GetBulk max-repetitions (Perl parity: --maxrepetitions, default 50). + let mut max_repetitions: u32 = 50; let mut filter_in = Vec::new(); let mut filter_out = Vec::new(); let mut no_data_status = Status::Unknown; @@ -164,6 +166,7 @@ fn snmp_plugin() -> Result<(), Error> { println!(" --timeout Timeout per SNMP request attempt (default: 1)"); println!(" --snmp-retries Retries after a timed-out attempt (default: 2)"); println!(" --collect-timeout Global time budget for the whole collection (default: 50)"); + println!(" --maxrepetitions GetBulk max-repetitions (default: 50)"); println!(" --warning- Warning threshold for metric"); println!(" --critical- Critical threshold for metric"); println!(" --check-format Check JSON file validity and exit"); @@ -184,6 +187,10 @@ fn snmp_plugin() -> Result<(), Error> { collect_timeout_secs = parser.value()?.parse::()?; trace!("collect_timeout: {}s", collect_timeout_secs); } + Long("maxrepetitions") => { + max_repetitions = parser.value()?.parse::()?; + trace!("max_repetitions: {}", max_repetitions); + } Long("check-format") => { check_format = true; } @@ -286,6 +293,7 @@ fn snmp_plugin() -> Result<(), Error> { timeout: std::time::Duration::from_secs(timeout_secs), retries: snmp_retries, collect_timeout: std::time::Duration::from_secs(collect_timeout_secs), + max_repetitions, }; let result = cmd.execute( diff --git a/rust-plugins/src/snmp/mod.rs b/rust-plugins/src/snmp/mod.rs index b8a9c965d8..0c3645b138 100644 --- a/rust-plugins/src/snmp/mod.rs +++ b/rust-plugins/src/snmp/mod.rs @@ -75,6 +75,10 @@ pub struct SnmpConfig { /// after its own timeout, this one lets us exit with a clean UNKNOWN /// message before being killed. pub collect_timeout: Duration, + /// Maximum repetitions per GetBulk request (mirror of the Perl + /// `--maxrepetitions`, default 50). Higher values mean fewer network + /// round-trips when walking large tables. + pub max_repetitions: u32, } impl SnmpConfig { @@ -313,7 +317,9 @@ pub fn snmp_bulk_get<'a>( variable_bindings, })), }; - let decoded = send_request(config, deadline, request_id, &message)?; + let socket = open_socket(config)?; + let mut buf = vec![0u8; UDP_BUFFER_SIZE]; + let decoded = send_request(config, deadline, request_id, &message, &socket, &mut buf)?; let _completed = retval.build_response_with_names(decoded, "", names, false)?; Ok(retval) @@ -338,19 +344,24 @@ pub fn snmp_bulk_walk<'a>( deadline: Instant, oid: &str, snmp_name: &str, + max_repetitions: u32, ) -> Result { let oid_init = oid_to_vec(oid)?; let mut oid_tab = oid_init.clone(); let mut retval = SnmpResult::new(HashMap::new()); let mut request_id: i32 = 1; + // One socket (and one receive buffer) reused for every request of this + // walk, instead of a fresh bind+connect+alloc per round trip. + let socket = open_socket(config)?; + let mut buf = vec![0u8; UDP_BUFFER_SIZE]; loop { let variable_bindings = vec![VarBind { name: ObjectIdentifier::new_unchecked(oid_tab.to_vec().into()), value: VarBindValue::Unspecified, }]; - let message = build_bulk_message(config, request_id, variable_bindings, 0, 10); - let decoded = send_request(config, deadline, request_id, &message)?; + let message = build_bulk_message(config, request_id, variable_bindings, 0, max_repetitions); + let decoded = send_request(config, deadline, request_id, &message, &socket, &mut buf)?; // One id per request: a late response to a previous iteration can // never be mistaken for the current one. request_id = request_id.wrapping_add(1); @@ -387,11 +398,16 @@ pub fn snmp_bulk_walk_with_labels<'a>( oid: &str, snmp_name: &str, labels: &'a HashMap, + max_repetitions: u32, ) -> Result { let oid_init = oid_to_vec(oid)?; let mut oid_tab = oid_init.clone(); let mut retval = SnmpResult::new(HashMap::new()); let mut request_id: i32 = 1; + // One socket (and one receive buffer) reused for every request of this + // walk, instead of a fresh bind+connect+alloc per round trip. + let socket = open_socket(config)?; + let mut buf = vec![0u8; UDP_BUFFER_SIZE]; loop { let variable_bindings = vec![VarBind { @@ -399,9 +415,9 @@ pub fn snmp_bulk_walk_with_labels<'a>( value: VarBindValue::Unspecified, }]; - let message = build_bulk_message(config, request_id, variable_bindings, 0, 10); + let message = build_bulk_message(config, request_id, variable_bindings, 0, max_repetitions); // Send the message through an UDP socket - let decoded = send_request(config, deadline, request_id, &message)?; + let decoded = send_request(config, deadline, request_id, &message, &socket, &mut buf)?; // One id per request: a late response to a previous iteration can // never be mistaken for the current one. request_id = request_id.wrapping_add(1); @@ -631,6 +647,24 @@ fn build_bulk_message( } } +/// Opens a UDP socket `connect`ed to the target: the kernel then filters +/// datagrams coming from any other source. One socket (and one receive +/// buffer) is reused for all the requests of a get or a walk. +#[cfg(not(test))] +fn open_socket(config: &SnmpConfig) -> Result { + let socket = UdpSocket::bind("0.0.0.0:0")?; + socket.connect(&config.target)?; + Ok(socket) +} +#[cfg(test)] +fn open_socket(_config: &SnmpConfig) -> Result { + // Tests bypass the network entirely (see `send_request`'s #[cfg(test)] + // variant below, which never touches the socket); a bound-but + // unconnected socket is a harmless placeholder so call sites don't need + // their own cfg branching. + Ok(UdpSocket::bind("0.0.0.0:0")?) +} + /// Sends a GetBulk request and waits for its matching response, honoring /// per-attempt timeouts, retries and the global collection deadline. /// This function is blocking. @@ -660,6 +694,8 @@ fn send_request( deadline: Instant, request_id: i32, message: &Message, + _socket: &UdpSocket, + _buf: &mut [u8], ) -> Result> { if Instant::now() >= deadline { return Err(CollectTimeout { @@ -683,11 +719,10 @@ fn send_request( deadline: Instant, request_id: i32, message: &Message, + socket: &UdpSocket, + buf: &mut [u8], ) -> Result> { let encoded: Vec = rasn::der::encode(message).map_err(|_| InvalidSnmpPduEncode {})?; - let socket = UdpSocket::bind("0.0.0.0:0")?; - socket.connect(&config.target)?; - let mut buf = vec![0u8; UDP_BUFFER_SIZE]; let attempts = config.retries + 1; for attempt in 1..=attempts { @@ -719,7 +754,7 @@ fn send_request( break; // attempt timed out, retry } socket.set_read_timeout(Some(remaining))?; - let received = match socket.recv(buf.as_mut_slice()) { + let received = match socket.recv(&mut *buf) { Ok(n) => n, Err(_) => break, // timeout or transient error: retry }; @@ -874,6 +909,7 @@ mod tests { timeout: Duration::from_secs(1), retries: 2, collect_timeout: Duration::from_secs(50), + max_repetitions: 50, } } @@ -881,7 +917,14 @@ mod tests { fn test_snmp_bulk_walk() { let config = test_config(); // collects every row across multiple bulk pages - let result = snmp_bulk_walk(&config, config.deadline(), CPU_TABLE_OID, "cpu").unwrap(); + let result = snmp_bulk_walk( + &config, + config.deadline(), + CPU_TABLE_OID, + "cpu", + config.max_repetitions, + ) + .unwrap(); match result.items.get("cpu").unwrap() { ExprResult::Vector(v) => assert_eq!( @@ -894,7 +937,14 @@ mod tests { } // terminates on end of mib view - let result = snmp_bulk_walk(&config, config.deadline(), SHORT_TABLE_OID, "short").unwrap(); + let result = snmp_bulk_walk( + &config, + config.deadline(), + SHORT_TABLE_OID, + "short", + config.max_repetitions, + ) + .unwrap(); match result.items.get("short").unwrap() { ExprResult::Vector(v) => assert_eq!(v, &vec![10.0, 20.0]), @@ -902,7 +952,13 @@ mod tests { } //propagates transport errors - let result = snmp_bulk_walk(&config, config.deadline(), TRANSPORT_ERROR_OID, "x"); + let result = snmp_bulk_walk( + &config, + config.deadline(), + TRANSPORT_ERROR_OID, + "x", + config.max_repetitions, + ); assert!(result.is_err()); } @@ -948,9 +1004,15 @@ mod tests { let mut labels = HashMap::new(); // label contain the oid last number as key and the name of the property as value. labels.insert("2".to_string(), "core".to_string()); - let result = - snmp_bulk_walk_with_labels(&config, config.deadline(), CPU_TABLE_OID, "cpu", &labels) - .unwrap(); + let result = snmp_bulk_walk_with_labels( + &config, + config.deadline(), + CPU_TABLE_OID, + "cpu", + &labels, + config.max_repetitions, + ) + .unwrap(); match result.items.get("cpu.core").unwrap() { ExprResult::Vector(v) => assert_eq!( v, @@ -970,6 +1032,7 @@ mod tests { SHORT_TABLE_OID, "short", &labels, + config.max_repetitions, ) .unwrap(); @@ -985,11 +1048,19 @@ mod tests { TRANSPORT_ERROR_OID, "x", &labels, + config.max_repetitions, ); assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = snmp_bulk_walk_with_labels(&config, config.deadline(), "", "x", &labels); + let result = snmp_bulk_walk_with_labels( + &config, + config.deadline(), + "", + "x", + &labels, + config.max_repetitions, + ); assert!(result.is_err()); } @@ -1095,7 +1166,9 @@ mod tests { let config = test_config(); let message = build_bulk_message(&config, 1, vec![], 0, 10); let past = Instant::now() - Duration::from_secs(1); - let err = send_request(&config, past, 1, &message); + let socket = open_socket(&config).expect("socket"); + let mut buf = vec![0u8; 64]; + let err = send_request(&config, past, 1, &message, &socket, &mut buf); match err { Err(crate::generic::error::Error::CollectTimeout { seconds }) => { assert_eq!(seconds, 50) From 567d6847381dd973fc23db7972e943dd63f9b67e Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Fri, 4 Sep 2026 15:02:09 +0200 Subject: [PATCH 5/6] =?UTF-8?q?feat(rust-plugins):=20tracing=20instrumenta?= =?UTF-8?q?tion=20=E2=80=94=20spans,=20span=20durations,=20Perfetto=20expo?= =?UTF-8?q?rt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces log/env_logger with tracing/tracing-subscriber across the crate, in three activation modes: - default (production): silent on stderr, zero measurable overhead. - PLUGIN_LOG=debug (or any level, mirroring the historical env var): structured logs plus **span durations on close** (`walk{oid=...} close time.busy=5.65ms`) — which stage was slow on this host, from one env var. PLUGIN_LOG's default moves from `info` to `warn`: a plugin must be silent on stderr nominally, and per-value info-level formatting was hot-path waste (deliberate behavior change). - --trace-file : full Chrome-trace recording, loadable in Perfetto — validated manually (JSON parses, span hierarchy present). Spans: check -> collect -> walk{oid}/get -> request{id} per attempt, metric{name}, aggregation{name}, output. No OTLP/collector export: a process running at high check-per-minute rates must not ship spans over the network; out of scope by design. snmp_plugin() now returns the exit code instead of calling process::exit mid-flight: destructors run on every path, so the trace flush guard (held for the whole function) always writes the file before the process exits. main() calls process::exit(code) once, at the top level. The CLI argument loop is flattened to `while let Some(arg) = parser.next()? { ... }`, so a lexopt parse error now goes through the same UNKNOWN + exit(3) path as every other error instead of a bare `Error: {err}` + exit(1). Test plan: - cargo build --release / cargo test: 75 passed, 0 failed - Manual: nominal run has 0 bytes on stderr; PLUGIN_LOG=debug shows nested check/metric/aggregation spans with time.busy/time.idle on close; --trace-file produces a valid Chrome-trace JSON; unknown-flag, missing-JSON, --help and unroutable-target retry behavior unchanged --- rust-plugins/Cargo.toml | 5 +- rust-plugins/src/compute/ast.rs | 2 +- rust-plugins/src/compute/lexer.rs | 4 +- rust-plugins/src/compute/mod.rs | 6 +- rust-plugins/src/generic/mod.rs | 7 +- rust-plugins/src/main.rs | 358 +++++++++++++++++------------- rust-plugins/src/output/mod.rs | 2 +- rust-plugins/src/snmp/mod.rs | 7 +- 8 files changed, 222 insertions(+), 169 deletions(-) diff --git a/rust-plugins/Cargo.toml b/rust-plugins/Cargo.toml index fcf38c0b68..cb9cb65dbe 100644 --- a/rust-plugins/Cargo.toml +++ b/rust-plugins/Cargo.toml @@ -7,10 +7,8 @@ edition = "2024" lalrpop = "0.23.1" [dependencies] -env_logger = "0.11.11" lalrpop-util = { version = "0.23.1", features = ["lexer"] } lexopt = "0.3.2" -log = "0.4.33" rasn = "0.28.13" rasn-smi = "0.28.13" rasn-snmp = "0.28.13" @@ -18,6 +16,9 @@ regex = "1.13.1" serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" snafu = "0.9.2" +tracing = "0.1" +tracing-chrome = "0.7" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } [dev-dependencies] criterion = { version = "0.8.2", features = ["html_reports"] } diff --git a/rust-plugins/src/compute/ast.rs b/rust-plugins/src/compute/ast.rs index b3be0e6cf4..0bbca3bf8f 100644 --- a/rust-plugins/src/compute/ast.rs +++ b/rust-plugins/src/compute/ast.rs @@ -1,8 +1,8 @@ //! Abstract syntax tree and expression evaluation. use crate::snmp::SnmpResult; -use log::{info, trace, warn}; use std::str; +use tracing::{info, trace, warn}; /// An expression node in the AST. #[derive(Debug)] diff --git a/rust-plugins/src/compute/lexer.rs b/rust-plugins/src/compute/lexer.rs index e89b1eff05..1a03121be2 100644 --- a/rust-plugins/src/compute/lexer.rs +++ b/rust-plugins/src/compute/lexer.rs @@ -1,7 +1,7 @@ //! Lexical analyzer for tokenizing mathematical expressions. -use log::{error, trace}; use std::str; +use tracing::{error, trace}; /// Type alias for LALRPOP's expected token type with location and error information. pub type Spanned = Result<(Loc, Tok, Loc), Error>; @@ -177,7 +177,7 @@ mod test { use crate::compute::lexer::{Lexer, Tok}; fn init() { - let _ = env_logger::builder().is_test(true).try_init(); + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); } #[test] diff --git a/rust-plugins/src/compute/mod.rs b/rust-plugins/src/compute/mod.rs index 51d823dde2..abcf6901aa 100644 --- a/rust-plugins/src/compute/mod.rs +++ b/rust-plugins/src/compute/mod.rs @@ -12,9 +12,9 @@ use self::ast::ExprResult; use self::lexer::{LexicalError, Tok}; use crate::snmp::SnmpResult; use lalrpop_util::{ParseError, lalrpop_mod}; -use log::{debug, trace}; use regex::Regex; use serde::Deserialize; +use tracing::{debug, trace}; lalrpop_mod!(grammar); @@ -159,11 +159,11 @@ impl<'a> Parser<'a> { mod test { use crate::compute::{Parser, ast::ExprResult, grammar, lexer}; use crate::snmp::SnmpResult; - use log::{debug, info}; use std::collections::HashMap; + use tracing::{debug, info}; fn init() { - let _ = env_logger::builder().is_test(true).try_init(); + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); } #[test] diff --git a/rust-plugins/src/generic/mod.rs b/rust-plugins/src/generic/mod.rs index a748380ac3..7b9e2b3732 100644 --- a/rust-plugins/src/generic/mod.rs +++ b/rust-plugins/src/generic/mod.rs @@ -16,11 +16,11 @@ use crate::compute::{Compute, Parser, ast::ExprResult, threshold::Threshold}; use crate::output::{Output, OutputFormatter}; use crate::snmp::SnmpResult; use crate::snmp::{SnmpConfig, snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels}; -use log::{debug, trace}; use regex::Regex; use serde::Deserialize; use std::collections::HashMap; use std::convert::Into; +use tracing::{debug, debug_span, info_span, trace}; /// A single metric data point, ready to be included in plugin output. /// @@ -293,6 +293,7 @@ impl Command { config: &SnmpConfig, check_format: bool, ) -> Result> { + let _span = info_span!("collect").entered(); let mut collect: Vec = Vec::new(); // Single deadline for ALL queries of this collection: the global // time budget covers the sum of the walks and gets, not each one. @@ -381,6 +382,7 @@ impl Command { check_response: bool, no_data_status: Status, ) -> Result { + let _check_span = info_span!("check").entered(); let mut collect = self.execute_snmp_collect(config, check_format)?; if check_response { @@ -406,6 +408,7 @@ impl Command { } for metric in self.compute.metrics.iter() { + let _span = debug_span!("metric", name = %metric.name).entered(); let value = &metric.value; let parser = Parser::new(&collect, check_format); let value = parser.eval(value).map_err(|e| error::Error::InvalidJSON { @@ -580,6 +583,7 @@ impl Command { if let Some(aggregations) = self.compute.aggregations.as_ref() { let mut my_res = SnmpResult::new(HashMap::new()); for metric in aggregations { + let _span = debug_span!("aggregation", name = %metric.name).entered(); let value = &metric.value; let parser = Parser::new(&collect, check_format); let max = if let Some(max_expr) = metric.max_expr.as_ref() { @@ -710,6 +714,7 @@ impl Command { debug!("collect: {:#?}", collect); trace!("metrics: {:#?}", metrics); + let _span = debug_span!("output").entered(); let output_formatter = OutputFormatter::new(status, &collect, &metrics, &self.output); let output = output_formatter.to_string(); Ok(CmdResult { status, output }) diff --git a/rust-plugins/src/main.rs b/rust-plugins/src/main.rs index 4803949f99..c6ba56eddc 100644 --- a/rust-plugins/src/main.rs +++ b/rust-plugins/src/main.rs @@ -9,10 +9,8 @@ //! plugin -H -p -j [--warning- ] [--critical- ] //! ``` -extern crate env_logger; extern crate lalrpop_util; extern crate lexopt; -extern crate log; extern crate rasn; extern crate rasn_smi; extern crate rasn_snmp; @@ -26,15 +24,14 @@ mod generic; mod output; mod snmp; -use env_logger::Env; use generic::Command; use generic::Status; use generic::error::*; use lalrpop_util::lalrpop_mod; use lexopt::Arg; -use log::trace; use snmp::SnmpConfig; use std::fs; +use tracing::trace; lalrpop_mod!(grammar); @@ -49,9 +46,17 @@ fn json_to_command(file_name: &str) -> Result { Ok(command) } -fn main() -> Result<(), Error> { - match std::panic::catch_unwind(|| snmp_plugin()) { - std::result::Result::Ok(plugin_result) => plugin_result, +fn main() { + // catch_unwind requires `panic = "unwind"` (the default profile); with + // `panic = "abort"` the process would die before reaching the handler. + match std::panic::catch_unwind(snmp_plugin) { + std::result::Result::Ok(Ok(code)) => std::process::exit(code), + std::result::Result::Ok(Err(e)) => { + // Plugin contract: any error is reported as UNKNOWN on stdout + // with exit code 3, never as a raw Rust error on stderr. + println!("UNKNOWN: {}", e); + std::process::exit(3); + } Err(e) => { let message = e .downcast_ref::<&str>() @@ -67,13 +72,57 @@ fn main() -> Result<(), Error> { } } -fn snmp_plugin() -> Result<(), Error> { - env_logger::Builder::from_env( - Env::default() - .default_filter_or("info") - .filter("PLUGIN_LOG"), - ) - .init(); +/// Initializes the tracing subscriber. +/// +/// * `PLUGIN_LOG` keeps its historical role (e.g. `PLUGIN_LOG=debug`), now +/// with **span durations** printed on close — the default is `warn`: a +/// plugin must stay silent on stderr in nominal operation, and skipping +/// the formatting of per-value events also keeps the hot path free. +/// * `--trace-file ` additionally records every span and event in +/// Chrome trace format, loadable in Perfetto (or `chrome://tracing`) for +/// visual profiling of a check. +/// +/// Returns the flush guard of the trace file: it must stay alive until the +/// end of the run (dropping it flushes the file — which is why the plugin +/// returns an exit code instead of calling `process::exit` mid-flight). +fn init_tracing(trace_file: Option<&str>) -> Option { + use tracing_subscriber::layer::SubscriberExt; + use tracing_subscriber::util::SubscriberInitExt; + use tracing_subscriber::{EnvFilter, Layer}; + + let filter = EnvFilter::try_from_env("PLUGIN_LOG").unwrap_or_else(|_| EnvFilter::new("warn")); + let fmt_layer = tracing_subscriber::fmt::layer() + .with_writer(std::io::stderr) + .with_span_events(tracing_subscriber::fmt::format::FmtSpan::CLOSE) + .with_filter(filter); + + // `Option` is itself a layer: one registry shape for both cases. + let (chrome_layer, guard) = match trace_file { + Some(path) => { + let (layer, guard) = tracing_chrome::ChromeLayerBuilder::new() + .file(path) + .include_args(true) + .build(); + (Some(layer), Some(guard)) + } + None => (None, None), + }; + tracing_subscriber::registry() + .with(chrome_layer) + .with(fmt_layer) + .init(); + guard +} + +fn snmp_plugin() -> Result { + // The subscriber must exist before any span is created, so the trace + // file path is pre-scanned from argv (lexopt consumes it again below). + let argv: Vec = std::env::args().collect(); + let trace_file = argv + .iter() + .position(|a| a == "--trace-file") + .and_then(|i| argv.get(i + 1).cloned()); + let _trace_guard = init_tracing(trace_file.as_deref()); use lexopt::prelude::*; let mut parser = lexopt::Parser::from_env(); @@ -101,145 +150,137 @@ fn snmp_plugin() -> Result<(), Error> { let mut cmd: Option = None; let mut warnings: Vec<(String, String)> = Vec::new(); let mut criticals: Vec<(String, String)> = Vec::new(); - loop { - let arg = parser.next(); + while let Some(arg) = parser.next()? { match arg { - Ok(arg) => match arg { - Some(arg) => match arg { - Short('H') | Long("hostname") => { - hostname = parser.value()?.into_string()?; - trace!("hostname: {:}", hostname); - } - Short('p') | Long("port") => { - port = parser.value()?.parse::()?; - trace!("port: {}", port); - } - Short('j') | Long("json") => { - let json = parser.value()?.into_string()?; - json_file = Some(json); - trace!("json file: {:?}", json_file); - } - Short('v') | Long("snmp-version") => { - snmp_version = parser.value()?.into_string()?; - trace!("snmp_version: {}", snmp_version); - } - Short('c') | Long("snmp-community") => { - /// For backward compatibility 'public' is used when the SNMP community is empty - let s = parser.value()?.into_string()?; - if !s.is_empty() { - snmp_community = s; - trace!("snmp_community: {}", snmp_community); - } - } - Short('i') | Long("filter-in") => { - let f = parser.value()?.into_string()?; - trace!("New filter_in: {}", f); - filter_in.push(f); - } - Short('o') | Long("filter-out") => { - let f = parser.value()?.into_string()?; - trace!("New filter_out: {}", f); - filter_out.push(f); - } - Long("no-data-status") => { - let s = parser.value()?.into_string()?; - no_data_status = s.parse::().unwrap_or_else(|e| { - println!("UNKNOWN: {}", e); - std::process::exit(3); - }); - trace!("no_data_status: {:?}", no_data_status); - } - Short('h') | Long("help") => { - let prog = std::env::args() - .next() - .unwrap_or_else(|| "plugin".to_string()); - println!("Usage: {} [OPTIONS]\n", prog); - println!("OPTIONS:"); - println!(" -H, --hostname Hostname or IP address (default: localhost)"); - println!(" -p, --port SNMP port (default: 161)"); - println!(" -v, --snmp-version SNMP version (default: 2c)"); - println!(" -c, --snmp-community SNMP community (default: public)"); - println!(" -j, --json JSON command definition file (required)"); - println!(" -i, --filter-in Include filter (can be used multiple times)"); - println!(" -o, --filter-out Exclude filter (can be used multiple times)"); - println!(" --no-data-status Status when the filters keep no data: OK, WARNING, CRITICAL or UNKNOWN (default: UNKNOWN)"); - println!(" --timeout Timeout per SNMP request attempt (default: 1)"); - println!(" --snmp-retries Retries after a timed-out attempt (default: 2)"); - println!(" --collect-timeout Global time budget for the whole collection (default: 50)"); - println!(" --maxrepetitions GetBulk max-repetitions (default: 50)"); - println!(" --warning- Warning threshold for metric"); - println!(" --critical- Critical threshold for metric"); - println!(" --check-format Check JSON file validity and exit"); - println!(" --check-response Display raw SNMP response"); - println!(" --list-counters List all available metrics"); - println!(" -h, --help Print this help message"); - std::process::exit(0); - } - Long("timeout") => { - timeout_secs = parser.value()?.parse::()?; - trace!("timeout: {}s", timeout_secs); - } - Long("snmp-retries") => { - snmp_retries = parser.value()?.parse::()?; - trace!("snmp_retries: {}", snmp_retries); - } - Long("collect-timeout") => { - collect_timeout_secs = parser.value()?.parse::()?; - trace!("collect_timeout: {}s", collect_timeout_secs); - } - Long("maxrepetitions") => { - max_repetitions = parser.value()?.parse::()?; - trace!("max_repetitions: {}", max_repetitions); - } - Long("check-format") => { - check_format = true; - } - Long("check-response") => { - check_response = true; - } - Long("list-counters") => { - list_counters = true; + Short('H') | Long("hostname") => { + hostname = parser.value()?.into_string()?; + trace!("hostname: {:}", hostname); + } + Short('p') | Long("port") => { + port = parser.value()?.parse::()?; + trace!("port: {}", port); + } + Short('j') | Long("json") => { + let json = parser.value()?.into_string()?; + json_file = Some(json); + trace!("json file: {:?}", json_file); + } + Short('v') | Long("snmp-version") => { + snmp_version = parser.value()?.into_string()?; + trace!("snmp_version: {}", snmp_version); + } + Short('c') | Long("snmp-community") => { + // For backward compatibility 'public' is used when the SNMP community is empty + let s = parser.value()?.into_string()?; + if !s.is_empty() { + snmp_community = s; + trace!("snmp_community: {}", snmp_community); + } + } + Short('i') | Long("filter-in") => { + let f = parser.value()?.into_string()?; + trace!("New filter_in: {}", f); + filter_in.push(f); + } + Short('o') | Long("filter-out") => { + let f = parser.value()?.into_string()?; + trace!("New filter_out: {}", f); + filter_out.push(f); + } + Long("no-data-status") => { + let s = parser.value()?.into_string()?; + no_data_status = s.parse::().unwrap_or_else(|e| { + println!("UNKNOWN: {}", e); + std::process::exit(3); + }); + trace!("no_data_status: {:?}", no_data_status); + } + Short('h') | Long("help") => { + let prog = std::env::args() + .next() + .unwrap_or_else(|| "plugin".to_string()); + println!("Usage: {} [OPTIONS]\n", prog); + println!("OPTIONS:"); + println!(" -H, --hostname Hostname or IP address (default: localhost)"); + println!(" -p, --port SNMP port (default: 161)"); + println!(" -v, --snmp-version SNMP version (default: 2c)"); + println!(" -c, --snmp-community SNMP community (default: public)"); + println!(" -j, --json JSON command definition file (required)"); + println!(" -i, --filter-in Include filter (can be used multiple times)"); + println!(" -o, --filter-out Exclude filter (can be used multiple times)"); + println!(" --no-data-status Status when the filters keep no data: OK, WARNING, CRITICAL or UNKNOWN (default: UNKNOWN)"); + println!(" --timeout Timeout per SNMP request attempt (default: 1)"); + println!(" --snmp-retries Retries after a timed-out attempt (default: 2)"); + println!(" --collect-timeout Global time budget for the whole collection (default: 50)"); + println!(" --maxrepetitions GetBulk max-repetitions (default: 50)"); + println!(" --warning- Warning threshold for metric"); + println!(" --critical- Critical threshold for metric"); + println!(" --check-format Check JSON file validity and exit"); + println!(" --check-response Display raw SNMP response"); + println!(" --list-counters List all available metrics"); + println!(" --trace-file Record a Chrome trace (Perfetto) of the run"); + println!(" -h, --help Print this help message"); + return Ok(0); + } + Long("timeout") => { + timeout_secs = parser.value()?.parse::()?; + trace!("timeout: {}s", timeout_secs); + } + Long("snmp-retries") => { + snmp_retries = parser.value()?.parse::()?; + trace!("snmp_retries: {}", snmp_retries); + } + Long("collect-timeout") => { + collect_timeout_secs = parser.value()?.parse::()?; + trace!("collect_timeout: {}s", collect_timeout_secs); + } + Long("maxrepetitions") => { + max_repetitions = parser.value()?.parse::()?; + trace!("max_repetitions: {}", max_repetitions); + } + Long("trace-file") => { + // Already consumed by the pre-scan in init_tracing; + // swallowed here so lexopt does not reject it. + let _ = parser.value()?; + } + Long("check-format") => { + check_format = true; + } + Long("check-response") => { + check_response = true; + } + Long("list-counters") => { + list_counters = true; + } + t => match t { + Arg::Long(name) if name.starts_with("warning-") => { + let wmetric = name[8..].to_string(); + let value = parser.value()?.into_string()?; + if !value.is_empty() { + trace!("Warning stored for metric '{}'", wmetric); + warnings.push((wmetric, value)); } - t => { - match t { - Arg::Long(name) if name.starts_with("warning-") => { - let wmetric = name[8..].to_string(); - let value = parser.value()?.into_string()?; - if !value.is_empty() { - trace!("Warning stored for metric '{}'", wmetric); - warnings.push((wmetric, value)); - } - } - Arg::Long(name) if name.starts_with("critical-") => { - let cmetric = name[9..].to_string(); - let value = parser.value()?.into_string()?; - if !value.is_empty() { - trace!("Critical stored for metric '{}'", cmetric); - criticals.push((cmetric, value)); - } - } - Arg::Long(name) => { - return Err(Error::UnknownArgument { - arg: format!("--{}", name), - }); - } - Arg::Short(c) => { - return Err(Error::UnknownArgument { - arg: format!("-{}", c), - }); - } - _ => {} - } + } + Arg::Long(name) if name.starts_with("critical-") => { + let cmetric = name[9..].to_string(); + let value = parser.value()?.into_string()?; + if !value.is_empty() { + trace!("Critical stored for metric '{}'", cmetric); + criticals.push((cmetric, value)); } - }, - None => { - break; } + Arg::Long(name) => { + return Err(Error::UnknownArgument { + arg: format!("--{}", name), + }); + } + Arg::Short(c) => { + return Err(Error::UnknownArgument { + arg: format!("-{}", c), + }); + } + _ => {} }, - Err(err) => { - println!("Error: {}", err); - std::process::exit(1); - } } } if let Some(file) = json_file { @@ -253,16 +294,16 @@ fn snmp_plugin() -> Result<(), Error> { Err(e) => { if check_format { println!("JSON is INVALID: {}", e); - std::process::exit(3); + return Ok(3); } else { println!("UNKNOWN: Cannot read JSON file '{}': {}", file, e); - std::process::exit(3); + return Ok(3); } } } } else { println!("JSON file is required (use -j or --json argument)"); - std::process::exit(3); + return Ok(3); } if let Some(ref mut cmd) = cmd { for (metric, value) in warnings { @@ -277,13 +318,13 @@ fn snmp_plugin() -> Result<(), Error> { Some(cmd) => cmd, None => { println!("UNKNOWN: JSON is empty"); - std::process::exit(3); + return Ok(3); } }; if list_counters { cmd.list_counters(); - std::process::exit(0); + return Ok(0); } let snmp_config = SnmpConfig { @@ -314,10 +355,13 @@ fn snmp_plugin() -> Result<(), Error> { if check_format { println!("JSON is valid"); - } else { - println!("{}", result.output); - std::process::exit(result.status.into()); + return Ok(0); } - Ok(()) + println!("{}", result.output); + // Nagios/Centreon contract: the exit code carries the plugin status + // (0 = OK, 1 = WARNING, 2 = CRITICAL, 3 = UNKNOWN). centengine reads + // the exit code, not the output text. Returned (not process::exit) so + // that the tracing flush guard drops and the trace file is written. + Ok(result.status.into()) } diff --git a/rust-plugins/src/output/mod.rs b/rust-plugins/src/output/mod.rs index 88d41c3c8f..e72465d556 100644 --- a/rust-plugins/src/output/mod.rs +++ b/rust-plugins/src/output/mod.rs @@ -6,8 +6,8 @@ use crate::compute::Parser; use crate::compute::ast::ExprResult; use crate::generic::{Perfdata, Status}; use crate::snmp::SnmpResult; -use log::error; use serde::Deserialize; +use tracing::error; /// Configurable status messages and separators for plugin output. #[derive(Deserialize, Debug)] diff --git a/rust-plugins/src/snmp/mod.rs b/rust-plugins/src/snmp/mod.rs index 0c3645b138..797c351e81 100644 --- a/rust-plugins/src/snmp/mod.rs +++ b/rust-plugins/src/snmp/mod.rs @@ -9,7 +9,6 @@ //! OCTET STRING, OBJECT IDENTIFIER, IpAddress, Counter32/64, Gauge32/Unsigned32, //! TimeTicks, Opaque) are decoded; see `value_from_varbind`. -extern crate log; extern crate rasn; extern crate rasn_smi; extern crate rasn_snmp; @@ -28,7 +27,6 @@ use crate::generic::error::Error::RequestTimeout; use crate::generic::error::Error::SnmpAgentError; use crate::generic::error::Error::WalkTooLarge; use crate::generic::error::Result; -use log::{trace, warn}; use rasn::types::ObjectIdentifier; use rasn_smi::v2::{ApplicationSyntax, ObjectSyntax, SimpleSyntax}; use rasn_snmp::v2::BulkPdu; @@ -44,6 +42,7 @@ use std::collections::HashMap; use std::convert::TryInto; use std::net::UdpSocket; use std::time::{Duration, Instant}; +use tracing::{debug_span, trace, trace_span, warn}; /// Maximum size of a UDP datagram; SNMP bulk responses can be large, /// a smaller buffer would silently truncate them and break BER decoding. @@ -291,6 +290,7 @@ pub fn snmp_bulk_get<'a>( oid_list: &Vec<&str>, names: &Vec<&str>, ) -> Result { + let _span = debug_span!("get", oids = oid_list.len()).entered(); let mut oids_tab: Vec> = vec![]; for oid_str in oid_list { oids_tab.push(oid_to_vec(oid_str)?); @@ -346,6 +346,7 @@ pub fn snmp_bulk_walk<'a>( snmp_name: &str, max_repetitions: u32, ) -> Result { + let _span = debug_span!("walk", oid).entered(); let oid_init = oid_to_vec(oid)?; let mut oid_tab = oid_init.clone(); let mut retval = SnmpResult::new(HashMap::new()); @@ -400,6 +401,7 @@ pub fn snmp_bulk_walk_with_labels<'a>( labels: &'a HashMap, max_repetitions: u32, ) -> Result { + let _span = debug_span!("walk", oid).entered(); let oid_init = oid_to_vec(oid)?; let mut oid_tab = oid_init.clone(); let mut retval = SnmpResult::new(HashMap::new()); @@ -722,6 +724,7 @@ fn send_request( socket: &UdpSocket, buf: &mut [u8], ) -> Result> { + let _span = trace_span!("request", id = request_id).entered(); let encoded: Vec = rasn::der::encode(message).map_err(|_| InvalidSnmpPduEncode {})?; let attempts = config.retries + 1; From dff4638fc569c7a912427806a567bed1a3da0727 Mon Sep 17 00:00:00 2001 From: Julien Mathis Date: Fri, 4 Sep 2026 15:21:39 +0200 Subject: [PATCH 6/6] feat(rust-plugins): rate/delta DSL primitive with persistent state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SNMP counters (ifInOctets, ...) are monotonically increasing values; turning them into per-second rates requires the value and timestamp of the previous run. Adds a "rate": true field on a collect entry: { "name": "if", "oid": "1.3.6.1.2.1.2.2.1", "query": "Walk", "labels": { "1.2": "descr", "1.10": "in", "1.16": "out" }, "rate": true } Every numeric value of the entry becomes a per-second rate computed against the previous run; string columns (labels) are left untouched and vectors stay aligned. Design: - Identity = full OID, never the table position — a new interface appearing between runs cannot shift its neighbors' rates. - State files (new src/state.rs module): mode 0600 + atomic write (temp file + rename + fsync) — never world-readable, never half-written. A read failure (missing/corrupt file) is never fatal (fresh start); a write failure IS fatal (a stale reference would silently produce wrong rates forever). - 32-bit counter wraparound corrected; a 64-bit decrease or a missing previous instance is treated as a reset and yields one aligned 0.0 sample rather than desynchronizing the vector. - First run: "OK: Buffer creation", exit 0 (Perl parity). - New --statefile-dir CLI option (Perl parity, default /var/lib/centreon/centplugins). - Instrumented with state_read/state_write spans (visible via PLUGIN_LOG=debug or --trace-file, see the tracing PR). SnmpResult gains a `samples: Vec<(item_key, full_oid, value)>` field, populated only when a collect entry asks for rates (capture_samples threaded through the walk/get functions and process_response), so apply_rate can rebuild each numeric vector in original push order while keying state persistence by OID. Test plan: - cargo build --release / cargo test: 84 passed, 0 failed (9 new: wraparound, reset, dt<=0, state roundtrip + 0600 assert + corrupt-file recovery, two-run rate integration, new-instance alignment, sample capture) - --check-format validates the new example (examples/new-traffic-rate.json) - Manual: state files created under a temp statefile-dir are mode 0600 and survive a corrupt-content injection without crashing (exercised by the state.rs unit tests directly against the real filesystem) --- rust-plugins/examples/new-traffic-rate.json | 35 +++ rust-plugins/src/generic/error.rs | 7 + rust-plugins/src/generic/mod.rs | 222 ++++++++++++++++-- rust-plugins/src/main.rs | 11 + rust-plugins/src/snmp/mod.rs | 106 ++++++++- rust-plugins/src/state.rs | 248 ++++++++++++++++++++ 6 files changed, 602 insertions(+), 27 deletions(-) create mode 100644 rust-plugins/examples/new-traffic-rate.json create mode 100644 rust-plugins/src/state.rs diff --git a/rust-plugins/examples/new-traffic-rate.json b/rust-plugins/examples/new-traffic-rate.json new file mode 100644 index 0000000000..8e3f4e3a1a --- /dev/null +++ b/rust-plugins/examples/new-traffic-rate.json @@ -0,0 +1,35 @@ +{ + "collect": { + "snmp": [ + { + "name": "if", + "oid": "1.3.6.1.2.1.2.2.1", + "query": "Walk", + "labels": { + "1.2": "descr", + "1.10": "in", + "1.16": "out" + }, + "rate": true + } + ] + }, + "compute": { + "metrics": [ + { + "name": "traffic.in.bytespersecond", + "value": "{if.in}", + "prefix": "{if.descr}", + "uom": "B/s", + "threshold-suffix": "in" + }, + { + "name": "traffic.out.bytespersecond", + "value": "{if.out}", + "prefix": "{if.descr}", + "uom": "B/s", + "threshold-suffix": "out" + } + ] + } +} diff --git a/rust-plugins/src/generic/error.rs b/rust-plugins/src/generic/error.rs index d5da270b6d..d8d16eadd7 100644 --- a/rust-plugins/src/generic/error.rs +++ b/rust-plugins/src/generic/error.rs @@ -83,6 +83,13 @@ pub enum Error { #[snafu(display("SNMP collection exceeded the global timeout of {}s", seconds))] CollectTimeout { seconds: u64 }, + #[snafu(display( + "Could not persist the state file {} ({}): rates would be computed against a stale reference", + path, + reason + ))] + StatefileWrite { path: String, reason: String }, + #[snafu(display( "No valid SNMP response from {} after {} attempts (timeout {}s per attempt)", url, diff --git a/rust-plugins/src/generic/mod.rs b/rust-plugins/src/generic/mod.rs index 7b9e2b3732..62ad50a283 100644 --- a/rust-plugins/src/generic/mod.rs +++ b/rust-plugins/src/generic/mod.rs @@ -16,6 +16,7 @@ use crate::compute::{Compute, Parser, ast::ExprResult, threshold::Threshold}; use crate::output::{Output, OutputFormatter}; use crate::snmp::SnmpResult; use crate::snmp::{SnmpConfig, snmp_bulk_get, snmp_bulk_walk, snmp_bulk_walk_with_labels}; +use crate::state::{Snapshot, StateStore, compute_rate}; use regex::Regex; use serde::Deserialize; use std::collections::HashMap; @@ -112,6 +113,70 @@ fn worst(a: Status, b: Status) -> Status { if a.severity() > b.severity() { a } else { b } } +/// Current Unix time in seconds (never fails after 1970). +fn unix_now() -> f64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0) +} + +/// Replaces the collected counter values of a rate entry with per-second +/// rates, using the snapshot persisted by the previous run. +/// +/// The state is keyed by full OID: the identity of an instance across runs +/// is its OID, never its position in the table. String columns (labels such +/// as `ifDescr`) are left untouched — only numeric samples are transformed, +/// so vectors stay aligned. +/// +/// # Returns +/// `Ok(true)` when there was no previous snapshot (first run): the caller +/// reports `OK: Buffer creation` and exits 0, Perl-style. +fn apply_rate( + result: &mut SnmpResult, + entry: &Snmp, + config: &SnmpConfig, + now: f64, +) -> Result { + let store = StateStore::new(&config.statefile_dir); + let key = StateStore::rate_key(&config.target, &entry.name, &entry.oid); + let previous = store.load(&key); + + let mut values = HashMap::new(); + for (_, oid, value) in &result.samples { + values.insert(oid.clone(), *value); + } + store.save( + &key, + &Snapshot { + timestamp: now, + values, + }, + )?; + + let Some(previous) = previous else { + return Ok(true); + }; + let dt = now - previous.timestamp; + + // Rebuild each numeric vector in sample (= push) order. A missing + // previous instance, a counter reset or dt <= 0 yields one aligned 0.0 + // sample: dropping it would desalign the column from its siblings. + let mut rates: HashMap> = HashMap::new(); + for (item_key, oid, value) in &result.samples { + let rate = previous + .values + .get(oid) + .and_then(|old| compute_rate(*old, *value, dt)) + .unwrap_or(0.0); + rates.entry(item_key.clone()).or_default().push(rate); + } + for (item_key, vector) in rates { + result.items.insert(item_key, ExprResult::Vector(vector)); + } + Ok(false) +} + /// Type of SNMP query to perform for a given OID. #[derive(Deserialize, Debug)] enum QueryType { @@ -136,6 +201,14 @@ pub struct Snmp { /// (defaults to the global value, see the `--maxrepetitions` CLI option). #[serde(rename = "max-repetitions")] max_repetitions: Option, + /// When `true`, every numeric value of this entry is converted into a + /// per-second rate using the previous run (state persisted under + /// `--statefile-dir`, keyed by full OID). First run: the plugin + /// reports `OK: Buffer creation` and exits 0, Perl-style. 32-bit + /// counter wraparound is corrected; a counter reset or a missing + /// previous instance yields one aligned 0.0 sample. + #[serde(default)] + rate: bool, } /// Groups all SNMP queries that must be executed before computing metrics. @@ -288,13 +361,17 @@ impl Command { } /// Executes all configured SNMP queries (Get and Walk operations) and returns the results. + /// Returns the collected results, plus `true` when at least one rate + /// entry had no previous state (first run): the caller then reports + /// `OK: Buffer creation` instead of computing metrics. fn execute_snmp_collect( &self, config: &SnmpConfig, check_format: bool, - ) -> Result> { + ) -> Result<(Vec, bool)> { let _span = info_span!("collect").entered(); let mut collect: Vec = Vec::new(); + let mut buffer_creation = false; // Single deadline for ALL queries of this collection: the global // time budget covers the sum of the walks and gets, not each one. let deadline = config.deadline(); @@ -317,7 +394,7 @@ impl Command { } collect.push(SnmpResult::new(items)); } - return Ok(collect); + return Ok((collect, buffer_creation)); } let mut to_get = Vec::new(); let mut get_name = Vec::new(); @@ -325,25 +402,40 @@ impl Command { match s.query { QueryType::Walk => { let max_repetitions = s.max_repetitions.unwrap_or(config.max_repetitions); - if let Some(lab) = &s.labels { - let r = snmp_bulk_walk_with_labels( + let mut r = if let Some(lab) = &s.labels { + snmp_bulk_walk_with_labels( config, deadline, &s.oid, &s.name, lab, max_repetitions, - )?; - if !r.items.is_empty() { - collect.push(r); - } + s.rate, + )? } else { - let r = snmp_bulk_walk(config, deadline, &s.oid, &s.name, max_repetitions)?; - if !r.items.is_empty() { - collect.push(r); - } + snmp_bulk_walk(config, deadline, &s.oid, &s.name, max_repetitions, s.rate)? + }; + if s.rate { + buffer_creation |= apply_rate(&mut r, s, config, unix_now())?; + } + if !r.items.is_empty() { + collect.push(r); } } + // Rate entries need their samples isolated per entry, so + // they are queried individually instead of joining the + // batched get below. + QueryType::Get if s.rate => { + let mut r = snmp_bulk_get( + config, + deadline, + &vec![s.oid.as_str()], + &vec![s.name.as_str()], + true, + )?; + buffer_creation |= apply_rate(&mut r, s, config, unix_now())?; + collect.push(r); + } QueryType::Get => { to_get.push(s.oid.as_str()); get_name.push(s.name.as_str()); @@ -352,13 +444,13 @@ impl Command { } if !to_get.is_empty() { - let r = snmp_bulk_get(config, deadline, &to_get, &get_name); + let r = snmp_bulk_get(config, deadline, &to_get, &get_name, false); collect.push(r?); } if collect.is_empty() { return Err(error::Error::EmptyResponse {}); } - Ok(collect) + Ok((collect, buffer_creation)) } /// Executes the complete plugin pipeline: SNMP collection, metric computation, filtering, and output formatting. @@ -383,7 +475,16 @@ impl Command { no_data_status: Status, ) -> Result { let _check_span = info_span!("check").entered(); - let mut collect = self.execute_snmp_collect(config, check_format)?; + let (mut collect, buffer_creation) = self.execute_snmp_collect(config, check_format)?; + + if buffer_creation { + // First run of a rate entry: no reference to compute rates from. + // Perl-style behavior: report the buffer build and exit OK. + return Ok(CmdResult { + status: Status::Ok, + output: "OK: Buffer creation".to_string(), + }); + } if check_response { return self.format_raw_response(&collect); @@ -789,3 +890,94 @@ mod tests { )); } } + +#[cfg(test)] +mod rate_tests { + use super::*; + use crate::snmp::SnmpResult; + + fn test_setup(dir_tag: &str) -> (SnmpConfig, Snmp) { + let dir = + std::env::temp_dir().join(format!("rate-test-{}-{}", dir_tag, std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let config = SnmpConfig { + target: "10.0.0.1:161".to_string(), + version: "2c".to_string(), + community: "public".to_string(), + timeout: std::time::Duration::from_secs(1), + retries: 0, + collect_timeout: std::time::Duration::from_secs(5), + max_repetitions: 10, + statefile_dir: dir, + }; + let entry: Snmp = serde_json::from_str( + r#"{"name":"if","oid":"1.3.6.1.2.1.2.2.1","query":"Walk","rate":true}"#, + ) + .expect("valid entry JSON"); + (config, entry) + } + + fn result_with(samples: Vec<(&str, &str, f64)>) -> SnmpResult { + let mut result = SnmpResult::new(HashMap::new()); + let mut vector = Vec::new(); + for (key, oid, value) in samples { + vector.push(value); + result + .samples + .push((key.to_string(), oid.to_string(), value)); + } + result + .items + .insert("if".to_string(), ExprResult::Vector(vector)); + result + } + + #[test] + fn rate_entries_turn_counters_into_rates_across_runs() { + let (config, entry) = test_setup("basic"); + + // Run 1: no previous state -> buffer creation. + let mut run1 = result_with(vec![("if", "oid.10.1", 1000.0), ("if", "oid.10.2", 2000.0)]); + let first = apply_rate(&mut run1, &entry, &config, 100.0).expect("run 1"); + assert!(first, "first run must report buffer creation"); + + // Run 2, 10s later, +600 and +1200 octets -> 60 and 120 per second. + let mut run2 = result_with(vec![("if", "oid.10.1", 1600.0), ("if", "oid.10.2", 3200.0)]); + let first = apply_rate(&mut run2, &entry, &config, 110.0).expect("run 2"); + assert!(!first); + match run2.items.get("if") { + Some(ExprResult::Vector(v)) => assert_eq!(v, &vec![60.0, 120.0]), + other => panic!("expected rate Vector([60, 120]), got {:?}", other), + } + + // Run 3: a NEW instance appears -> one aligned 0.0 sample, the known + // instances keep their rates. + let mut run3 = result_with(vec![ + ("if", "oid.10.1", 2200.0), + ("if", "oid.10.2", 4400.0), + ("if", "oid.10.3", 500.0), + ]); + let first = apply_rate(&mut run3, &entry, &config, 120.0).expect("run 3"); + assert!(!first); + match run3.items.get("if") { + Some(ExprResult::Vector(v)) => assert_eq!(v, &vec![60.0, 120.0, 0.0]), + other => panic!("expected Vector([60, 120, 0]), got {:?}", other), + } + + let _ = std::fs::remove_dir_all(&config.statefile_dir); + } + + #[test] + fn same_timestamp_yields_aligned_zeroes_not_a_division_by_zero() { + let (config, entry) = test_setup("dtzero"); + let mut run1 = result_with(vec![("if", "oid.10.1", 1000.0)]); + apply_rate(&mut run1, &entry, &config, 100.0).expect("run 1"); + let mut run2 = result_with(vec![("if", "oid.10.1", 1600.0)]); + apply_rate(&mut run2, &entry, &config, 100.0).expect("run 2"); + match run2.items.get("if") { + Some(ExprResult::Vector(v)) => assert_eq!(v, &vec![0.0]), + other => panic!("expected Vector([0.0]), got {:?}", other), + } + let _ = std::fs::remove_dir_all(&config.statefile_dir); + } +} diff --git a/rust-plugins/src/main.rs b/rust-plugins/src/main.rs index c6ba56eddc..ed3da8e25f 100644 --- a/rust-plugins/src/main.rs +++ b/rust-plugins/src/main.rs @@ -23,6 +23,7 @@ mod compute; mod generic; mod output; mod snmp; +mod state; use generic::Command; use generic::Status; @@ -140,6 +141,8 @@ fn snmp_plugin() -> Result { let mut collect_timeout_secs: u64 = 50; // GetBulk max-repetitions (Perl parity: --maxrepetitions, default 50). let mut max_repetitions: u32 = 50; + // Directory for rate/delta state files (Perl parity: --statefile-dir). + let mut statefile_dir = "/var/lib/centreon/centplugins".to_string(); let mut filter_in = Vec::new(); let mut filter_out = Vec::new(); let mut no_data_status = Status::Unknown; @@ -213,6 +216,9 @@ fn snmp_plugin() -> Result { println!(" --snmp-retries Retries after a timed-out attempt (default: 2)"); println!(" --collect-timeout Global time budget for the whole collection (default: 50)"); println!(" --maxrepetitions GetBulk max-repetitions (default: 50)"); + println!( + " --statefile-dir Directory for rate/delta state files (default: /var/lib/centreon/centplugins)" + ); println!(" --warning- Warning threshold for metric"); println!(" --critical- Critical threshold for metric"); println!(" --check-format Check JSON file validity and exit"); @@ -238,6 +244,10 @@ fn snmp_plugin() -> Result { max_repetitions = parser.value()?.parse::()?; trace!("max_repetitions: {}", max_repetitions); } + Long("statefile-dir") => { + statefile_dir = parser.value()?.into_string()?; + trace!("statefile_dir: {}", statefile_dir); + } Long("trace-file") => { // Already consumed by the pre-scan in init_tracing; // swallowed here so lexopt does not reject it. @@ -335,6 +345,7 @@ fn snmp_plugin() -> Result { retries: snmp_retries, collect_timeout: std::time::Duration::from_secs(collect_timeout_secs), max_repetitions, + statefile_dir: std::path::PathBuf::from(statefile_dir), }; let result = cmd.execute( diff --git a/rust-plugins/src/snmp/mod.rs b/rust-plugins/src/snmp/mod.rs index 797c351e81..4d2b99e347 100644 --- a/rust-plugins/src/snmp/mod.rs +++ b/rust-plugins/src/snmp/mod.rs @@ -78,6 +78,9 @@ pub struct SnmpConfig { /// `--maxrepetitions`, default 50). Higher values mean fewer network /// round-trips when walking large tables. pub max_repetitions: u32, + /// Directory where rate/delta state files are stored (mirror of the + /// Perl `--statefile-dir`). + pub statefile_dir: std::path::PathBuf, } impl SnmpConfig { @@ -262,6 +265,10 @@ pub struct SnmpResult { /// Number of in-subtree variable bindings processed by this walk, /// checked against [`MAX_WALK_VARBINDS`]. processed: usize, + /// Numeric samples captured for rate computation, in push order: + /// `(item key, full OID, value)`. Only filled when the collect entry + /// asked for rates. + pub samples: Vec<(String, String, f64)>, } /// Retrieves the exact values of multiple OIDs in a single `GetRequest`. @@ -289,6 +296,7 @@ pub fn snmp_bulk_get<'a>( deadline: Instant, oid_list: &Vec<&str>, names: &Vec<&str>, + capture_samples: bool, ) -> Result { let _span = debug_span!("get", oids = oid_list.len()).entered(); let mut oids_tab: Vec> = vec![]; @@ -321,7 +329,8 @@ pub fn snmp_bulk_get<'a>( let mut buf = vec![0u8; UDP_BUFFER_SIZE]; let decoded = send_request(config, deadline, request_id, &message, &socket, &mut buf)?; - let _completed = retval.build_response_with_names(decoded, "", names, false)?; + let _completed = + retval.build_response_with_names(decoded, "", names, false, capture_samples)?; Ok(retval) } @@ -345,6 +354,7 @@ pub fn snmp_bulk_walk<'a>( oid: &str, snmp_name: &str, max_repetitions: u32, + capture_samples: bool, ) -> Result { let _span = debug_span!("walk", oid).entered(); let oid_init = oid_to_vec(oid)?; @@ -367,7 +377,7 @@ pub fn snmp_bulk_walk<'a>( // never be mistaken for the current one. request_id = request_id.wrapping_add(1); - let completed = retval.build_response(decoded, oid, snmp_name, true)?; + let completed = retval.build_response(decoded, oid, snmp_name, true, capture_samples)?; if completed { break; @@ -400,6 +410,7 @@ pub fn snmp_bulk_walk_with_labels<'a>( snmp_name: &str, labels: &'a HashMap, max_repetitions: u32, + capture_samples: bool, ) -> Result { let _span = debug_span!("walk", oid).entered(); let oid_init = oid_to_vec(oid)?; @@ -424,7 +435,14 @@ pub fn snmp_bulk_walk_with_labels<'a>( // never be mistaken for the current one. request_id = request_id.wrapping_add(1); - let completed = retval.build_response_with_labels(decoded, oid, snmp_name, labels, true)?; + let completed = retval.build_response_with_labels( + decoded, + oid, + snmp_name, + labels, + true, + capture_samples, + )?; if completed { break; } @@ -440,6 +458,7 @@ impl SnmpResult { items, last_oid: Vec::new(), processed: 0, + samples: Vec::new(), } } @@ -500,6 +519,7 @@ impl SnmpResult { decoded: Message, oid: &str, walk: bool, + capture_samples: bool, mut key_for: impl FnMut(usize, &str) -> Vec, ) -> Result { let mut completed = false; @@ -544,8 +564,16 @@ impl SnmpResult { let Some(typ) = value_from_varbind(&var.value)? else { continue; }; + let numeric = match &typ { + ValueType::Integer(i) => Some(*i as f64), + ValueType::Counter64(c) => Some(*c as f64), + ValueType::String(_) => None, + }; for key in key_for(idx, &name) { + if capture_samples && let Some(value) = numeric { + self.samples.push((key.clone(), name.clone(), value)); + } _ = self.store(key, typ.clone())?; } } @@ -563,8 +591,9 @@ impl SnmpResult { snmp_name: &str, labels: &'a HashMap, walk: bool, + capture_samples: bool, ) -> Result { - self.process_response(decoded, oid, walk, |_idx, name| { + self.process_response(decoded, oid, walk, capture_samples, |_idx, name| { let prefix = name.rfind('.').map_or(name, |i| &name[..i]); labels .iter() @@ -582,8 +611,9 @@ impl SnmpResult { oid: &str, names: &Vec<&str>, walk: bool, + capture_samples: bool, ) -> Result { - self.process_response(decoded, oid, walk, |idx, _name| { + self.process_response(decoded, oid, walk, capture_samples, |idx, _name| { vec![names[idx].to_string()] }) } @@ -596,8 +626,9 @@ impl SnmpResult { oid: &str, snmp_name: &str, walk: bool, + capture_samples: bool, ) -> Result { - self.process_response(decoded, oid, walk, |_idx, _name| { + self.process_response(decoded, oid, walk, capture_samples, |_idx, _name| { vec![snmp_name.to_string()] }) } @@ -913,6 +944,7 @@ mod tests { retries: 2, collect_timeout: Duration::from_secs(50), max_repetitions: 50, + statefile_dir: std::env::temp_dir(), } } @@ -926,6 +958,7 @@ mod tests { CPU_TABLE_OID, "cpu", config.max_repetitions, + false, ) .unwrap(); @@ -946,6 +979,7 @@ mod tests { SHORT_TABLE_OID, "short", config.max_repetitions, + false, ) .unwrap(); @@ -961,6 +995,7 @@ mod tests { TRANSPORT_ERROR_OID, "x", config.max_repetitions, + false, ); assert!(result.is_err()); } @@ -974,6 +1009,7 @@ mod tests { config.deadline(), &vec!["1.3.6.1.2.1.1.3.0", "1.3.6.1.2.1.1.5.0"], &vec!["uptime", "name"], + false, ) .unwrap(); @@ -992,11 +1028,12 @@ mod tests { config.deadline(), &vec![TRANSPORT_ERROR_OID], &vec!["x"], + false, ); assert!(result.is_err()); // propagates invalid-oid errors before any network call - let result = snmp_bulk_get(&config, config.deadline(), &vec![""], &vec!["x"]); + let result = snmp_bulk_get(&config, config.deadline(), &vec![""], &vec!["x"], false); assert!(result.is_err()); } @@ -1014,6 +1051,7 @@ mod tests { "cpu", &labels, config.max_repetitions, + false, ) .unwrap(); match result.items.get("cpu.core").unwrap() { @@ -1036,6 +1074,7 @@ mod tests { "short", &labels, config.max_repetitions, + false, ) .unwrap(); @@ -1052,6 +1091,7 @@ mod tests { "x", &labels, config.max_repetitions, + false, ); assert!(result.is_err()); @@ -1063,6 +1103,7 @@ mod tests { "x", &labels, config.max_repetitions, + false, ); assert!(result.is_err()); } @@ -1136,13 +1177,13 @@ mod tests { // forever without this guard. let msg = response_message(vec![("1.3.6.1.2.5", 1), ("1.3.6.1.2.4", 2)]); let mut result = SnmpResult::new(HashMap::new()); - let err = result.build_response(msg, "1.3.6.1.2", "v", true); + let err = result.build_response(msg, "1.3.6.1.2", "v", true, false); assert!(err.is_err(), "backwards OID must be an error"); // An OID equal to the previous one must fail too. let msg = response_message(vec![("1.3.6.1.2.5", 1), ("1.3.6.1.2.5", 2)]); let mut result = SnmpResult::new(HashMap::new()); - let err = result.build_response(msg, "1.3.6.1.2", "v", true); + let err = result.build_response(msg, "1.3.6.1.2", "v", true, false); assert!(err.is_err(), "repeated OID must be an error"); } @@ -1155,7 +1196,7 @@ mod tests { .collect(); let msg = response_message(bindings.iter().map(|(oid, v)| (oid.as_str(), *v)).collect()); let mut result = SnmpResult::new(HashMap::new()); - let err = result.build_response(msg, "1.3.6.1.2", "v", true); + let err = result.build_response(msg, "1.3.6.1.2", "v", true, false); match err { Err(crate::generic::error::Error::WalkTooLarge { max }) => { assert_eq!(max, MAX_WALK_VARBINDS) @@ -1402,7 +1443,7 @@ mod tests { let decoded = response_message(vec![("1.3.6.1.2.1.1.3.0", 1), ("1.3.6.1.2.1.1.9.0", 2)]); let completed = result - .build_response_with_names(decoded, "", &vec!["uptime", "count"], false) + .build_response_with_names(decoded, "", &vec!["uptime", "count"], false, false) .unwrap(); assert!(!completed); @@ -1429,7 +1470,14 @@ mod tests { labels.insert("16".to_string(), "out".to_string()); let completed = result - .build_response_with_labels(decoded, "1.3.6.1.2.1.2.2.1", "iface", &labels, false) + .build_response_with_labels( + decoded, + "1.3.6.1.2.1.2.2.1", + "iface", + &labels, + false, + false, + ) .unwrap(); assert!(!completed); @@ -1442,4 +1490,38 @@ mod tests { &ExprResult::Vector(vec![200.0]) ); } + + #[test] + fn capture_records_numeric_samples_with_their_oid() { + let decoded = Message { + version: 1.into(), + community: "public".as_bytes().into(), + data: Pdus::Response(Response(Pdu { + request_id: 1, + error_status: 0, + error_index: 0, + variable_bindings: vec![ + integer_varbind("1.3.6.1.2.10.1", 41), + VarBind { + name: ObjectIdentifier::new_unchecked( + oid_to_vec("1.3.6.1.2.2.1").unwrap().into(), + ), + value: VarBindValue::Value(ObjectSyntax::Simple(SimpleSyntax::String( + OctetString::from_static(b"eth0"), + ))), + }, + ], + })), + }; + let mut result = SnmpResult::new(HashMap::new()); + let names = vec!["traffic", "descr"]; + result + .build_response_with_names(decoded, "", &names, false, true) + .expect("build_response should succeed"); + // Only the numeric varbind is sampled, with its full OID. + assert_eq!( + result.samples, + vec![("traffic".to_string(), "1.3.6.1.2.10.1".to_string(), 41.0)] + ); + } } diff --git a/rust-plugins/src/state.rs b/rust-plugins/src/state.rs new file mode 100644 index 0000000000..f229a269ab --- /dev/null +++ b/rust-plugins/src/state.rs @@ -0,0 +1,248 @@ +// +// Copyright 2026-Present Centreon (http://www.centreon.com/) +// +// Centreon is a full-fledged industry-strength solution that meets +// the needs in IT infrastructure and application monitoring for +// service performance. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +//! Persistent state between two plugin executions, for delta/rate metrics. +//! +//! SNMP counters (`ifInOctets`, ...) are monotonically increasing values: +//! turning them into rates (B/s, packets/s) requires the value and timestamp +//! of the previous run. This module stores one small JSON snapshot per +//! collect entry, keyed by target + entry name + OID. +//! +//! Security posture (lesson from the Perl audit, S5): state files are +//! created with mode `0600` and written atomically (temp file + rename) — +//! never world-readable, never half-written. + +use crate::generic::error::Error::StatefileWrite; +use crate::generic::error::Result; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use tracing::{debug_span, warn}; + +/// Largest value a 32-bit SNMP counter can hold. +const U32_COUNTER_MAX: f64 = 4_294_967_295.0; + +/// One persisted collection snapshot: the values of a collect entry, +/// keyed by full OID, plus the collection timestamp. +#[derive(Debug, Serialize, Deserialize, PartialEq)] +pub struct Snapshot { + /// Unix epoch of the collection, in seconds. + pub timestamp: f64, + /// Collected numeric values, keyed by full OID (stable identity of an + /// instance across runs — positions in a table are not). + pub values: HashMap, +} + +/// Reads and writes [`Snapshot`]s under a state directory. +pub struct StateStore { + dir: PathBuf, +} + +impl StateStore { + /// Creates a store rooted at `dir` (mirror of the Perl + /// `--statefile-dir`, default `/var/lib/centreon/centplugins`). + pub fn new(dir: &Path) -> StateStore { + StateStore { + dir: dir.to_path_buf(), + } + } + + /// Builds the state key for a collect entry: same target + same entry + /// name + same base OID share the same state, which is the intended + /// behavior for a check re-executed by the poller. + pub fn rate_key(target: &str, entry_name: &str, oid: &str) -> String { + let raw = format!("rust-snmp_{}_{}_{}", target, entry_name, oid); + raw.chars() + .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' }) + .collect() + } + + fn path_for(&self, key: &str) -> PathBuf { + self.dir.join(format!("{}.json", key)) + } + + /// Loads the previous snapshot for `key`. + /// + /// Read problems are never fatal: a missing, unreadable or corrupt file + /// is treated as "no previous run" (the plugin then rebuilds its buffer) + /// — a poller must not go UNKNOWN because a cache file was damaged. + pub fn load(&self, key: &str) -> Option { + let _span = debug_span!("state_read", key).entered(); + let path = self.path_for(key); + let content = match std::fs::read_to_string(&path) { + Ok(content) => content, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return None, + Err(e) => { + warn!( + "unreadable state file {:?} ({}), rebuilding buffer", + path, e + ); + return None; + } + }; + match serde_json::from_str(&content) { + Ok(snapshot) => Some(snapshot), + Err(e) => { + warn!("corrupt state file {:?} ({}), rebuilding buffer", path, e); + None + } + } + } + + /// Persists `snapshot` under `key`, atomically and in mode `0600`. + /// + /// A write failure IS fatal: without a persisted reference the next run + /// would compute rates against a stale base — better a clean UNKNOWN + /// now than silently wrong values forever. + pub fn save(&self, key: &str, snapshot: &Snapshot) -> Result<()> { + let _span = debug_span!("state_write", key).entered(); + let path = self.path_for(key); + let write = || -> std::io::Result<()> { + std::fs::create_dir_all(&self.dir)?; + let tmp = self.dir.join(format!("{}.tmp.{}", key, std::process::id())); + { + use std::io::Write; + let mut options = std::fs::OpenOptions::new(); + options.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + // Never world-readable: state may carry data an operator + // considers sensitive (audit S5). + options.mode(0o600); + } + let mut file = options.open(&tmp)?; + file.write_all( + serde_json::to_string(snapshot) + .expect("Snapshot serialization cannot fail") + .as_bytes(), + )?; + file.sync_all()?; + } + // Atomic replacement: a concurrent reader sees either the old or + // the new snapshot, never a truncated file. + std::fs::rename(&tmp, &path) + }; + write().map_err(|e| StatefileWrite { + path: path.display().to_string(), + reason: e.to_string(), + }) + } +} + +/// Converts a counter pair into a per-second rate. +/// +/// * `dt <= 0` → `None` (two collections at the same instant, or clock gone +/// backwards: no meaningful rate). +/// * decreasing value → 32-bit wraparound correction when the previous value +/// still fit in 32 bits; otherwise the counter was reset (agent reboot) +/// and `None` is returned. Note the classic blind spot, shared with the +/// Perl implementation: a *reset* of a 32-bit counter is indistinguishable +/// from a wrap and produces one over-estimated sample. +pub fn compute_rate(old: f64, new: f64, dt: f64) -> Option { + if dt <= 0.0 { + return None; + } + let mut delta = new - old; + if delta < 0.0 { + if old <= U32_COUNTER_MAX { + delta += U32_COUNTER_MAX + 1.0; + } + if delta < 0.0 { + return None; + } + } + Some(delta / dt) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rate_of_a_normal_increase() { + assert_eq!(compute_rate(1000.0, 1600.0, 60.0), Some(10.0)); + } + + #[test] + fn rate_needs_a_positive_time_delta() { + assert_eq!(compute_rate(1000.0, 1600.0, 0.0), None); + assert_eq!(compute_rate(1000.0, 1600.0, -5.0), None); + } + + #[test] + fn rate_corrects_a_32bit_wraparound() { + // old close to 2^32, new wrapped to a small value. + let old = 4_294_967_290.0; + let new = 10.0; + // delta = 10 - 4294967290 + 4294967296 = 16 + assert_eq!(compute_rate(old, new, 4.0), Some(4.0)); + } + + #[test] + fn rate_treats_a_64bit_decrease_as_a_reset() { + // old beyond 32 bits: a decrease cannot be a 32-bit wrap. + let old = 10_000_000_000.0; + assert_eq!(compute_rate(old, 5.0, 60.0), None); + } + + #[test] + fn state_roundtrip_is_atomic_and_private() { + let dir = std::env::temp_dir().join(format!("state-test-{}", std::process::id())); + let store = StateStore::new(&dir); + let key = StateStore::rate_key("127.0.0.1:161", "if", "1.3.6.1.2.1.2.2.1"); + assert!( + key.chars().all(|c| c.is_ascii_alphanumeric() || c == '_'), + "key must be filesystem-safe: {}", + key + ); + + assert!(store.load(&key).is_none(), "no state on first run"); + + let snapshot = Snapshot { + timestamp: 1000.0, + values: HashMap::from([("1.3.6.1.2.1.2.2.1.10.1".to_string(), 42.0)]), + }; + store.save(&key, &snapshot).expect("save should succeed"); + assert_eq!(store.load(&key), Some(snapshot)); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(dir.join(format!("{}.json", key))) + .expect("state file exists") + .permissions() + .mode(); + assert_eq!(mode & 0o777, 0o600, "state file must be private (0600)"); + } + + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn corrupt_state_is_a_fresh_start_not_a_crash() { + let dir = std::env::temp_dir().join(format!("state-corrupt-{}", std::process::id())); + std::fs::create_dir_all(&dir).expect("mkdir"); + std::fs::write(dir.join("k.json"), b"{ not json").expect("write"); + let store = StateStore::new(&dir); + assert!(store.load("k").is_none()); + let _ = std::fs::remove_dir_all(&dir); + } +}