From 39b929fcb413e01e76ef9bd69e94fc26ffb3777d Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:50:32 +0200 Subject: [PATCH 1/5] bench(eval): measure percentile and median checks over large windows --- crates/rsigma-eval/benches/correlation.rs | 87 ++++++++++++++++++++++- 1 file changed, 85 insertions(+), 2 deletions(-) diff --git a/crates/rsigma-eval/benches/correlation.rs b/crates/rsigma-eval/benches/correlation.rs index 8cdb52fd7..5c9e5cf37 100644 --- a/crates/rsigma-eval/benches/correlation.rs +++ b/crates/rsigma-eval/benches/correlation.rs @@ -1,8 +1,8 @@ //! Correlation engine benchmarks for rsigma-eval. //! //! Measures event_count and temporal correlation performance, end-to-end -//! throughput with mixed detection + correlation, and state map pressure -//! from many unique group keys. +//! throughput with mixed detection + correlation, state map pressure from +//! many unique group keys, and percentile/median checks over large windows. mod datagen; @@ -375,6 +375,88 @@ level: high group.finish(); } +// --------------------------------------------------------------------------- +// Benchmark: percentile and median over large windows +// --------------------------------------------------------------------------- + +fn bench_correlation_numeric_window(c: &mut Criterion) { + let mut group = c.benchmark_group("correlation_numeric_window"); + group.sample_size(20); + + // One group fed one event per second into a sliding window of `size` + // seconds, so after the warm-up every measured event is checked against + // a full window of `size` values. + let yaml_for = |corr_type: &str, size: usize| { + format!( + r#" +title: Latency Sample +id: bench-latency-001 +logsource: + product: web +detection: + selection: + EventType: 'request' + condition: selection +level: low +--- +title: Latency Window +id: corr-latency-001 +correlation: + type: {corr_type} + rules: + - bench-latency-001 + group-by: + - Host + timespan: {size}s + condition: + field: latency + percentile: 95 + gte: 1000000000 +level: high +"# + ) + }; + + let n_events = 1_000usize; + let mut rng = datagen::rng(); + let event_values: Vec = (0..n_events) + .map(|_| { + serde_json::json!({ + "EventType": "request", + "Host": "web-01", + "latency": rng.random_range(0.0..5_000.0), + }) + }) + .collect(); + let events: Vec = event_values.iter().map(JsonEvent::borrow).collect(); + + for corr_type in ["value_percentile", "value_median"] { + for size in [1_000usize, 5_000] { + let collection = parse_sigma_yaml(&yaml_for(corr_type, size)).unwrap(); + let mut engine = CorrelationEngine::new(CorrelationConfig::default()); + engine.add_collection(&collection).unwrap(); + let mut ts = 1_000_000i64; + for i in 0..size { + let _ = engine.process_event_at(&events[i % n_events], ts); + ts += 1; + } + + group.throughput(criterion::Throughput::Elements(n_events as u64)); + group.bench_with_input(BenchmarkId::new(corr_type, size), &events, |b, events| { + b.iter(|| { + for event in events { + let result = engine.process_event_at(black_box(event), ts); + black_box(&result); + ts += 1; + } + }); + }); + } + } + + group.finish(); +} + // --------------------------------------------------------------------------- // Criterion harness // --------------------------------------------------------------------------- @@ -387,5 +469,6 @@ criterion_group!( bench_correlation_batch, bench_correlation_state_pressure, bench_correlation_window_modes, + bench_correlation_numeric_window, ); criterion_main!(benches); From 6f684d2b445f33cd4387277fc57db1846951178c Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:50:32 +0200 Subject: [PATCH 2/5] perf(eval): select percentile and median ranks instead of sorting the window --- crates/rsigma-eval/src/correlation/tests.rs | 96 +++++++++++++-- crates/rsigma-eval/src/correlation/window.rs | 116 ++++++++----------- 2 files changed, 139 insertions(+), 73 deletions(-) diff --git a/crates/rsigma-eval/src/correlation/tests.rs b/crates/rsigma-eval/src/correlation/tests.rs index 08cbb8e1d..7a0d549ef 100644 --- a/crates/rsigma-eval/src/correlation/tests.rs +++ b/crates/rsigma-eval/src/correlation/tests.rs @@ -451,13 +451,13 @@ fn test_check_condition_with_extended_expr() { #[test] fn test_percentile_linear_interp_single() { - assert!((percentile_linear_interp(&[42.0], 50.0) - 42.0).abs() < f64::EPSILON); + assert!((percentile_linear_interp(&mut [42.0], 50.0) - 42.0).abs() < f64::EPSILON); } #[test] fn test_percentile_linear_interp_basic() { // Values: [1, 2, 3, 4, 5] - let values = &[1.0, 2.0, 3.0, 4.0, 5.0]; + let values = &mut [1.0, 2.0, 3.0, 4.0, 5.0]; // 0th percentile = 1.0 assert!((percentile_linear_interp(values, 0.0) - 1.0).abs() < f64::EPSILON); // 25th percentile = 2.0 @@ -473,7 +473,7 @@ fn test_percentile_linear_interp_basic() { #[test] fn test_percentile_linear_interp_interpolation() { // Values: [10, 20, 30, 40] - let values = &[10.0, 20.0, 30.0, 40.0]; + let values = &mut [10.0, 20.0, 30.0, 40.0]; // 50th percentile: rank = 0.5 * 3 = 1.5, interp between 20 and 30 = 25 assert!((percentile_linear_interp(values, 50.0) - 25.0).abs() < f64::EPSILON); } @@ -481,12 +481,92 @@ fn test_percentile_linear_interp_interpolation() { #[test] fn test_percentile_linear_interp_1st_percentile() { // Values: [1, 2, 3, ..., 100] - let values: Vec = (1..=100).map(|x| x as f64).collect(); + let mut values: Vec = (1..=100).map(|x| x as f64).collect(); // 1st percentile = 1.0 + 0.01 * 99 * (2.0 - 1.0) ~ 1.99 - let p1 = percentile_linear_interp(&values, 1.0); + let p1 = percentile_linear_interp(&mut values, 1.0); assert!((p1 - 1.99).abs() < 0.01); } +/// Percentile by full sort, the definition the selection-based version must +/// reproduce exactly. +fn sorted_percentile(values: &[f64], percentile: f64) -> f64 { + let mut sorted = values.to_vec(); + sorted.sort_by(f64::total_cmp); + let n = sorted.len(); + if n == 1 { + return sorted[0]; + } + let rank = percentile.clamp(0.0, 100.0) / 100.0 * (n - 1) as f64; + let (lower, upper) = (rank.floor() as usize, rank.ceil() as usize); + if lower == upper || upper >= n { + sorted[lower.min(n - 1)] + } else { + sorted[lower] + (rank - lower as f64) * (sorted[upper] - sorted[lower]) + } +} + +/// Deterministic unsorted samples with repeats and signed zeros. +fn scrambled_samples(n: usize) -> Vec { + let mut x: u64 = 0x9E37_79B9_7F4A_7C15 ^ n as u64; + (0..n) + .map(|i| { + x = x + .wrapping_mul(6_364_136_223_846_793_005) + .wrapping_add(1_442_695_040_888_963_407); + match i % 7 { + 0 => 0.0, + 1 => -0.0, + 2 => ((x >> 40) % 16) as f64, + _ => ((x >> 11) as f64 / (1u64 << 53) as f64 - 0.5) * 1e6, + } + }) + .collect() +} + +#[test] +fn test_percentile_selection_matches_sorted_reference() { + for n in (1..=64).chain([99, 100, 1_000, 5_001]) { + let samples = scrambled_samples(n); + for p in (0..=100).map(f64::from).chain([0.5, 33.3, 99.9]) { + let got = percentile_linear_interp(&mut samples.clone(), p); + let want = sorted_percentile(&samples, p); + assert_eq!( + got.to_bits(), + want.to_bits(), + "n={n} p={p}: {got} != {want}" + ); + } + } +} + +#[test] +fn test_window_median_matches_sorted_reference() { + let cond = CompiledCondition { + field: Some(vec!["latency".to_string()]), + predicates: vec![(ConditionOperator::Gte, f64::MIN)], + percentile: None, + }; + for n in (1..=64).chain([99, 100, 1_000, 5_000]) { + let samples = scrambled_samples(n); + let mut state = WindowState::new_for(CorrelationType::ValueMedian); + for (i, v) in samples.iter().enumerate() { + state.push_numeric(i as i64, *v); + } + let mut sorted = samples.clone(); + sorted.sort_by(f64::total_cmp); + let mid = n / 2; + let want = if n % 2 == 0 { + (sorted[mid - 1] + sorted[mid]) / 2.0 + } else { + sorted[mid] + }; + let checked = state.check_condition(&cond, CorrelationType::ValueMedian, &[], None); + let current = state.current_value(CorrelationType::ValueMedian, &[], None); + assert_eq!(checked.map(f64::to_bits), Some(want.to_bits()), "n={n}"); + assert_eq!(current.map(f64::to_bits), Some(want.to_bits()), "n={n}"); + } +} + #[test] fn test_value_percentile_check_condition() { let mut state = WindowState::new_for(CorrelationType::ValuePercentile); @@ -509,14 +589,14 @@ fn test_value_percentile_check_condition() { #[test] fn test_percentile_0th_and_100th() { - let values = &[5.0, 10.0, 15.0, 20.0]; + let values = &mut [5.0, 10.0, 15.0, 20.0]; assert!((percentile_linear_interp(values, 0.0) - 5.0).abs() < f64::EPSILON); assert!((percentile_linear_interp(values, 100.0) - 20.0).abs() < f64::EPSILON); } #[test] fn test_percentile_two_values() { - let values = &[10.0, 20.0]; + let values = &mut [10.0, 20.0]; // 50th percentile between 10 and 20 = 15 assert!((percentile_linear_interp(values, 50.0) - 15.0).abs() < f64::EPSILON); // 25th percentile = 12.5 @@ -525,7 +605,7 @@ fn test_percentile_two_values() { #[test] fn test_percentile_clamps_out_of_range() { - let values = &[1.0, 2.0, 3.0]; + let values = &mut [1.0, 2.0, 3.0]; // Negative percentile clamps to 0 assert!((percentile_linear_interp(values, -10.0) - 1.0).abs() < f64::EPSILON); // > 100 clamps to 100 diff --git a/crates/rsigma-eval/src/correlation/window.rs b/crates/rsigma-eval/src/correlation/window.rs index 40293829b..e8d7ebac6 100644 --- a/crates/rsigma-eval/src/correlation/window.rs +++ b/crates/rsigma-eval/src/correlation/window.rs @@ -270,44 +270,13 @@ impl WindowState { // Proper percentile calculation using linear interpolation. // The condition threshold represents a percentile rank (0-100). // We compute the value at that percentile from the window data. - if entries.is_empty() { - return None; - } - let mut values: Vec = entries - .iter() - .map(|(_, v)| *v) - .filter(|v| v.is_finite()) - .collect(); - if values.is_empty() { - return None; - } - values.sort_by(|a, b| a.total_cmp(b)); - let percentile_rank = condition.percentile.map(|p| p as f64).unwrap_or(50.0); - let pval = percentile_linear_interp(&values, percentile_rank); - return Some(pval); + return window_percentile(entries, condition.percentile); } (WindowState::NumericAgg { entries }, CorrelationType::ValueMedian) => { // An empty window has no median. Returning `0.0` here would // spuriously satisfy predicates like `lte: 0` or `eq: 0`, so // match the percentile branch and skip evaluation. - if entries.is_empty() { - return None; - } - let mut values: Vec = entries - .iter() - .map(|(_, v)| *v) - .filter(|v| v.is_finite()) - .collect(); - if values.is_empty() { - return None; - } - values.sort_by(|a, b| a.total_cmp(b)); - let mid = values.len() / 2; - if values.len().is_multiple_of(2) && values.len() >= 2 { - (values[mid - 1] + values[mid]) / 2.0 - } else { - values[mid] - } + window_median(entries)? } _ => return None, // mismatched state/type }; @@ -370,35 +339,10 @@ impl WindowState { } } (WindowState::NumericAgg { entries }, CorrelationType::ValuePercentile) => { - let mut values: Vec = entries - .iter() - .map(|(_, v)| *v) - .filter(|v| v.is_finite()) - .collect(); - if values.is_empty() { - return None; - } - values.sort_by(|a, b| a.total_cmp(b)); - let rank = percentile.map(|p| p as f64).unwrap_or(50.0); - Some(percentile_linear_interp(&values, rank)) + window_percentile(entries, percentile) } (WindowState::NumericAgg { entries }, CorrelationType::ValueMedian) => { - let mut values: Vec = entries - .iter() - .map(|(_, v)| *v) - .filter(|v| v.is_finite()) - .collect(); - if values.is_empty() { - return None; - } - values.sort_by(|a, b| a.total_cmp(b)); - let mid = values.len() / 2; - let v = if values.len().is_multiple_of(2) && values.len() >= 2 { - (values[mid - 1] + values[mid]) / 2.0 - } else { - values[mid] - }; - Some(v) + window_median(entries) } _ => None, } @@ -572,12 +516,50 @@ pub(super) fn eval_temporal_expr( } } +/// The window's finite values, in arrival order. +fn finite_values(entries: &VecDeque<(i64, f64)>) -> Vec { + entries + .iter() + .map(|(_, v)| *v) + .filter(|v| v.is_finite()) + .collect() +} + +/// Value at `percentile` (default 50) across the window's finite values, or +/// `None` when there are none. +fn window_percentile(entries: &VecDeque<(i64, f64)>, percentile: Option) -> Option { + let mut values = finite_values(entries); + if values.is_empty() { + return None; + } + let rank = percentile.map(|p| p as f64).unwrap_or(50.0); + Some(percentile_linear_interp(&mut values, rank)) +} + +/// Median of the window's finite values, averaging the two middle values for +/// an even count, or `None` when there are none. +fn window_median(entries: &VecDeque<(i64, f64)>) -> Option { + let mut values = finite_values(entries); + let n = values.len(); + if n == 0 { + return None; + } + let (below, &mut upper, _) = values.select_nth_unstable_by(n / 2, f64::total_cmp); + if n.is_multiple_of(2) { + let lower = below.iter().copied().max_by(f64::total_cmp)?; + Some((lower + upper) / 2.0) + } else { + Some(upper) + } +} + /// Compute the value at a given percentile rank using linear interpolation. /// -/// Returns 0.0 if `values` is empty. -/// `values` must be sorted in ascending order. +/// Returns 0.0 if `values` is empty. `values` may be in any order and is +/// partially reordered in place, which keeps this O(n) on average instead of +/// a full sort per evaluation. /// `percentile` is from 0.0 to 100.0. -pub(super) fn percentile_linear_interp(values: &[f64], percentile: f64) -> f64 { +pub(super) fn percentile_linear_interp(values: &mut [f64], percentile: f64) -> f64 { if values.is_empty() { return 0.0; } @@ -596,9 +578,13 @@ pub(super) fn percentile_linear_interp(values: &[f64], percentile: f64) -> f64 { let upper = rank.ceil() as usize; let fraction = rank - lower as f64; + let (_, &mut low, above) = values.select_nth_unstable_by(lower.min(n - 1), f64::total_cmp); if lower == upper || upper >= n { - values[lower.min(n - 1)] + low } else { - values[lower] + fraction * (values[upper] - values[lower]) + // `above` holds every value ordered after `low`, so its minimum is + // the next value in sorted order. + let high = above.iter().copied().min_by(f64::total_cmp).unwrap_or(low); + low + fraction * (high - low) } } From c798799eaa85558a4ad1d8292a2291e9ce7feeaa Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:50:32 +0200 Subject: [PATCH 3/5] perf(eval): carry only the source and group key through correlation chaining --- .../rsigma-eval/src/correlation_engine/mod.rs | 22 +++--- .../src/correlation_engine/tests.rs | 67 +++++++++++++++++++ 2 files changed, 78 insertions(+), 11 deletions(-) diff --git a/crates/rsigma-eval/src/correlation_engine/mod.rs b/crates/rsigma-eval/src/correlation_engine/mod.rs index 75513eac6..ca9335d5e 100644 --- a/crates/rsigma-eval/src/correlation_engine/mod.rs +++ b/crates/rsigma-eval/src/correlation_engine/mod.rs @@ -1104,10 +1104,13 @@ impl CorrelationEngine { out_indices: &mut Vec, ) { debug_assert_eq!(fired.len(), fired_indices.len()); - let mut pending: Vec<(usize, EvaluationResult)> = fired_indices + // Chaining only needs which correlation fired and for which group, so + // `pending` carries that instead of cloning results and their events. + // Only correlation results chain. Detections never reach here. + let mut pending: Vec<(usize, Vec<(String, String)>)> = fired_indices .iter() - .copied() - .zip(fired.iter().cloned()) + .zip(fired) + .filter_map(|(&idx, result)| Some((idx, result.as_correlation()?.group_key.clone()))) .collect(); let mut depth = 0; @@ -1118,16 +1121,12 @@ impl CorrelationEngine { #[allow(clippy::type_complexity)] let mut work: Vec<(usize, Vec<(String, String)>, String)> = Vec::new(); let mut seen = std::collections::HashSet::<(usize, String)>::new(); - for (source_idx, result) in &pending { - // Only correlation results chain. Detections never reach here. - let Some(body) = result.as_correlation() else { - continue; - }; + for (source_idx, group_key) in &pending { for key in self.chain_lookup_keys(*source_idx) { if let Some(indices) = self.rule_index.get(&key) { for &corr_idx in indices { if seen.insert((corr_idx, key.clone())) { - work.push((corr_idx, body.group_key.clone(), key.clone())); + work.push((corr_idx, group_key.clone(), key.clone())); } } } @@ -1202,6 +1201,7 @@ impl CorrelationEngine { } let corr = &self.correlations[corr_idx]; + let fired_group_key = group_key.to_pairs(&corr.group_by); let result = EvaluationResult { header: RuleHeader { rule_title: corr.title.clone(), @@ -1213,7 +1213,7 @@ impl CorrelationEngine { }, body: ResultBody::Correlation(CorrelationBody { correlation_type: corr_type, - group_key: group_key.to_pairs(&corr.group_by), + group_key: fired_group_key.clone(), aggregated_value: agg_value, timespan_secs: timespan, // Chained correlations don't include events @@ -1223,7 +1223,7 @@ impl CorrelationEngine { event_refs: None, }), }; - next_pending.push((corr_idx, result.clone())); + next_pending.push((corr_idx, fired_group_key)); out.push(result); out_indices.push(corr_idx); self.last_alert.insert(alert_key.clone(), ts); diff --git a/crates/rsigma-eval/src/correlation_engine/tests.rs b/crates/rsigma-eval/src/correlation_engine/tests.rs index b59fd74c7..f3b901a51 100644 --- a/crates/rsigma-eval/src/correlation_engine/tests.rs +++ b/crates/rsigma-eval/src/correlation_engine/tests.rs @@ -2991,6 +2991,73 @@ level: high } } +#[test] +fn test_chaining_in_full_event_mode() { + // Child results carry retained events in full mode; the chained parent + // still sees the child's group and fires without events of its own. + let yaml = r#" +title: click +id: det-rule +logsource: + product: test +detection: + selection: + action: click + condition: selection +--- +title: correlation A +id: corr-a +correlation: + type: event_count + rules: + - det-rule + group-by: + - User + timespan: 5m + condition: + gte: 2 +level: high +--- +title: correlation B +id: corr-b +correlation: + type: event_count + rules: + - corr-a + group-by: + - User + timespan: 5m + condition: + gte: 2 +level: high +"#; + let collection = parse_sigma_yaml(yaml).unwrap(); + let mut engine = CorrelationEngine::new(CorrelationConfig { + correlation_event_mode: CorrelationEventMode::Full, + max_correlation_events: 10, + ..emit_every_hop() + }); + engine.add_collection(&collection).unwrap(); + + let mut last = None; + for i in 0..3 { + let v = json!({"action": "click", "User": "alice"}); + last = Some(engine.process_event_at(&JsonEvent::borrow(&v), 1_000 + i)); + } + let result = last.unwrap(); + let bodies: Vec<_> = result + .correlations() + .map(|c| (c.header.rule_title.as_str(), c.as_correlation().unwrap())) + .collect(); + let [("correlation A", a), ("correlation B", b)] = bodies.as_slice() else { + panic!("expected A then chained B, got {bodies:?}"); + }; + assert_eq!(a.events.as_ref().map(Vec::len), Some(3)); + assert!(b.events.is_none()); + assert_eq!(b.group_key, [("User".to_string(), "alice".to_string())]); + assert_eq!(b.aggregated_value, 2.0); +} + #[test] fn test_chaining_parent_referenced_by_child_name() { let yaml = r#" From 169b5c4379f520cfffa4dc9d24a9bca2283f9462 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:50:32 +0200 Subject: [PATCH 4/5] docs(changelog): note the percentile selection and chaining changes --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a70af1735..b23819d3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to RSigma are documented in this file. Each entry correspond ## [Unreleased] +### Percentile and median checks no longer sort the window + +`value_percentile` and `value_median` correlations sorted every finite value in the group's window on each event fed into the group. They now select the needed ranks with `select_nth_unstable_by`, which is O(n) on average and returns exactly the same values, including interpolated percentiles and even-count medians. In the new `correlation_numeric_window` benchmark, a group checked against a 5,000-value window processes events about 4x faster. Chained correlations also stop cloning every fired result, including its retained events in full event mode, just to read which correlation fired and for which group. + ### OTLP log decoding is fuzzed (#617) OTLP log ingestion decodes untrusted protobuf and flattens each record to JSON, but it was the only untrusted input surface without a `cargo-fuzz` harness. The new `fuzz_otlp_logs_decode` target decodes arbitrary bytes as `ExportLogsServiceRequest`, converts whatever decodes with `logs_request_to_raw_events`, and asserts each event is valid JSON. It runs in the weekly fuzz workflow from a committed seed corpus. Conversion recurses once per nested `AnyValue`, so its depth depends on prost's decode recursion limit; a new unit test fails if a dependency turns that limit off through prost's `no-recursion-limit` feature. From 2c36519f2597a5668c147bea3b1ccf0ba563eeac Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Sun, 18 Oct 2026 20:50:49 +0200 Subject: [PATCH 5/5] docs(changelog): add the PR number to the percentile selection entry --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b23819d3a..297f4e7e0 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] -### Percentile and median checks no longer sort the window +### Percentile and median checks no longer sort the window (#619) `value_percentile` and `value_median` correlations sorted every finite value in the group's window on each event fed into the group. They now select the needed ranks with `select_nth_unstable_by`, which is O(n) on average and returns exactly the same values, including interpolated percentiles and even-count medians. In the new `correlation_numeric_window` benchmark, a group checked against a 5,000-value window processes events about 4x faster. Chained correlations also stop cloning every fired result, including its retained events in full event mode, just to read which correlation fired and for which group.