From 4d383ac3c2d750a3543824c1bf847205ca1714e2 Mon Sep 17 00:00:00 2001 From: Jay Zhu Date: Mon, 17 Aug 2026 21:54:50 -0600 Subject: [PATCH] feat(health)!: bound OTLP export queues Signed-off-by: Jay Zhu --- crates/health/example/config.example.toml | 6 + crates/health/src/config.rs | 43 +++++ crates/health/src/otlp/mod.rs | 31 ++++ crates/health/src/sink/dedup_queue.rs | 111 +++++++++++- crates/health/src/sink/otlp.rs | 207 ++++++++++++++++++---- docs/observability/core_metrics.md | 2 + 6 files changed, 359 insertions(+), 41 deletions(-) diff --git a/crates/health/example/config.example.toml b/crates/health/example/config.example.toml index dabeb2d97e..533716b180 100644 --- a/crates/health/example/config.example.toml +++ b/crates/health/example/config.example.toml @@ -112,6 +112,11 @@ enabled = false [[sinks.otlp.targets]] endpoint = "http://localhost:4317" batch_size = 512 +# Maximum entries waiting in each of this target's independent log and metric +# queues. The oldest entry is dropped when a full queue receives a new event +# identity. Must be greater than zero. Default: 32768. Changes require a service +# restart. +queue_capacity = 32768 flush_interval = "2s" # Include Redfish diagnostic payload fields in OTLP logs for this target. @@ -132,6 +137,7 @@ include_alert_details = false # [[sinks.otlp.targets]] # endpoint = "https://telemetry.example.com:4317" # batch_size = 1024 +# queue_capacity = 32768 # flush_interval = "5s" # include_diagnostics = true # include_alert_details = true diff --git a/crates/health/src/config.rs b/crates/health/src/config.rs index d48b6ad820..e8c61b77ff 100644 --- a/crates/health/src/config.rs +++ b/crates/health/src/config.rs @@ -541,6 +541,15 @@ pub struct OtlpTargetConfig { #[serde(default = "OtlpTargetConfig::default_batch_size")] pub batch_size: usize, + /// Maximum number of entries waiting in each signal queue for this target. + /// + /// Logs and metrics have independent queues of this size. When a queue is + /// full, inserting a new identity drops its oldest entry. The active export + /// batch is bounded separately by `batch_size`. Defaults to 32,768 and must + /// be greater than zero. + #[serde(default = "OtlpTargetConfig::default_queue_capacity")] + pub queue_capacity: usize, + /// Maximum time to wait before flushing a non-empty batch for either /// signal. Defaults to two seconds. #[serde( @@ -569,10 +578,16 @@ pub struct OtlpTargetConfig { } impl OtlpTargetConfig { + pub(crate) const DEFAULT_QUEUE_CAPACITY: usize = 32_768; + fn default_batch_size() -> usize { 512 } + fn default_queue_capacity() -> usize { + Self::DEFAULT_QUEUE_CAPACITY + } + fn default_flush_interval() -> std::time::Duration { std::time::Duration::from_secs(2) } @@ -584,6 +599,10 @@ impl OtlpTargetConfig { return Err(format!("{path}.batch_size must be greater than 0")); } + if self.queue_capacity == 0 { + return Err(format!("{path}.queue_capacity must be greater than 0")); + } + if self.flush_interval.is_zero() { return Err(format!("{path}.flush_interval must be greater than 0")); } @@ -2249,6 +2268,7 @@ mod tests { endpoint: endpoint.to_string(), tls: None, batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, @@ -2993,6 +3013,7 @@ endpoint = "https://site.example:4317" [[targets]] endpoint = "https://central.example:4317" batch_size = 1024 +queue_capacity = 2048 flush_interval = "5s" include_diagnostics = true include_alert_details = true @@ -3013,8 +3034,15 @@ reload_interval = "30s" assert_eq!(targets.len(), 2); assert!(targets[0].tls.is_none()); assert_eq!(targets[0].batch_size, 512); + + assert_eq!( + targets[0].queue_capacity, + OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY + ); + assert_eq!(targets[0].flush_interval, Duration::from_secs(2)); assert_eq!(targets[1].batch_size, 1024); + assert_eq!(targets[1].queue_capacity, 2048); assert_eq!(targets[1].flush_interval, Duration::from_secs(5)); assert!(targets[1].include_diagnostics); assert!(!targets[0].include_alert_details); @@ -3080,6 +3108,16 @@ reload_interval = "30s" "sinks.otlp.targets[2].batch_size must be greater than 0".to_string() ), + IndexedOtlpTarget { + index: 2, + target: OtlpTargetConfig { + queue_capacity: 0, + ..otlp_target("http://site.example:4317") + }, + } => FailsWith( + "sinks.otlp.targets[2].queue_capacity must be greater than 0".to_string() + ), + IndexedOtlpTarget { index: 2, target: OtlpTargetConfig { @@ -3252,6 +3290,7 @@ reload_interval = "30s" targets: vec![OtlpTargetConfig { endpoint: "http://localhost:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, @@ -3290,6 +3329,7 @@ reload_interval = "30s" targets: vec![OtlpTargetConfig { endpoint: "http://localhost:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: true, include_alert_details: false, @@ -3308,6 +3348,7 @@ reload_interval = "30s" OtlpTargetConfig { endpoint: "http://site.example:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, @@ -3316,6 +3357,7 @@ reload_interval = "30s" OtlpTargetConfig { endpoint: "http://central.example:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: true, include_alert_details: false, @@ -3362,6 +3404,7 @@ reload_interval = "30s" targets: vec![OtlpTargetConfig { endpoint: "http://localhost:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, diff --git a/crates/health/src/otlp/mod.rs b/crates/health/src/otlp/mod.rs index 603959e0e9..64f3262338 100644 --- a/crates/health/src/otlp/mod.rs +++ b/crates/health/src/otlp/mod.rs @@ -22,6 +22,7 @@ pub mod metrics_drain; use std::time::Duration; use carbide_instrument::LabelValue; +use opentelemetry::StringValue; use tonic::transport::{Channel, ClientTlsConfig, Endpoint}; use crate::HealthError; @@ -37,6 +38,34 @@ pub(crate) enum OtlpSignal { Metrics, } +/// An OTLP endpoint selected from the finite target list loaded at startup. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(crate) struct ConfiguredOtlpTarget(pub(crate) String); + +impl LabelValue for ConfiguredOtlpTarget { + fn label_value(&self) -> StringValue { + self.0.clone().into() + } +} + +/// An OTLP queue dropped its oldest entry to admit a new identity. +#[derive(carbide_instrument::Event)] +#[event( + event_name = "otlp_queue_entry_dropped", + metric_name = "carbide_health_otlp_queue_dropped_total", + component = "nico-hardware-health", + log = off, + metric = counter, + message = "otlp queue dropped its oldest entry", + describe = "Number of OTLP queue entries dropped because a per-target queue reached capacity, by target and signal." +)] +pub(crate) struct OtlpQueueEntryDropped { + #[label] + pub(crate) target: ConfiguredOtlpTarget, + #[label] + pub(crate) signal: OtlpSignal, +} + /// Builds an OTLP endpoint with the target's current TLS or mTLS policy. /// /// HTTPS targets without an explicit TLS profile use platform trust roots. An @@ -224,6 +253,7 @@ mod tests { endpoint: format!("https://{address}"), tls: None, batch_size: 1, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(1), include_diagnostics: false, include_alert_details: false, @@ -264,6 +294,7 @@ mod tests { endpoint: format!("https://{address}"), tls: None, batch_size: 1, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: Duration::from_secs(1), include_diagnostics: false, include_alert_details: false, diff --git a/crates/health/src/sink/dedup_queue.rs b/crates/health/src/sink/dedup_queue.rs index bf83a55360..4ba6cf728b 100644 --- a/crates/health/src/sink/dedup_queue.rs +++ b/crates/health/src/sink/dedup_queue.rs @@ -18,12 +18,15 @@ //! Latest-wins dedup queue. //! //! Generic queue keyed by `K` that replaces the value when the same key -//! is pushed again. Used by health report sinks (keyed by machine/rack +//! is pushed again. A bounded queue evicts the oldest distinct key when a new +//! key arrives at capacity. Used by health report sinks (keyed by machine/rack //! + report source) and OtlpSink (keyed by event type identity string). use std::collections::{HashMap, VecDeque}; use std::hash::Hash; +use std::num::NonZeroUsize; use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::sync::Notify; @@ -32,8 +35,23 @@ struct QueueState { ready: VecDeque, } +/// Result of saving a value by its deduplication key. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum SaveOutcome { + /// A new key was added without evicting another entry. + Inserted, + + /// The value for an existing key was replaced in its current queue position. + Replaced, + + /// A new key was added after evicting the oldest distinct key. + DroppedOldest, +} + pub(crate) struct DedupQueue { state: Mutex>, + capacity: Option, + bounded_len: AtomicUsize, notify: Notify, } @@ -44,26 +62,65 @@ impl DedupQueue { values: HashMap::new(), ready: VecDeque::new(), }), + capacity: None, + bounded_len: AtomicUsize::new(0), + notify: Notify::new(), + } + } + + /// Creates a queue limited to `capacity` distinct keys. + pub(super) fn bounded(capacity: NonZeroUsize) -> Self { + Self { + state: Mutex::new(QueueState { + values: HashMap::new(), + ready: VecDeque::new(), + }), + capacity: Some(capacity), + bounded_len: AtomicUsize::new(0), notify: Notify::new(), } } - /// returns true if an existing value was replaced - pub(super) fn save_latest(&self, key: K, value: V) -> bool { - let replaced; + /// Saves the latest value for `key` and wakes a waiting consumer. + /// + /// Replacing an existing key preserves its queue position. When a bounded + /// queue is full, inserting a new key evicts the oldest distinct key. + pub(super) fn save_latest(&self, key: K, value: V) -> SaveOutcome { + let outcome; + { let mut state = self.state.lock().expect("dedup queue mutex poisoned"); if state.values.contains_key(&key) { state.values.insert(key, value); - replaced = true; + + outcome = SaveOutcome::Replaced; } else { + outcome = if self + .capacity + .is_some_and(|capacity| capacity.get() == state.values.len()) + { + if let Some(oldest) = state.ready.pop_front() { + state.values.remove(&oldest); + SaveOutcome::DroppedOldest + } else { + SaveOutcome::Inserted + } + } else { + SaveOutcome::Inserted + }; + state.values.insert(key.clone(), value); state.ready.push_back(key); - replaced = false; + } + + if outcome == SaveOutcome::Inserted && self.capacity.is_some() { + self.bounded_len + .store(state.values.len(), Ordering::Release); } } + self.notify.notify_one(); - replaced + outcome } pub(super) async fn next(&self) -> (K, V) { @@ -79,6 +136,11 @@ impl DedupQueue { let mut state = self.state.lock().expect("dedup queue mutex poisoned"); while let Some(key) = state.ready.pop_front() { if let Some(value) = state.values.remove(&key) { + if self.capacity.is_some() { + self.bounded_len + .store(state.values.len(), Ordering::Release); + } + return Some((key, value)); } } @@ -88,6 +150,13 @@ impl DedupQueue { pub(crate) async fn notified(&self) { self.notify.notified().await; } + + /// Returns a lock-free snapshot of the number of entries in a bounded queue. + /// + /// Unbounded queues do not track their depth and return zero. + pub(super) fn len(&self) -> usize { + self.bounded_len.load(Ordering::Acquire) + } } #[cfg(test)] @@ -148,4 +217,32 @@ mod tests { assert_eq!(val_a, 99); assert_eq!(queue.pop().unwrap().0, "b"); } + + #[test] + fn bounded_queue_drops_oldest_distinct_key_and_tracks_depth() { + let queue = DedupQueue::::bounded(NonZeroUsize::new(2).unwrap()); + + assert_eq!(queue.save_latest("a".into(), 1), SaveOutcome::Inserted); + assert_eq!(queue.save_latest("b".into(), 2), SaveOutcome::Inserted); + assert_eq!(queue.save_latest("c".into(), 3), SaveOutcome::DroppedOldest); + + assert_eq!(queue.len(), 2); + assert_eq!(queue.pop(), Some(("b".to_string(), 2))); + assert_eq!(queue.len(), 1); + assert_eq!(queue.pop(), Some(("c".to_string(), 3))); + assert_eq!(queue.len(), 0); + } + + #[test] + fn bounded_queue_replacement_does_not_evict() { + let queue = DedupQueue::::bounded(NonZeroUsize::new(2).unwrap()); + + queue.save_latest("a".into(), 1); + queue.save_latest("b".into(), 2); + + assert_eq!(queue.save_latest("a".into(), 3), SaveOutcome::Replaced); + assert_eq!(queue.len(), 2); + assert_eq!(queue.pop(), Some(("a".to_string(), 3))); + assert_eq!(queue.pop(), Some(("b".to_string(), 2))); + } } diff --git a/crates/health/src/sink/otlp.rs b/crates/health/src/sink/otlp.rs index 0298e459ef..2fb1d13a10 100644 --- a/crates/health/src/sink/otlp.rs +++ b/crates/health/src/sink/otlp.rs @@ -15,11 +15,14 @@ * limitations under the License. */ +use std::num::NonZeroUsize; use std::sync::Arc; +use carbide_instrument::emit; +use opentelemetry::KeyValue; use prometheus::{Counter, CounterVec, Opts}; -use super::dedup_queue::DedupQueue; +use super::dedup_queue::{DedupQueue, SaveOutcome}; use super::event_mapper::RedfishEventMapper; use super::{CollectorEvent, DataSink, EventContext, LogRecord, MetricSample}; use crate::HealthError; @@ -27,6 +30,7 @@ use crate::config::OtlpTargetConfig; use crate::metrics::MetricsManager; use crate::otlp::drain::OtlpDrainTask; use crate::otlp::metrics_drain::OtlpMetricsDrainTask; +use crate::otlp::{ConfiguredOtlpTarget, OtlpQueueEntryDropped, OtlpSignal}; pub(crate) type OtlpQueue = DedupQueue; pub(crate) type OtlpMetricsQueue = DedupQueue; @@ -47,6 +51,7 @@ pub(crate) struct OtlpSink { metrics_queue: Arc, replaced_total: Counter, metrics_replaced_total: Counter, + target: ConfiguredOtlpTarget, mapper: Arc, include_diagnostics: bool, } @@ -57,6 +62,7 @@ pub struct OtlpSink { metrics_queue: Arc, replaced_total: Counter, metrics_replaced_total: Counter, + target: ConfiguredOtlpTarget, mapper: Arc, include_diagnostics: bool, } @@ -87,8 +93,7 @@ impl OtlpSink { /// Creates one independently queued sink for each configured OTLP target. /// /// The returned order matches `configs`. Each sink starts separate log and - /// metric drain tasks, and its queue replacement counters use its position - /// in `configs` as the bounded `target_index` label. + /// metric drain tasks. Queue metrics identify the configured endpoint. /// /// # Errors /// @@ -107,9 +112,9 @@ impl OtlpSink { let replaced_total = CounterVec::new( Opts::new( format!("{prefix}_otlp_sink_replaced_total"), - "total log events replaced in the otlp queue before drain could process them, labeled by configured target index", + "total log events replaced in the otlp queue before drain could process them, labeled by target", ), - &["target_index"], + &["target"], )?; metrics_manager @@ -119,9 +124,9 @@ impl OtlpSink { let metrics_replaced_total = CounterVec::new( Opts::new( format!("{prefix}_otlp_sink_metrics_replaced_total"), - "total metric samples replaced in the otlp queue before drain could process them, labeled by configured target index", + "total metric samples replaced in the otlp queue before drain could process them, labeled by target", ), - &["target_index"], + &["target"], )?; metrics_manager @@ -130,14 +135,21 @@ impl OtlpSink { let mut sinks = Vec::with_capacity(configs.len()); - for (target_index, config) in configs.iter().enumerate() { - let queue: Arc = Arc::new(DedupQueue::new()); - let metrics_queue: Arc = Arc::new(DedupQueue::new()); - let target_index = target_index.to_string(); - let replaced_total = replaced_total.get_metric_with_label_values(&[&target_index])?; + for config in configs { + let capacity = NonZeroUsize::new(config.queue_capacity).ok_or_else(|| { + HealthError::GenericError("otlp queue capacity must be greater than 0".to_string()) + })?; + + let queue: Arc = Arc::new(DedupQueue::bounded(capacity)); + let metrics_queue: Arc = Arc::new(DedupQueue::bounded(capacity)); + + let target = ConfiguredOtlpTarget(config.endpoint.clone()); + + let replaced_total = + replaced_total.get_metric_with_label_values(&[&config.endpoint])?; let metrics_replaced_total = - metrics_replaced_total.get_metric_with_label_values(&[&target_index])?; + metrics_replaced_total.get_metric_with_label_values(&[&config.endpoint])?; let drain = OtlpDrainTask::new(queue.clone(), config.clone()); handle.spawn(drain.run()); @@ -157,14 +169,32 @@ impl OtlpSink { metrics_queue, replaced_total, metrics_replaced_total, + target, mapper: mapper.clone(), include_diagnostics: config.include_diagnostics, }); } + register_queue_depth_metric(&sinks); + Ok(sinks) } + /// Records the observable result of one log or metric queue insertion. + fn record_save_outcome(&self, outcome: SaveOutcome, signal: OtlpSignal) { + match outcome { + SaveOutcome::Inserted => {} + SaveOutcome::Replaced => match signal { + OtlpSignal::Logs => self.replaced_total.inc(), + OtlpSignal::Metrics => self.metrics_replaced_total.inc(), + }, + SaveOutcome::DroppedOldest => emit(OtlpQueueEntryDropped { + target: self.target.clone(), + signal, + }), + } + } + /// Enqueues the emitted log record using the parent event identity. fn enqueue_log_event(&self, context: &EventContext, record: &LogRecord) { let record = record.emitted_log_record(self.include_diagnostics); @@ -175,9 +205,8 @@ impl OtlpSink { let event = CollectorEvent::Log(Box::new(record.into_owned())); - if self.queue.save_latest(key, (context.clone(), event)) { - self.replaced_total.inc(); - } + let outcome = self.queue.save_latest(key, (context.clone(), event)); + self.record_save_outcome(outcome, OtlpSignal::Logs); } } @@ -195,11 +224,17 @@ impl OtlpSink { mapper: Arc, include_diagnostics: bool, ) -> Self { + let capacity = match NonZeroUsize::new(OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY) { + Some(capacity) => capacity, + None => NonZeroUsize::MIN, + }; + Self { - queue: Arc::new(DedupQueue::new()), - metrics_queue: Arc::new(DedupQueue::new()), + queue: Arc::new(DedupQueue::bounded(capacity)), + metrics_queue: Arc::new(DedupQueue::bounded(capacity)), replaced_total: Counter::new("bench_replaced", "bench").unwrap(), metrics_replaced_total: Counter::new("bench_metrics_replaced", "bench").unwrap(), + target: ConfiguredOtlpTarget("bench".to_string()), mapper, include_diagnostics, } @@ -230,12 +265,11 @@ impl DataSink for OtlpSink { if let CollectorEvent::Metric(sample) = event { let key = metric_queue_key(context, sample); - if self + let outcome = self .metrics_queue - .save_latest(key, (context.clone(), (**sample).clone())) - { - self.metrics_replaced_total.inc(); - } + .save_latest(key, (context.clone(), (**sample).clone())); + + self.record_save_outcome(outcome, OtlpSignal::Metrics); return Ok(()); } @@ -265,19 +299,65 @@ impl DataSink for OtlpSink { _ => return Ok(()), }; - if self.queue.save_latest(key, (context.clone(), event)) { - self.replaced_total.inc(); - } + let outcome = self.queue.save_latest(key, (context.clone(), event)); + self.record_save_outcome(outcome, OtlpSignal::Logs); Ok(()) } } +/// Registers queue-depth observations for every configured OTLP target. +/// +/// The callback runs only when OpenTelemetry collects metrics. It reads an +/// atomic depth snapshot without locking or scanning either queue. Weak +/// references avoid extending a sink or queue lifetime solely for metrics. +fn register_queue_depth_metric(sinks: &[OtlpSink]) { + let queues = sinks + .iter() + .map(|sink| { + ( + sink.target.0.clone(), + Arc::downgrade(&sink.queue), + Arc::downgrade(&sink.metrics_queue), + ) + }) + .collect::>(); + + opentelemetry::global::meter("carbide-health") + .u64_observable_gauge("carbide_health_otlp_queue_depth") + .with_description("Number of entries waiting in an OTLP queue, by target and signal.") + .with_callback(move |observer| { + for (target, logs, metrics) in &queues { + if let Some(logs) = logs.upgrade() { + observer.observe( + logs.len() as u64, + &[ + KeyValue::new("target", target.clone()), + KeyValue::new("signal", "logs"), + ], + ); + } + + if let Some(metrics) = metrics.upgrade() { + observer.observe( + metrics.len() as u64, + &[ + KeyValue::new("target", target.clone()), + KeyValue::new("signal", "metrics"), + ], + ); + } + } + }) + .build(); +} + #[cfg(test)] mod tests { use std::borrow::Cow; use std::str::FromStr; + use carbide_instrument::testing::MetricsCapture; use mac_address::MacAddress; use super::*; @@ -392,6 +472,17 @@ mod tests { OtlpSink::new_for_bench_with_diagnostics(Arc::new(OpenBmcEventMapper), false) } + fn bounded_test_sink(capacity: usize) -> OtlpSink { + let mut sink = test_sink(); + let capacity = NonZeroUsize::new(capacity).expect("test capacity must be nonzero"); + + sink.queue = Arc::new(DedupQueue::bounded(capacity)); + sink.metrics_queue = Arc::new(DedupQueue::bounded(capacity)); + sink.target = ConfiguredOtlpTarget("http://bounded.example:4317".to_string()); + + sink + } + #[test] fn is_otlp_log_relevant_excludes_metric_events() { assert!(!is_otlp_log_relevant(&metric_event())); @@ -527,6 +618,53 @@ mod tests { assert!(sink.queue.pop().is_some()); } + #[test] + fn bounded_signal_queues_drop_oldest_and_report_depth() { + let metrics = MetricsCapture::start(); + let sink = bounded_test_sink(1); + let context = test_context(); + + register_queue_depth_metric(std::slice::from_ref(&sink)); + + sink.handle_event(&context, &log_event("OpenBMC.0.1.First", "[]")); + sink.handle_event(&context, &log_event("OpenBMC.0.1.Second", "[]")); + + sink.handle_event( + &context, + &metric_event_with_name("first", "a", "gauge", "state"), + ); + + sink.handle_event( + &context, + &metric_event_with_name("second", "b", "gauge", "state"), + ); + + let target = "http://bounded.example:4317"; + + for signal in ["logs", "metrics"] { + assert_eq!( + metrics.counter_delta( + "carbide_health_otlp_queue_dropped_total", + &[("target", target), ("signal", signal)], + ), + 1.0, + "{signal} queue should report one eviction", + ); + + assert_eq!( + metrics.gauge_value( + "carbide_health_otlp_queue_depth", + &[("target", target), ("signal", signal)], + ), + 1.0, + "{signal} queue should remain at capacity", + ); + } + + assert_eq!(sink.queue.len(), 1); + assert_eq!(sink.metrics_queue.len(), 1); + } + #[test] fn composite_fans_log_event_to_each_otlp_target_queue() { let first = Arc::new(test_sink()); @@ -549,11 +687,12 @@ mod tests { } #[tokio::test] - async fn new_many_labels_replacement_counters_by_target_index() { + async fn new_many_labels_replacement_counters_by_target() { let configs = vec![ OtlpTargetConfig { endpoint: "http://first.example:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: std::time::Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, @@ -562,6 +701,7 @@ mod tests { OtlpTargetConfig { endpoint: "http://second.example:4317".to_string(), batch_size: 512, + queue_capacity: OtlpTargetConfig::DEFAULT_QUEUE_CAPACITY, flush_interval: std::time::Duration::from_secs(2), include_diagnostics: false, include_alert_details: false, @@ -587,22 +727,21 @@ mod tests { .export_metrics() .expect("metrics should export"); - assert!( - metrics - .contains("otlp_replacement_test_otlp_sink_replaced_total{target_index=\"0\"} 1") - ); + assert!(metrics.contains( + "otlp_replacement_test_otlp_sink_replaced_total{target=\"http://first.example:4317\"} 1" + )); assert!( metrics - .contains("otlp_replacement_test_otlp_sink_replaced_total{target_index=\"1\"} 0") + .contains("otlp_replacement_test_otlp_sink_replaced_total{target=\"http://second.example:4317\"} 0") ); assert!(metrics.contains( - "otlp_replacement_test_otlp_sink_metrics_replaced_total{target_index=\"0\"} 0" + "otlp_replacement_test_otlp_sink_metrics_replaced_total{target=\"http://first.example:4317\"} 0" )); assert!(metrics.contains( - "otlp_replacement_test_otlp_sink_metrics_replaced_total{target_index=\"1\"} 1" + "otlp_replacement_test_otlp_sink_metrics_replaced_total{target=\"http://second.example:4317\"} 1" )); } diff --git a/docs/observability/core_metrics.md b/docs/observability/core_metrics.md index cae896af6a..a643e29b15 100644 --- a/docs/observability/core_metrics.md +++ b/docs/observability/core_metrics.md @@ -106,6 +106,8 @@ This file contains a list of metrics exported by NVIDIA Infra Controller (NICo). carbide_gpus_total_countgaugeNumber of GPUs in the NICo deployment carbide_gpus_usable_countgaugeNumber of remaining GPUs in the NICo deployment available for immediate instance creation carbide_health_otlp_export_failures_totalcounterNumber of OTLP export batches dropped after a send failure, by signal and gRPC status code. +carbide_health_otlp_queue_depthgaugeNumber of entries waiting in an OTLP queue, by target and signal. +carbide_health_otlp_queue_dropped_totalcounterNumber of OTLP queue entries dropped because a per-target queue reached capacity, by target and signal. carbide_health_redfish_sse_event_record_resolution_failures_totalcounterNumber of Redfish SSE event records dropped after a referenced record could not be resolved, by failure reason. carbide_health_report_submissions_totalcounterNumber of health report submissions to the NICo API, by report target and outcome. carbide_host_reprovision_retries_totalcounterNumber of times a failed host firmware upgrade was retried during host reprovisioning