From 437b813924cbd7f9e98595a2da7e7a556b1d34c6 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 18:59:53 +0200 Subject: [PATCH 1/5] fix(runtime): cap tracked inhibition sources with max_inhibit_sources --- crates/rsigma-cli/src/daemon/metrics.rs | 19 ++++ .../src/alert_pipeline/config.rs | 25 +++- .../src/alert_pipeline/inhibit.rs | 107 ++++++++++++++++-- .../rsigma-runtime/src/alert_pipeline/mod.rs | 54 +++++++-- crates/rsigma-runtime/src/metrics.rs | 3 + 5 files changed, 184 insertions(+), 24 deletions(-) diff --git a/crates/rsigma-cli/src/daemon/metrics.rs b/crates/rsigma-cli/src/daemon/metrics.rs index 910272fcb..1aa75d7c4 100644 --- a/crates/rsigma-cli/src/daemon/metrics.rs +++ b/crates/rsigma-cli/src/daemon/metrics.rs @@ -86,6 +86,7 @@ pub struct Metrics { pub silences_active: IntGauge, pub inhibited_total: IntCounterVec, pub inhibit_sources_active: IntGauge, + pub inhibit_sources_refused_total: IntCounterVec, pub rule_false_positive_ratio: GaugeVec, pub dispositions_total: IntCounterVec, pub disposition_ingest_total: IntCounterVec, @@ -852,12 +853,23 @@ impl Metrics { )) .unwrap(); inhibit_sources_active.set(0); + let inhibit_sources_refused_total = IntCounterVec::new( + Opts::new( + "rsigma_inhibit_sources_refused_total", + "Inhibition sources not tracked because max_inhibit_sources was reached, by rule name", + ), + &["rule"], + ) + .unwrap(); registry .register(Box::new(inhibited_total.clone())) .unwrap(); registry .register(Box::new(inhibit_sources_active.clone())) .unwrap(); + registry + .register(Box::new(inhibit_sources_refused_total.clone())) + .unwrap(); // Triage feedback loop (#70). The gauge and the per-rule disposition // counter label on `rule_title` to line up with @@ -1225,6 +1237,7 @@ impl Metrics { silences_active, inhibited_total, inhibit_sources_active, + inhibit_sources_refused_total, rule_false_positive_ratio, dispositions_total, disposition_ingest_total, @@ -1427,6 +1440,12 @@ impl MetricsHook for Metrics { self.inhibit_sources_active.set(count); } + fn on_inhibit_source_refused(&self, rule: &str) { + self.inhibit_sources_refused_total + .with_label_values(&[rule]) + .inc(); + } + fn on_capture_event_dropped(&self, reason: &str) { self.capture_events_dropped_total .with_label_values(&[reason]) diff --git a/crates/rsigma-runtime/src/alert_pipeline/config.rs b/crates/rsigma-runtime/src/alert_pipeline/config.rs index 67b30e863..e4eb3d6ad 100644 --- a/crates/rsigma-runtime/src/alert_pipeline/config.rs +++ b/crates/rsigma-runtime/src/alert_pipeline/config.rs @@ -47,6 +47,11 @@ const DEFAULT_MAX_ACTIVE_ALERTS: usize = 100_000; /// surface against unbounded silence creation. pub const DEFAULT_MAX_DYNAMIC_SILENCES: usize = 1_000; +/// Default ceiling on tracked inhibition sources across all rules. Once full, +/// results that would open a new source are not tracked, bounding memory when +/// the `equal` selectors carry high-cardinality values. +const DEFAULT_MAX_INHIBIT_SOURCES: usize = 100_000; + /// Top-level alert-pipeline config file. /// /// ```yaml @@ -88,6 +93,10 @@ pub struct AlertPipelineFile { /// Inhibition rules. An active source mutes matching targets. #[serde(default)] pub inhibit_rules: Vec, + /// Ceiling on tracked inhibition sources across all rules. Past this, a + /// result that would open a new source is not tracked. Defaults to 100000. + #[serde(default)] + pub max_inhibit_sources: Option, } /// `inhibit_rules:` entry. @@ -349,7 +358,11 @@ pub fn build_alert_pipeline( let inhibit = if file.inhibit_rules.is_empty() { None } else { - Some(build_inhibit(file.inhibit_rules)?) + Some(build_inhibit( + file.inhibit_rules, + file.max_inhibit_sources + .unwrap_or(DEFAULT_MAX_INHIBIT_SOURCES), + )?) }; let max_silences = file.max_silences.unwrap_or(DEFAULT_MAX_DYNAMIC_SILENCES); @@ -366,7 +379,10 @@ pub fn build_alert_pipeline( } /// Validate the `inhibit_rules:` list into an [`InhibitConfig`]. -fn build_inhibit(rules: Vec) -> Result { +fn build_inhibit( + rules: Vec, + max_sources: usize, +) -> Result { let mut out = Vec::with_capacity(rules.len()); for (i, rule) in rules.into_iter().enumerate() { if rule.source_match.is_empty() || rule.target_match.is_empty() { @@ -388,7 +404,10 @@ fn build_inhibit(rules: Vec) -> Result, + /// Ceiling on tracked sources across all rules. At the cap, a result that + /// would open a new source is still evaluated as a target but is not + /// tracked; already-tracked sources keep refreshing. + pub max_sources: usize, } /// Active-source index: `(rule_name, equal_key) -> last_seen`. @@ -71,12 +75,14 @@ impl InhibitStore { /// Evaluate a result. If it is an inhibited target, return the inhibiting /// rule name (and do not record it as a source). Otherwise record it as a /// source for every rule whose `source_match` it satisfies, and return - /// `None`. + /// `None`. A new source refused by `max_sources` is reported to + /// `on_refused` with its rule name. pub fn evaluate( &mut self, cfg: &InhibitConfig, result: &EvaluationResult, now: i64, + mut on_refused: impl FnMut(&str), ) -> Option { // 1. Inhibited target? Checked against sources from prior results. for rule in &cfg.rules { @@ -95,7 +101,13 @@ impl InhibitStore { for rule in &cfg.rules { if rule.source_match.matches(result) { let key = (rule.name.clone(), equal_key(&rule.equal, result)); - self.sources.insert(key, now); + if let Some(last) = self.sources.get_mut(&key) { + *last = now; + } else if self.sources.len() < cfg.max_sources { + self.sources.insert(key, now); + } else { + on_refused(&rule.name); + } } } None @@ -137,6 +149,25 @@ impl InhibitStore { } } + /// Evict the least recently seen sources beyond `max`, returning how many + /// were evicted. + pub(crate) fn enforce_cap(&mut self, max: usize) -> usize { + let excess = self.sources.len().saturating_sub(max); + if excess == 0 { + return 0; + } + let mut by_age: Vec<(i64, (String, String))> = self + .sources + .iter() + .map(|(key, &last_seen)| (last_seen, key.clone())) + .collect(); + by_age.sort_unstable(); + for (_, key) in by_age.into_iter().take(excess) { + self.sources.remove(&key); + } + excess + } + /// Count of currently-active sources. pub fn active_count(&self, cfg: &InhibitConfig, now: i64) -> usize { self.sources @@ -216,6 +247,7 @@ mod tests { equal: vec![Selector::parse("match.SourceIp").unwrap()], duration: Duration::from_secs(300), }], + max_sources: 100, } } @@ -225,17 +257,17 @@ mod tests { let mut store = InhibitStore::default(); // A critical source on 10.0.0.1 registers. assert_eq!( - store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0), + store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {}), None ); // A high target on the same IP is inhibited. assert_eq!( - store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1), + store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1, |_| {}), Some("critical-inhibits-high".to_string()) ); // A high target on a different IP is not inhibited. assert_eq!( - store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 2), + store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 2, |_| {}), None ); } @@ -244,10 +276,10 @@ mod tests { fn source_expires_after_duration() { let cfg = cfg(); let mut store = InhibitStore::default(); - store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0); + store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {}); // After the 300s window, the source no longer inhibits. assert_eq!( - store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 400), + store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 400, |_| {}), None ); } @@ -264,12 +296,13 @@ mod tests { equal: vec![Selector::parse("match.SourceIp").unwrap()], duration: Duration::from_secs(300), }], + max_sources: 100, }; let mut store = InhibitStore::default(); - store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 0); + store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 0, |_| {}); // The same kind of result is a source, not an inhibited target. assert_eq!( - store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1), + store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 1, |_| {}), None ); } @@ -278,10 +311,64 @@ mod tests { fn gc_and_active_count() { let cfg = cfg(); let mut store = InhibitStore::default(); - store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0); + store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 0, |_| {}); assert_eq!(store.active_count(&cfg, 100), 1); assert_eq!(store.active_count(&cfg, 400), 0, "past duration"); store.gc(&cfg, 400); assert!(store.is_empty()); } + + #[test] + fn new_sources_past_the_cap_are_refused() { + let cfg = InhibitConfig { + max_sources: 2, + ..cfg() + }; + let mut store = InhibitStore::default(); + let mut refused = Vec::new(); + for (i, ip) in ["10.0.0.1", "10.0.0.2", "10.0.0.3"].into_iter().enumerate() { + store.evaluate(&cfg, &detection(ip, Level::Critical), i as i64, |rule| { + refused.push(rule.to_string()) + }); + } + assert_eq!(refused, ["critical-inhibits-high"]); + assert_eq!(store.active_count(&cfg, 10), 2); + + // The refused source does not inhibit; tracked ones still do and + // still refresh at the cap. + assert_eq!( + store.evaluate(&cfg, &detection("10.0.0.3", Level::High), 10, |_| {}), + None + ); + store.evaluate(&cfg, &detection("10.0.0.1", Level::Critical), 200, |_| { + panic!("refreshing a tracked source is not a refusal") + }); + assert_eq!( + store.evaluate(&cfg, &detection("10.0.0.1", Level::High), 450, |_| {}), + Some("critical-inhibits-high".to_string()) + ); + } + + #[test] + fn enforce_cap_keeps_the_most_recently_seen() { + let cfg = cfg(); + let mut store = InhibitStore::default(); + for (ip, now) in [("10.0.0.1", 30), ("10.0.0.2", 10), ("10.0.0.3", 20)] { + store.evaluate(&cfg, &detection(ip, Level::Critical), now, |_| {}); + } + assert_eq!(store.enforce_cap(5), 0); + assert_eq!(store.enforce_cap(2), 1); + assert_eq!( + store.evaluate(&cfg, &detection("10.0.0.2", Level::High), 40, |_| {}), + None, + "the least recently seen source was evicted" + ); + for ip in ["10.0.0.1", "10.0.0.3"] { + assert!( + store + .evaluate(&cfg, &detection(ip, Level::High), 40, |_| {}) + .is_some() + ); + } + } } diff --git a/crates/rsigma-runtime/src/alert_pipeline/mod.rs b/crates/rsigma-runtime/src/alert_pipeline/mod.rs index 3e0a9457e..1134e99c3 100644 --- a/crates/rsigma-runtime/src/alert_pipeline/mod.rs +++ b/crates/rsigma-runtime/src/alert_pipeline/mod.rs @@ -217,7 +217,9 @@ impl AlertPipeline { // ones about to be silenced) as active sources, so a silenced // source still inhibits its targets. if let Some(icfg) = self.inhibit.as_ref() - && let Some(rule) = state.inhibit.evaluate(icfg, &result, now) + && let Some(rule) = state.inhibit.evaluate(icfg, &result, now, |rule| { + metrics.on_inhibit_source_refused(rule); + }) { metrics.on_alert_pipeline_inhibited(&rule); continue; @@ -426,15 +428,17 @@ impl AlertPipeline { } if let Some(icfg) = self.inhibit.as_ref() { state.inhibit.restore(snap.inhibit_sources, icfg, now); + trim.inhibit_sources = state.inhibit.enforce_cap(icfg.max_sources); } trim.log("restore"); true } /// Trim carried-over state to this pipeline's caps: the least recently - /// seen dedup alerts beyond `max_active_alerts` and incidents beyond - /// `max_open_incidents` are evicted without a `resolved` record, and each - /// incident is cut to the per-incident caps. Call it when this pipeline + /// seen dedup alerts beyond `max_active_alerts`, incidents beyond + /// `max_open_incidents`, and inhibition sources beyond + /// `max_inhibit_sources` are evicted (alerts and incidents without a + /// `resolved` record), and each incident is cut to the per-incident caps. Call it when this pipeline /// replaces another on hot-reload, while holding the state lock the /// pipeline processes under, so a lowered cap applies at once. /// [`restore`](Self::restore) applies the same trim. @@ -446,6 +450,9 @@ impl AlertPipeline { if let Some(g) = self.group.as_ref() { trim.incidents = state.incidents.enforce_caps(&g.caps); } + if let Some(icfg) = self.inhibit.as_ref() { + trim.inhibit_sources = state.inhibit.enforce_cap(icfg.max_sources); + } trim.log("reload"); trim } @@ -458,15 +465,18 @@ pub struct CapTrim { pub dedup_alerts: usize, /// Incidents evicted by `max_open_incidents`. pub incidents: usize, + /// Inhibition sources evicted by `max_inhibit_sources`. + pub inhibit_sources: usize, } impl CapTrim { fn log(&self, context: &str) { - if self.dedup_alerts > 0 || self.incidents > 0 { + if self.dedup_alerts > 0 || self.incidents > 0 || self.inhibit_sources > 0 { tracing::warn!( context, dedup_alerts = self.dedup_alerts, incidents = self.incidents, + inhibit_sources = self.inhibit_sources, "Evicted the least recently seen alert-pipeline state to fit the configured caps" ); } @@ -699,8 +709,17 @@ mod tests { ); } + const INHIBIT: &str = "inhibit_rules:\n - name: high\n source_match:\n - selector: level\n op: \"=\"\n value: high\n target_match:\n - selector: level\n op: \"=\"\n value: critical\n equal: [match.SourceIp]\n duration: 1h\n"; const UNCAPPED: &str = "dedup:\n fingerprint: [match.SourceIp]\n resolve_timeout: 1h\ngroup:\n by: [match.SourceIp]\n group_wait: 1h\n resolve_timeout: 1h\n"; - const CAPPED: &str = "dedup:\n fingerprint: [match.SourceIp]\n resolve_timeout: 1h\n max_active_alerts: 3\ngroup:\n by: [match.SourceIp]\n group_wait: 1h\n resolve_timeout: 1h\n caps:\n max_open_incidents: 3\n"; + const CAPPED: &str = "dedup:\n fingerprint: [match.SourceIp]\n resolve_timeout: 1h\n max_active_alerts: 3\ngroup:\n by: [match.SourceIp]\n group_wait: 1h\n resolve_timeout: 1h\n caps:\n max_open_incidents: 3\nmax_inhibit_sources: 3\n"; + + fn uncapped() -> AlertPipeline { + pipeline(&format!("{UNCAPPED}{INHIBIT}")) + } + + fn capped() -> AlertPipeline { + pipeline(&format!("{CAPPED}{INHIBIT}")) + } fn ip(i: i64) -> String { format!("10.0.0.{i}") @@ -708,12 +727,13 @@ mod tests { /// Six alerts and incidents, one per IP, with IP `i` last seen at `i`. fn six_alerts() -> AlertPipelineState { - let p = pipeline(UNCAPPED); + let p = uncapped(); let mut st = AlertPipelineState::default(); for i in 0..6 { run(&p, &ip(i), Level::High, &mut st, i); } assert_eq!((st.dedup.len(), st.incidents.len()), (6, 6)); + assert_eq!(st.inhibit.snapshot().len(), 6); st } @@ -728,6 +748,17 @@ mod tests { grouped.sort(); assert_eq!(grouped, [ip(3), ip(4), ip(5)]); assert_eq!(st.dedup.len(), 3); + let icfg = p.inhibit.as_ref().unwrap(); + assert_eq!(st.inhibit.active_count(icfg, 10), 3); + for i in 2..6 { + let target = detection(&ip(i), Level::Critical); + assert_eq!( + st.inhibit.evaluate(icfg, &target, 10, |_| {}).is_some(), + i >= 3, + "only the newest inhibition sources survive: {}", + ip(i) + ); + } for i in 3..6 { assert!( run(p, &ip(i), Level::High, st, 10).is_empty(), @@ -744,8 +775,8 @@ mod tests { #[test] fn restore_keeps_the_most_recent_entries_within_lowered_caps() { - let snap = pipeline(UNCAPPED).snapshot(&six_alerts()); - let capped = pipeline(CAPPED); + let snap = uncapped().snapshot(&six_alerts()); + let capped = capped(); let mut st = AlertPipelineState::default(); assert!(capped.restore(&mut st, snap, 10)); assert_keeps_newest_three(&capped, &mut st); @@ -754,12 +785,13 @@ mod tests { #[test] fn enforce_caps_trims_state_carried_into_a_reloaded_pipeline() { let mut st = six_alerts(); - let capped = pipeline(CAPPED); + let capped = capped(); assert_eq!( capped.enforce_caps(&mut st), CapTrim { dedup_alerts: 3, - incidents: 3 + incidents: 3, + inhibit_sources: 3, } ); assert_keeps_newest_three(&capped, &mut st); diff --git a/crates/rsigma-runtime/src/metrics.rs b/crates/rsigma-runtime/src/metrics.rs index 6976824f4..6470bc359 100644 --- a/crates/rsigma-runtime/src/metrics.rs +++ b/crates/rsigma-runtime/src/metrics.rs @@ -129,6 +129,9 @@ pub trait MetricsHook: Send + Sync { fn on_alert_pipeline_inhibited(&self, _rule: &str) {} /// Report the current number of active inhibition sources. fn set_inhibit_sources_active(&self, _count: i64) {} + /// A result that would have opened a new inhibition source for `rule` was + /// not tracked because `max_inhibit_sources` was reached. + fn on_inhibit_source_refused(&self, _rule: &str) {} /// A risk-annotation outcome. `action` is one of `scored` / `no_entity` / /// `skipped` (out of scope). From 940198e8bcbb5ec4ba51e3241dd9bd96a3ca4dd7 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 18:59:53 +0200 Subject: [PATCH 2/5] feat(eval): warn at load about long correlation windows with no group cap --- .../rsigma-eval/src/correlation_engine/mod.rs | 24 +++++++++++++++ .../src/correlation_engine/tests.rs | 29 +++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/crates/rsigma-eval/src/correlation_engine/mod.rs b/crates/rsigma-eval/src/correlation_engine/mod.rs index 34e608067..75513eac6 100644 --- a/crates/rsigma-eval/src/correlation_engine/mod.rs +++ b/crates/rsigma-eval/src/correlation_engine/mod.rs @@ -56,6 +56,11 @@ const SNAPSHOT_VERSION: u32 = 1; /// with a `WARN`. const MAX_CHAIN_DEPTH: usize = 10; +/// Timespan above which a correlation without a per-group entry cap is +/// warned about at load: a busy group's window retains `timespan` x event +/// rate entries, which `max_state_entries` does not see. +const LONG_WINDOW_WARN_SECS: u64 = 3600; + /// Stateful correlation engine. /// /// Wraps the stateless `Engine` for detection rules and adds time-windowed @@ -285,6 +290,15 @@ impl CorrelationEngine { self.apply_custom_attributes(&effective.custom_attributes); let compiled = compile_correlation(effective)?; + if self.has_unbounded_long_window(&compiled) { + log::warn!( + "Correlation '{}' has a {}s timespan and no per-group entry cap; \ + a busy group retains timespan x event rate entries. \ + Set max_group_entries (--max-group-entries) or the rsigma.max_group_entries attribute to bound it.", + compiled.id.as_deref().unwrap_or(&compiled.title), + compiled.timespan_secs + ); + } let idx = self.correlations.len(); // Index by each referenced rule ID/name @@ -310,6 +324,16 @@ impl CorrelationEngine { Ok(()) } + /// Whether `corr` keeps a window longer than [`LONG_WINDOW_WARN_SECS`] + /// with no per-group entry cap from the rule or the engine config. + fn has_unbounded_long_window(&self, corr: &CompiledCorrelation) -> bool { + corr.timespan_secs > LONG_WINDOW_WARN_SECS + && corr + .max_group_entries + .or(self.config.max_group_entries) + .is_none() + } + /// Add all rules and correlations from a parsed collection. /// /// Detection rules are added first (so they're available for correlation diff --git a/crates/rsigma-eval/src/correlation_engine/tests.rs b/crates/rsigma-eval/src/correlation_engine/tests.rs index 297693720..b59fd74c7 100644 --- a/crates/rsigma-eval/src/correlation_engine/tests.rs +++ b/crates/rsigma-eval/src/correlation_engine/tests.rs @@ -573,6 +573,35 @@ fn windowed_engine_with_group_cap(window_lines: &str, cap: usize) -> Correlation engine } +#[test] +fn test_long_window_without_group_cap_is_flagged() { + let flagged = |window_lines: &str, config: CorrelationConfig| { + let collection = parse_sigma_yaml(&window_mode_yaml(window_lines)).unwrap(); + let mut engine = CorrelationEngine::new(config); + engine.add_collection(&collection).unwrap(); + engine.has_unbounded_long_window(&engine.correlations[0]) + }; + assert!(flagged(" timespan: 2h", CorrelationConfig::default())); + assert!(!flagged(" timespan: 1h", CorrelationConfig::default())); + assert!(!flagged( + " timespan: 2h", + CorrelationConfig { + max_group_entries: Some(10_000), + ..Default::default() + } + )); + + let yaml = window_mode_yaml(" timespan: 24h").replace( + "id: corr-win\n", + "id: corr-win\nrsigma.max_group_entries: '10000'\n", + ); + let mut engine = CorrelationEngine::new(CorrelationConfig::default()); + engine + .add_collection(&parse_sigma_yaml(&yaml).unwrap()) + .unwrap(); + assert!(!engine.has_unbounded_long_window(&engine.correlations[0])); +} + #[test] fn test_max_group_entries_caps_event_count() { // Cap of 2 retained entries: the gte-3 threshold can never be met, From 7b4f04fe69eabf8fa948b5922093930d24c5ae6b Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 18:59:54 +0200 Subject: [PATCH 3/5] docs: describe the inhibit source cap and the long-window warning --- CHANGELOG.md | 6 ++++++ docs/content/cli/engine/daemon.md | 2 +- docs/content/guide/alert-pipeline.md | 7 +++++-- docs/content/guide/performance-tuning.md | 2 +- docs/content/reference/metrics.md | 1 + 5 files changed, 14 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3252ace78..3ca2f66ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to RSigma are documented in this file. Each entry correspond ## [Unreleased] +### The inhibit store is capped, and long unbounded correlation windows warn at load + +`InhibitStore` was the only alert-pipeline store without a count cap: sources were pruned by `duration` only, so `equal` selectors over attacker-influenced values (source IPs, usernames) could grow it without bound. A new top-level `max_inhibit_sources` key (default 100000) caps tracked sources across all `inhibit_rules`. At the cap, a result that would open a new source is still checked as a target but is not tracked, and the refusal is counted on the new `rsigma_inhibit_sources_refused_total{rule}` counter; already-tracked sources keep refreshing. Restore and hot-reload trim the store to the cap, evicting the least recently seen sources, and `CapTrim` reports them as `inhibit_sources`. For library users, `AlertPipelineFile` gains `max_inhibit_sources`, and `MetricsHook` gains `on_inhibit_source_refused`, which defaults to a no-op. + +`max_group_entries` stays unbounded by default, because a finite default would silently stop `event_count` rules whose threshold exceeds it. Instead, a correlation whose `timespan` is longer than one hour and that has no per-group cap, from `--max-group-entries` or `rsigma.max_group_entries`, now logs a warning at load, since a busy group's window grows with `timespan` x event rate. + ### Delivery failures without a DLQ sink are logged (#615) When a sink exhausted its retries and no `--dlq` was set, the daemon counted the dropped result on `rsigma_dlq_events_total` and logged it once, at `debug`, so every later dropped alert left no trace in the logs. Each drop now produces a `warn` log, rate-limited to one every 10 seconds; each warning carries the number of results dropped since the previous one and the last delivery error. A failed write to a configured DLQ sink is now counted on the new `rsigma_dlq_write_failures_total` counter, and the delivery layer and the daemon's DLQ bridge log a failure they could not queue instead of discarding it silently. The docs now say that a sink commits a failed result by handing it to the DLQ writer, so the source ack does not wait for the DLQ sink's write. diff --git a/docs/content/cli/engine/daemon.md b/docs/content/cli/engine/daemon.md index 667c4e16e..838fe4a0c 100644 --- a/docs/content/cli/engine/daemon.md +++ b/docs/content/cli/engine/daemon.md @@ -155,7 +155,7 @@ See [TLS deployment](../../reference/security.md#tls-termination-for-the-api-lis | `--correlation-event-mode ` | `none` | `none`, `full` (deflate-compressed full bodies), `refs` (timestamp + ID only). | | `--max-correlation-events ` | `10` | Cap on stored events per correlation window. | | `--max-state-entries ` | `100000` | Hard cap on correlation state entries across all correlations and group keys. When reached, the stalest entries are evicted to 90% capacity and a warning is logged. | -| `--max-group-entries ` | unset | Cap on retained entries within a single correlation group's window state. Bounds within-window growth of chatty groups; oldest entries are dropped (session windows keep their span anchor). Unset = unbounded. | +| `--max-group-entries ` | unset | Cap on retained entries within a single correlation group's window state. Bounds within-window growth of chatty groups; oldest entries are dropped (session windows keep their span anchor). Unset = unbounded, and a correlation with a `timespan` longer than one hour and no `rsigma.max_group_entries` logs a warning at load. {{ added "unreleased" }} | | `--timestamp-field ` | unset | Field name to prepend to the timestamp extraction list. Repeatable. | | `--timestamp-fallback ` | `wallclock` | Behavior when no timestamp is found: `wallclock` (use wall clock time) or `skip` (skip correlation state for that event). Use `skip` for forensic replay. | diff --git a/docs/content/guide/alert-pipeline.md b/docs/content/guide/alert-pipeline.md index 1e20d5d07..bed13a8be 100644 --- a/docs/content/guide/alert-pipeline.md +++ b/docs/content/guide/alert-pipeline.md @@ -41,7 +41,9 @@ Two behaviors follow Alertmanager: - A **self-inhibition guard**: a result matching both `source_match` and `target_match` does not inhibit itself. - Inhibition is **non-transitive**: a *silenced* source still inhibits its targets (the source index is updated before silencing), but an *inhibited* target does not become a source. -Metrics: `rsigma_inhibited_total{rule}` (results muted, by rule name) and `rsigma_inhibit_sources_active` (currently-active sources). +Tracked sources across all rules are capped by `max_inhibit_sources` (default 100000), so `equal` selectors over high-cardinality values (source IPs, usernames from untrusted logs) cannot grow the store without bound. At the cap, a result that would open a new source is still checked as a target but is not tracked, and already-tracked sources keep refreshing. {{ added "unreleased" }} + +Metrics: `rsigma_inhibited_total{rule}` (results muted, by rule name), `rsigma_inhibit_sources_active` (currently-active sources), and `rsigma_inhibit_sources_refused_total{rule}` (new sources not tracked because the cap was reached). {{ added "unreleased" }} ## Silencing @@ -113,6 +115,7 @@ dedup: | `dedup.resolve_timeout` | `1h` | Idle timeout after which an active alert resolves and is evicted (humantime). | | `dedup.max_active_alerts` | `100000` | Ceiling on concurrently-active alerts. Once reached, a first-fire for a new fingerprint passes through un-deduped instead of growing the store, so a high-cardinality fingerprint cannot exhaust memory. The `rsigma_dedup_store_entries` gauge plateauing at this value signals saturation. | | `max_silences` | `1000` | Ceiling on concurrently-tracked dynamic (API) silences. `POST /api/v1/silences` returns `429` past this. Static silences from the config do not count against it. | +| `max_inhibit_sources` | `100000` | Ceiling on tracked inhibition sources across all `inhibit_rules`. Once reached, a result that would open a new source is not tracked (counted on `rsigma_inhibit_sources_refused_total{rule}`); it can still be inhibited, and existing sources keep refreshing. {{ added "unreleased" }} | ### Field selectors @@ -153,7 +156,7 @@ When the daemon runs with `--state-db `, the alert pipeline's state is per Restore is window-aware: dedup alerts past `resolve_timeout`, incidents past their `resolve_timeout`, silences past `ends_at`, and inhibition sources past their rule's `duration` are dropped during restore, so stale state never lingers. Deterministic `group_by` incident ids are preserved across the restart. -Restore and hot-reload also enforce the configured caps, so lowering one takes effect at once instead of when the excess ages out. {{ added "unreleased" }} Beyond `dedup.max_active_alerts` or `group.caps.max_open_incidents`, the least recently seen alerts or incidents are evicted without a `resolved` record, and each remaining incident is cut to `max_results_per_incident` samples and `max_entities_per_incident` values per selector. The daemon logs a warning with the eviction counts. A snapshot whose version does not match the current build is ignored with a warning (the daemon starts fresh). `--clear-state` skips the restore (and `--keep-state` forces it), matching the correlation-state flags. +Restore and hot-reload also enforce the configured caps, so lowering one takes effect at once instead of when the excess ages out. {{ added "unreleased" }} Beyond `dedup.max_active_alerts`, `group.caps.max_open_incidents`, or `max_inhibit_sources`, the least recently seen alerts, incidents, or inhibition sources are evicted (alerts and incidents without a `resolved` record), and each remaining incident is cut to `max_results_per_incident` samples and `max_entities_per_incident` values per selector. The daemon logs a warning with the eviction counts. A snapshot whose version does not match the current build is ignored with a warning (the daemon starts fresh). `--clear-state` skips the restore (and `--keep-state` forces it), matching the correlation-state flags. ## Grouping diff --git a/docs/content/guide/performance-tuning.md b/docs/content/guide/performance-tuning.md index 3e84c5f89..572e5410d 100644 --- a/docs/content/guide/performance-tuning.md +++ b/docs/content/guide/performance-tuning.md @@ -184,7 +184,7 @@ The improvements below are separate same-build comparisons on the pinned SigmaHQ Correlation state lives in memory unless `--state-db` writes periodic snapshots to SQLite. The hard cap is `max_state_entries`, default 100,000 `(correlation, group-key)` entries across all correlation rules, settable with `--max-state-entries` (or `daemon.correlation.max_state_entries` in the config file). When the cap is hit, the engine evicts the stalest 10% and emits a warning. -The cap bounds the number of groups, not the bytes within one. A single group's window state grows with `timespan` x event rate: 8 bytes per in-window event for `event_count`, 16 for the numeric aggregations (`value_sum`/`value_avg`/`value_percentile`/`value_median`), and roughly 32 bytes plus the value string for `value_count`. `--max-group-entries` (or the per-rule `rsigma.max_group_entries` custom attribute) caps that within-window growth; when a group exceeds it, the oldest entries are dropped, which can only under-count. Session windows always keep their oldest entry as the span anchor so truncation cannot silently extend the `timespan` cap. Unset means unbounded, the historical behavior. The measured shape (see [Benchmarks](../benchmarks.md#window-mode-memory-stress)): 1M unique session keys against the default cap peaked at 39.8 MiB, and a fully chatty `event_count` workload (100 groups sustaining 1 event/s through a 2h session cap) held 6.3 MiB. A live but quiet session group costs ~256 bytes, dominated by the group-key strings. +The cap bounds the number of groups, not the bytes within one. A single group's window state grows with `timespan` x event rate: 8 bytes per in-window event for `event_count`, 16 for the numeric aggregations (`value_sum`/`value_avg`/`value_percentile`/`value_median`), and roughly 32 bytes plus the value string for `value_count`. `--max-group-entries` (or the per-rule `rsigma.max_group_entries` custom attribute) caps that within-window growth; when a group exceeds it, the oldest entries are dropped, which can only under-count. Session windows always keep their oldest entry as the span anchor so truncation cannot silently extend the `timespan` cap. Unset means unbounded, the historical behavior. A correlation whose `timespan` is longer than one hour and has no cap from either source logs a warning at load naming the rule, since a busy group's window can then reach gigabytes (24h at 10k events/s is about 864M timestamps). {{ added "unreleased" }} The measured shape (see [Benchmarks](../benchmarks.md#window-mode-memory-stress)): 1M unique session keys against the default cap peaked at 39.8 MiB, and a fully chatty `event_count` workload (100 groups sustaining 1 event/s through a 2h session cap) held 6.3 MiB. A live but quiet session group costs ~256 bytes, dominated by the group-key strings. Window modes (`sliding`/`tumbling`/`session`) have identical per-event cost; they differ only in how long entries are retained. `tumbling` resets per-group state at each bucket boundary and is the cheapest under sustained load; `session` retains everything between the first event and the `timespan` cap, so it is the mode to watch on chatty groups. diff --git a/docs/content/reference/metrics.md b/docs/content/reference/metrics.md index e4bcb0641..16b4fcade 100644 --- a/docs/content/reference/metrics.md +++ b/docs/content/reference/metrics.md @@ -99,6 +99,7 @@ Exposed when the daemon is built with `daemon`. The fixed label sets on `rsigma_ | `rsigma_silences_active` | gauge | none | Currently-active silences. | | `rsigma_inhibited_total` | counter | `rule` | Results muted by an inhibition rule, by rule name. | | `rsigma_inhibit_sources_active` | gauge | none | Currently-active inhibition sources. | +| `rsigma_inhibit_sources_refused_total` | counter | `rule` | Results that would have opened a new inhibition source but were not tracked because `max_inhibit_sources` was reached, by rule name. Appears once a source is first refused. {{ added "unreleased" }} | | `rsigma_alert_pipeline_duration_seconds` | histogram | none | Alert-pipeline stage duration in seconds. | ## Risk-based alerting (9 metrics) From 6cbcc0d4b62b17db890c10758c820c5c73482f0f Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 19:00:15 +0200 Subject: [PATCH 4/5] docs(changelog): add the PR number to the inhibit source cap entry --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3ca2f66ff..6490e2685 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ All notable changes to RSigma are documented in this file. Each entry correspond ## [Unreleased] -### The inhibit store is capped, and long unbounded correlation windows warn at load +### The inhibit store is capped, and long unbounded correlation windows warn at load (#616) `InhibitStore` was the only alert-pipeline store without a count cap: sources were pruned by `duration` only, so `equal` selectors over attacker-influenced values (source IPs, usernames) could grow it without bound. A new top-level `max_inhibit_sources` key (default 100000) caps tracked sources across all `inhibit_rules`. At the cap, a result that would open a new source is still checked as a target but is not tracked, and the refusal is counted on the new `rsigma_inhibit_sources_refused_total{rule}` counter; already-tracked sources keep refreshing. Restore and hot-reload trim the store to the cap, evicting the least recently seen sources, and `CapTrim` reports them as `inhibit_sources`. For library users, `AlertPipelineFile` gains `max_inhibit_sources`, and `MetricsHook` gains `on_inhibit_source_refused`, which defaults to a no-op. From bf69d1ea9e33582672d4529729d32f69fc234be5 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:21:56 +0200 Subject: [PATCH 5/5] docs: count the inhibit source refusal metric in the catalogue totals --- docs/_data/vars.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/_data/vars.yml b/docs/_data/vars.yml index 48362210b..0786d2da0 100644 --- a/docs/_data/vars.yml +++ b/docs/_data/vars.yml @@ -18,8 +18,8 @@ rsigma: # Referenced as {{ rsigma.metrics.names }} etc. Keep in sync with # crates/rsigma-cli/src/daemon/metrics.rs and docs/content/reference/metrics.md. metrics: - names: 96 # total metric names in the definition catalogue - always: 91 # registered without feature gates (names - otlp - tls) + names: 97 # total metric names in the definition catalogue + always: 92 # registered without feature gates (names - otlp - tls) otlp: 3 # daemon-otlp tls: 2 # daemon-tls crates: