diff --git a/CHANGELOG.md b/CHANGELOG.md index a70af173..297f4e7e 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 (#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. + ### 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. diff --git a/crates/rsigma-eval/benches/correlation.rs b/crates/rsigma-eval/benches/correlation.rs index 8cdb52fd..5c9e5cf3 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); diff --git a/crates/rsigma-eval/src/correlation/tests.rs b/crates/rsigma-eval/src/correlation/tests.rs index 08cbb8e1..7a0d549e 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 40293829..e8d7ebac 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) } } diff --git a/crates/rsigma-eval/src/correlation_engine/mod.rs b/crates/rsigma-eval/src/correlation_engine/mod.rs index 75513eac..ca9335d5 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 b59fd74c..f3b901a5 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#"