feat(health)!: bound OTLP sink queues

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

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,

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,

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<K: Eq + Hash, V> {
     ready: VecDeque<K>,
 }
 
+/// 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<K: Eq + Hash + Clone, V> {
     state: Mutex<QueueState<K, V>>,
+    capacity: Option<NonZeroUsize>,
+    bounded_len: AtomicUsize,
     notify: Notify,
 }
 
@@ -44,26 +62,65 @@ impl<K: Eq + Hash + Clone, V> DedupQueue<K, V> {
                 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<K: Eq + Hash + Clone, V> DedupQueue<K, V> {
         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<K: Eq + Hash + Clone, V> DedupQueue<K, V> {
     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::<String, i32>::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::<String, i32>::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)));
+    }
 }