diff --git a/.gitleaks.toml b/.gitleaks.toml index 6a7d39c..a9d6be3 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -9,8 +9,7 @@ useDefault = true [allowlist] -description = "Bundled filebeat-compat pipeline data - ported public Vector templates and pinned elastic/integrations test fixtures, not credentials (e.g. the Meraki event subtype '8021x_eap_failure' pattern-matches a key rule)" +description = "Bundled filebeat-compat pipeline data - ported public Vector templates, not credentials (e.g. the Meraki event subtype '8021x_eap_failure' pattern-matches a key rule)" paths = [ '''pipelines/filebeat/filebeat\.vrl''', - '''tests/fixtures/filebeat/filebeat-testdata\.tar\.gz''', ] diff --git a/Cargo.lock b/Cargo.lock index 87d6dc3..8490bb1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -997,7 +997,6 @@ dependencies = [ "csv", "dotenvy", "figment", - "flate2", "maxminddb", "metrics", "rdkafka", @@ -1008,7 +1007,6 @@ dependencies = [ "serde", "serde_json", "sonic-rs", - "tar", "temp-env", "tempfile", "thiserror", @@ -1310,16 +1308,6 @@ dependencies = [ "flate2", ] -[[package]] -name = "filetime" -version = "0.2.29" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" -dependencies = [ - "cfg-if", - "libc", -] - [[package]] name = "find-msvc-tools" version = "0.1.14" @@ -4848,17 +4836,6 @@ dependencies = [ "nom 8.0.0", ] -[[package]] -name = "tar" -version = "0.4.46" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" -dependencies = [ - "filetime", - "libc", - "xattr", -] - [[package]] name = "temp-env" version = "0.3.6" @@ -6139,16 +6116,6 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" -[[package]] -name = "xattr" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" -dependencies = [ - "libc", - "rustix", -] - [[package]] name = "xxhash-rust" version = "0.8.19" diff --git a/Cargo.toml b/Cargo.toml index 0b645f8..49a9b2b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -125,23 +125,16 @@ tikv-jemallocator = { version = ">=0.7.0, <0.8", optional = true } # Direct rdkafka for the pgo-driver bin only. The wrapper uses scalo's Kafka transport. rdkafka = { version = ">=0.39, <0.40", features = ["tokio", "cmake-build"], optional = true } -# The pgo-driver bin reads the filebeat corpus archive in memory; same ranges as -# the dev-dependencies so cargo unifies to one of each. -flate2 = { version = ">=1.1.10, <2", optional = true } -tar = { version = ">=0.4.46, <0.5", optional = true } [features] default = [] jemalloc = ["dep:tikv-jemallocator"] -pgo-driver = ["dep:rdkafka", "dep:flate2", "dep:tar"] +pgo-driver = ["dep:rdkafka"] enrichment-mmdb = ["dep:maxminddb"] enrichment-sqlite = ["dep:rusqlite"] [dev-dependencies] tempfile = ">=3.27.0, <4" -# Reads the committed filebeat corpus tar.gz in memory. -flate2 = ">=1.1.10, <2" -tar = ">=0.4.46, <0.5" # A dev-dependency, not a crate feature, so CI's `features: default` still builds # the broker-free pipeline tests. scalo = { version = ">=2.14.30, <3", features = [ diff --git a/docs/internals.md b/docs/internals.md index 7bf48a5..c0d24f7 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -70,8 +70,8 @@ pure-VRL file plus its `timezones.csv` enrichment table. It is convenience data, not engine capability: opting in means pointing the standard `transforms.dir` and `enrichment_tables` config at those files. The engine carries no filebeat-specific code, and the integration tests -(`tests/integration/filebeat_pipeline.rs`) drive the bundle with the real -elastic/integrations pipeline test corpus, which doubles as a full-engine +(`tests/integration/filebeat_pipeline.rs`) drive the bundle with the synthetic +cases in `tests/fixtures/filebeat/cases.yaml`, which double as a full-engine workload. INTERIM: elastic compatibility is being replaced by `dfe-transform-elastic` diff --git a/pipelines/filebeat/README.md b/pipelines/filebeat/README.md index 5b568da..b02bee3 100644 --- a/pipelines/filebeat/README.md +++ b/pipelines/filebeat/README.md @@ -3,8 +3,8 @@ A pre-canned, opt-in VRL pipeline that keeps DFE 2.1 filebeat feeds working on dfe-transform-vrl. It is pure VRL plus one CSV enrichment table - no binary capability, no config surface. Use it if you want it, ignore it if -you do not. It also doubles as a real-data integration test workload for -the engine (`tests/integration/filebeat_pipeline.rs`). +you do not. It also doubles as an integration test workload for the engine +(`tests/integration/filebeat_pipeline.rs`). > **INTERIM.** Elastic compatibility is being replaced by > `dfe-transform-elastic` (Rust-native). New integrations should NOT @@ -89,7 +89,7 @@ still error the event out (dropped and counted), matching the 2.1 port routes slash-bearing zone names through `parse_timestamp`'s `timezone:` argument (DST-aware via the tz database). -## Known limitations (verified against the elastic golden corpus) +## Known limitations - `user_agent` classification follows VRL's parser, which differs cosmetically from elastic's ("Mac OSX" vs "Mac OS X"). A @@ -109,9 +109,8 @@ still error the event out (dropped and counted), matching the 2.1 Generated from dfe-vector-templates `origin/main` @ `96eb306` by `scripts/filebeat/extract.py` (verifies the seven umbrella templates are identical before collapsing them, strips geoip, unescapes Vector `$$`, -prepends `scripts/filebeat/prologue.vrl`). Validated line-by-line against -the elastic/integrations pipeline test corpus @ `c7bc530` - see -`tests/fixtures/filebeat/README.md`. +prepends `scripts/filebeat/prologue.vrl`). The tests replay the synthetic +lines in `tests/fixtures/filebeat/cases.yaml`. Treat `filebeat.vrl` as generated-then-patched: fixes belong here (it is the shipping artefact), but substantial rework should go back through the diff --git a/pipelines/filebeat/filebeat.vrl b/pipelines/filebeat/filebeat.vrl index 23bec9c..d463c96 100644 --- a/pipelines/filebeat/filebeat.vrl +++ b/pipelines/filebeat/filebeat.vrl @@ -63,8 +63,8 @@ if match(_msg, r'^<134>1 \d{9,10}\.\d+ ') { } else { # Cisco Umbrella exports are CSV; DFE 2.1 carried the log type in the # S3 object path. Honour an upstream-provided .log.file.path first, - # then fall back to column-shape detection (counts verified against - # the elastic/integrations pipeline test corpus). + # then fall back to column-shape detection (the umbrella cases in + # tests/fixtures/filebeat/cases.yaml cover each log type). _path = string(.log.file.path) ?? "" if contains(_path, "auditlogs") { _umbrella_type = "auditlogs" diff --git a/scripts/filebeat/fetch_testdata.py b/scripts/filebeat/fetch_testdata.py deleted file mode 100644 index 28d12e9..0000000 --- a/scripts/filebeat/fetch_testdata.py +++ /dev/null @@ -1,93 +0,0 @@ -"""Fetch the cisco filebeat pipeline test corpora from elastic/integrations -at a pinned SHA using the gh CLI (authenticated, read-only contents API), -for refreshing tests/fixtures/filebeat/filebeat-testdata.tar.gz. - -Usage: - python3 scripts/filebeat/fetch_testdata.py - tar -czf tests/fixtures/filebeat/filebeat-testdata.tar.gz \ - -C cisco_umbrella cisco_ios cisco_meraki LICENSE.txt - -Keep SHA aligned with the dfe-transform-elastic testdata pin unless the -bundled VRL is regenerated against a newer template vintage (see -tests/fixtures/filebeat/README.md). -""" - -from __future__ import annotations - -import base64 -import json -import subprocess -import sys -from pathlib import Path - -REPO = "elastic/integrations" -SHA = "c7bc53023288c7d34afd631c0c3e3552f6219419" - -DIRS = { - "cisco_umbrella/log": "packages/cisco_umbrella/data_stream/log/_dev/test/pipeline", - "cisco_ios/log": "packages/cisco_ios/data_stream/log/_dev/test/pipeline", - "cisco_meraki/log": "packages/cisco_meraki/data_stream/log/_dev/test/pipeline", -} - -EXTRAS = { - "LICENSE.txt": "LICENSE.txt", - "licenses/Elastic-2.0.txt": "licenses/Elastic-2.0.txt", - "cisco_umbrella/manifest.yml": "packages/cisco_umbrella/manifest.yml", - "cisco_ios/manifest.yml": "packages/cisco_ios/manifest.yml", - "cisco_meraki/manifest.yml": "packages/cisco_meraki/manifest.yml", -} - - -def gh_api(path: str) -> bytes: - res = subprocess.run( - ["gh", "api", path], - capture_output=True, - check=False, - ) - if res.returncode != 0: - print(f"FAIL: gh api {path}: {res.stderr.decode('utf-8', 'replace')}", - file=sys.stderr) - sys.exit(1) - return res.stdout - - -def fetch_file(repo_path: str, dest: Path) -> int: - obj = json.loads(gh_api(f"repos/{REPO}/contents/{repo_path}?ref={SHA}")) - if obj.get("encoding") != "base64": - print(f"FAIL: unexpected encoding for {repo_path}: " - f"{obj.get('encoding')}", file=sys.stderr) - sys.exit(1) - data = base64.b64decode(obj["content"]) - dest.parent.mkdir(parents=True, exist_ok=True) - dest.write_bytes(data) - return len(data) - - -def main() -> None: - if len(sys.argv) != 2: - print(__doc__, file=sys.stderr) - sys.exit(2) - out = Path(sys.argv[1]) - total = 0 - n = 0 - for local_dir, repo_dir in DIRS.items(): - listing = json.loads(gh_api(f"repos/{REPO}/contents/{repo_dir}?ref={SHA}")) - for entry in listing: - if entry["type"] != "file": - continue - size = fetch_file( - f"{repo_dir}/{entry['name']}", out / local_dir / entry["name"] - ) - total += size - n += 1 - print(f" {local_dir}/{entry['name']} {size}B") - for local_path, repo_path in EXTRAS.items(): - size = fetch_file(repo_path, out / local_path) - total += size - n += 1 - print(f" {local_path} {size}B") - print(f"fetched {n} files, {total} bytes, pinned {SHA}") - - -if __name__ == "__main__": - main() diff --git a/scripts/filebeat/prologue.vrl b/scripts/filebeat/prologue.vrl index b804168..062065d 100644 --- a/scripts/filebeat/prologue.vrl +++ b/scripts/filebeat/prologue.vrl @@ -63,8 +63,8 @@ if match(_msg, r'^<134>1 \d{9,10}\.\d+ ') { } else { # Cisco Umbrella exports are CSV; DFE 2.1 carried the log type in the # S3 object path. Honour an upstream-provided .log.file.path first, - # then fall back to column-shape detection (counts verified against - # the elastic/integrations pipeline test corpus). + # then fall back to column-shape detection (the umbrella cases in + # tests/fixtures/filebeat/cases.yaml cover each log type). _path = string(.log.file.path) ?? "" if contains(_path, "auditlogs") { _umbrella_type = "auditlogs" diff --git a/scripts/pgo-workload.sh b/scripts/pgo-workload.sh index 20bff06..2a6f0ea 100755 --- a/scripts/pgo-workload.sh +++ b/scripts/pgo-workload.sh @@ -10,8 +10,8 @@ # Usage: # scripts/pgo-workload.sh # -# Runs the shipped filebeat pipeline (pipelines/filebeat) over the committed -# elastic/integrations corpus, so a PGO-instrumented binary profiles the hot +# Runs the shipped filebeat pipeline (pipelines/filebeat) over the synthetic +# cases in tests/fixtures/filebeat/cases.yaml, so a PGO-instrumented binary profiles the hot # path it ships with: Kafka consume -> JSON parse -> VRL (parse_groks, # parse_timestamp, the timezones lookup) -> JSON serialise -> Kafka produce. # @@ -34,7 +34,7 @@ # - Writes an ephemeral wrapper config pointing at pipelines/filebeat # - Starts the passed-in wrapper binary in background # - Waits for the wrapper's /readyz to return 200 -# - Runs pgo-driver, which produces the corpus for the configured duration +# - Runs pgo-driver, which produces the case events for the configured duration # and exits non-zero unless the wrapper's /metrics shows delivered records # and fewer errors than deliveries -- so a workload that transformed # nothing fails here instead of profiling an idle binary @@ -295,7 +295,7 @@ if ! PGO_DRIVER_DURATION_SECS="$DURATION" \ PGO_DRIVER_RPS="${PGO_DRIVER_RPS:-5000}" \ PGO_DRIVER_METRICS_ADDR="127.0.0.1:$METRICS_PORT" \ "$PGO_DRIVER_PATH"; then - echo "error: the workload did not transform and deliver the corpus" >&2 + echo "error: the workload did not transform and deliver the cases" >&2 tail -100 "$WORK_DIR/wrapper.log" >&2 exit 1 fi diff --git a/src/bin/pgo_driver.rs b/src/bin/pgo_driver.rs index f6f07ba..9a47f60 100644 --- a/src/bin/pgo_driver.rs +++ b/src/bin/pgo_driver.rs @@ -8,7 +8,7 @@ //! PGO workload driver for `dfe-transform-vrl`. //! -//! Replays the elastic/integrations filebeat corpus onto the source topic, as +//! Replays the synthetic filebeat cases onto the source topic, as //! the input events the bundled filebeat pipeline reads, so a PGO-instrumented //! `dfe-transform-vrl` profiles the pipeline it ships with: Kafka consume -> //! JSON parse -> `parse_groks`, `parse_timestamp` and the timezones lookup -> @@ -45,9 +45,9 @@ clippy::cast_sign_loss )] -// The same corpus reader and input-event shape the filebeat test suites use. +// The same case loader and input-event shape the filebeat test suites use. #[path = "../../tests/common/filebeat.rs"] -mod filebeat_corpus; +mod filebeat_cases; use std::env; use std::sync::Arc; @@ -58,7 +58,7 @@ use rdkafka::ClientConfig; use rdkafka::producer::{FutureProducer, FutureRecord, Producer}; use tokio::time::{MissedTickBehavior, interval}; -use crate::filebeat_corpus as fb; +use crate::filebeat_cases as fb; /// How often the producer tops the sent count up to the configured rate. const PRODUCE_TICK: Duration = Duration::from_millis(10); @@ -68,11 +68,10 @@ async fn main() { let cfg = Config::from_env(); println!("pgo-driver starting: {cfg:#?}"); - let events = corpus_events(); + let events = case_events(); println!( - "pgo-driver: replaying {} filebeat corpus events from {} logs", - events.len(), - fb::all_logs().len() + "pgo-driver: replaying {} filebeat case events", + events.len() ); let producer: FutureProducer = match ClientConfig::new() @@ -178,20 +177,16 @@ fn env_parsed(key: &str, default: T) -> T { } // =========================================================================== -// Corpus +// Cases // =========================================================================== -/// Every corpus line as the input event the bundled pipeline reads. -fn corpus_events() -> Vec> { - let mut events = Vec::new(); - for log in fb::all_logs() { - let conf = fb::corpus_conf(log); - for line in fb::corpus_lines(log) { - let event = fb::input_event(&line, conf.as_ref()); - events.push(serde_json::to_vec(&event).expect("a corpus event serialises")); - } - } - assert!(!events.is_empty(), "the filebeat corpus has no events"); +/// Every case line as the input event the bundled pipeline reads. +fn case_events() -> Vec> { + let events: Vec> = fb::cases() + .iter() + .map(|case| serde_json::to_vec(&fb::case_event(case)).expect("a case event serialises")) + .collect(); + assert!(!events.is_empty(), "the filebeat cases file has no events"); events } @@ -228,7 +223,7 @@ impl Stats { } } -/// Queue corpus events at `cfg.rps` until `deadline`, cycling the corpus. +/// Queue case events at `cfg.rps` until `deadline`, cycling through them. async fn produce( producer: &FutureProducer, cfg: &Config, diff --git a/tests/TESTING.md b/tests/TESTING.md index d5d6abb..7d29fa7 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -8,7 +8,7 @@ Unit tests run without any external infrastructure, embedded in the source files cargo nextest run --lib ``` -The integration tests in `tests/` run without a broker: VRL compilation, transforms, config, enrichment, the filebeat corpus, the direct transport on local ports, and the pipeline over scalo's in-memory transport. +The integration tests in `tests/` run without a broker: VRL compilation, transforms, config, enrichment, the filebeat cases, the direct transport on local ports, and the pipeline over scalo's in-memory transport. The core tier is every test that is not `#[ignore]`: unit, integration, smoke, and the e2e tests that use a broker container scalo starts (`filebeat_kafka`, `held_acks`, `kafka_tls`). PR and push runs use it, and so does `hyperi-ci check`. diff --git a/tests/common/filebeat.rs b/tests/common/filebeat.rs index 98c9b3f..4af03dd 100644 --- a/tests/common/filebeat.rs +++ b/tests/common/filebeat.rs @@ -1,164 +1,108 @@ // Project: dfe-transform-vrl // File: tests/common/filebeat.rs -// Purpose: The filebeat corpus and its golden comparison, shared by both suites +// Purpose: The filebeat test cases and their comparison, shared by every suite that replays them // Language: Rust // // License: BUSL-1.1 // Copyright: (c) 2026 HYPERI PTY LIMITED -// A missing fixture is unrecoverable and has no caller to report it to, so the -// reader panics rather than threading a Result through every test. +// A missing or malformed case file is unrecoverable and has no caller to +// report it to, so the loader panics rather than threading a Result through +// every test. #![allow(dead_code, clippy::panic)] -//! The elastic/integrations filebeat corpus, and what counts as agreeing -//! with it. +//! The synthetic filebeat cases in `tests/fixtures/filebeat/cases.yaml`. //! -//! Two suites drive the same bundled pipeline over the same data: -//! `tests/integration/filebeat_pipeline.rs` runs it in process, and +//! Three suites drive the bundled pipeline over the same cases: +//! `tests/integration/filebeat_pipeline.rs` runs it in process, //! `tests/e2e/filebeat_kafka.rs` runs it through the app binary and a real -//! broker. They must agree on what a match IS, or the second one silently -//! grades on a different curve -- so the corpus reader, the ignore list, the -//! accepted divergences and the comparison all live here. -//! -//! The archive is read into memory and NEVER unpacked to the working tree: -//! it carries Elastic-licensed data whose terms travel with it. +//! broker, and `src/bin/pgo_driver.rs` replays the lines as a load. They must +//! agree on what a match IS, so the loader, the input shape and the +//! comparison all live here. -use std::collections::BTreeMap; +use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::OnceLock; +use serde::Deserialize; + /// A repo-relative path, resolved against this crate's root. pub fn repo_path(rel: &str) -> PathBuf { Path::new(env!("CARGO_MANIFEST_DIR")).join(rel) } // ========================================================================= -// The corpus +// The cases // ========================================================================= -/// Every umbrella log in the corpus, in the order the suites replay them. -pub const UMBRELLA_LOGS: &[&str] = &[ - "cisco_umbrella/log/test-umbrella-auditlogs.log", - "cisco_umbrella/log/test-umbrella-cloudfirewalllogs.log", - "cisco_umbrella/log/test-umbrella-dlplogs.log", - "cisco_umbrella/log/test-umbrella-dnslogs.log", - "cisco_umbrella/log/test-umbrella-intrusionlogs.log", - "cisco_umbrella/log/test-umbrella-iplogs.log", - "cisco_umbrella/log/test-umbrella-proxylogs.log", -]; - -/// Every `cisco_ios` log in the corpus. -pub const IOS_LOGS: &[&str] = &[ - "cisco_ios/log/test-asr920.log", - "cisco_ios/log/test-badauth.log", - "cisco_ios/log/test-cisco-ios.log", - "cisco_ios/log/test-date-format-tzoffset.log", - "cisco_ios/log/test-date-format.log", - "cisco_ios/log/test-syslog-header.log", - "cisco_ios/log/test-syslog.log", - "cisco_ios/log/test-tzoffset.log", -]; - -/// Every `cisco_meraki` log in the corpus. -pub const MERAKI_LOGS: &[&str] = &[ - "cisco_meraki/log/test-airmarshal-events.log", - "cisco_meraki/log/test-events.log", - "cisco_meraki/log/test-flows.log", - "cisco_meraki/log/test-ip-flow.log", - "cisco_meraki/log/test-security-events.log", - "cisco_meraki/log/test-urls.log", -]; - -/// Every log file the corpus carries, all three modules. -pub fn all_logs() -> Vec<&'static str> { - let mut out = Vec::with_capacity(UMBRELLA_LOGS.len() + IOS_LOGS.len() + MERAKI_LOGS.len()); - out.extend_from_slice(UMBRELLA_LOGS); - out.extend_from_slice(IOS_LOGS); - out.extend_from_slice(MERAKI_LOGS); - out -} - -/// The committed corpus archive, read once and held in memory. -pub fn corpus() -> &'static BTreeMap> { - static DATA: OnceLock>> = OnceLock::new(); - DATA.get_or_init(|| { - let path = repo_path("tests/fixtures/filebeat/filebeat-testdata.tar.gz"); - let file = std::fs::File::open(&path).expect("fixtures archive must exist"); - let mut archive = tar::Archive::new(flate2::read::GzDecoder::new(file)); - let mut map = BTreeMap::new(); - for entry in archive.entries().expect("read fixtures tar") { - let mut entry = entry.expect("fixtures tar entry"); - if !entry.header().entry_type().is_file() { - continue; - } - let name = entry - .path() - .expect("entry path") - .to_string_lossy() - .trim_start_matches("./") - .to_string(); - let mut buf = Vec::new(); - std::io::Read::read_to_end(&mut entry, &mut buf).expect("read entry"); - map.insert(name, buf); +/// The filebeat module a case's line belongs to. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Module { + Meraki, + Ios, + Umbrella, +} + +/// One input line and what the pipeline must make of it. +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Case { + /// Unique across the file. The e2e suite carries it through the broker to + /// pair each output with its case. + pub name: String, + pub module: Module, + /// The `_conf` object the DFE 2.1 deployment injected (`tz_offset`). + #[serde(default)] + pub conf: Option, + /// The raw log line, as the `message` field. + pub line: String, + /// A subset of the transformed event: every leaf named here must be + /// present and equal in the output. Fields the case does not name are not + /// checked. + pub expect: serde_json::Value, +} + +/// Every case in the file, read once. +pub fn cases() -> &'static [Case] { + static CASES: OnceLock> = OnceLock::new(); + CASES.get_or_init(|| { + let path = repo_path("tests/fixtures/filebeat/cases.yaml"); + let text = std::fs::read_to_string(&path) + .unwrap_or_else(|e| panic!("read {}: {e}", path.display())); + let cases: Vec = + scalo::yaml::from_str(&text, "cases.yaml", scalo::yaml::Profile::Data) + .unwrap_or_else(|e| panic!("parse cases.yaml: {e}")); + let mut seen = HashSet::new(); + for case in &cases { + assert!( + seen.insert(case.name.as_str()), + "duplicate case name {}", + case.name + ); } - assert!(!map.is_empty(), "fixtures archive is empty"); - map + cases }) } -/// One corpus member as text. -pub fn corpus_text(name: &str) -> String { - let bytes = corpus() - .get(name) - .unwrap_or_else(|| panic!("fixture {name} missing from archive")); - String::from_utf8_lossy(bytes).into_owned() -} - -/// The non-blank lines of one corpus log. -pub fn corpus_lines(name: &str) -> Vec { - corpus_text(name) - .lines() - .filter(|l| !l.trim().is_empty()) - .map(str::to_string) - .collect() -} - -/// The golden events for one corpus log. -pub fn corpus_expected(name: &str) -> Vec { - let doc: serde_json::Value = - serde_json::from_str(&corpus_text(name)).expect("expected.json parses"); - doc.get("expected") - .and_then(|e| e.as_array()) - .expect("expected array") - .clone() +/// The cases for one module. +pub fn cases_for(module: Module) -> Vec<&'static Case> { + cases().iter().filter(|c| c.module == module).collect() } -/// The `fields._conf` object for a corpus file (`tz_offset`, `tz_map`, ...): -/// `-config.yml` wins, else the module's `test-common-config.yml` -/// (mirrors the elastic test harness). -pub fn corpus_conf(log_name: &str) -> Option { - let per_file = format!("{log_name}-config.yml"); - let dir = log_name.rsplit_once('/').map_or("", |(d, _)| d); - let common = format!("{dir}/test-common-config.yml"); - for candidate in [per_file, common] { - if let Some(bytes) = corpus().get(&candidate) { - let text = std::str::from_utf8(bytes).expect("config yml is UTF-8"); - let doc: serde_json::Value = - scalo::yaml::from_str(text, &candidate, scalo::yaml::Profile::Data) - .expect("config yml parses"); - if let Some(conf) = doc.get("fields").and_then(|f| f.get("_conf")) { - return Some(conf.clone()); - } - } - } - None +/// One case by name. +pub fn case(name: &str) -> &'static Case { + cases() + .iter() + .find(|c| c.name == name) + .unwrap_or_else(|| panic!("no case named {name}")) } /// A DFE 2.1-shaped input event: `{message, tags, timestamp}` plus the /// `_conf` object the 2.1 deployment injected (`tz_offset`, `tz_map`). /// -/// This is the shape a transform reads off `{source}_land`, so both the -/// in-process suite and the broker round-trip build their input with it. +/// This is the shape a transform reads off `{source}_land`, so every suite +/// builds its input with it. pub fn input_event(message: &str, conf: Option<&serde_json::Value>) -> serde_json::Value { let mut obj = serde_json::json!({ "message": message, @@ -171,6 +115,11 @@ pub fn input_event(message: &str, conf: Option<&serde_json::Value>) -> serde_jso obj } +/// The input event for one case. +pub fn case_event(case: &Case) -> serde_json::Value { + input_event(&case.line, case.conf.as_ref()) +} + /// An IOS access-list line whose body is 5000 spaces, on which onig panics /// inside the filebeat program's grok. pub fn ios_grok_panic_line() -> String { @@ -181,198 +130,21 @@ pub fn ios_grok_panic_line() -> String { } // ========================================================================= -// What counts as agreeing with the goldens +// What counts as agreeing with a case // ========================================================================= -/// Paths where the port deliberately diverges from the elastic goldens. -/// geoip enrichment is stripped (dfe-loader owns geoip), event.ingested -/// is elastic-harness-dynamic, log.file.path carries our routing seed. -pub const IGNORED_PATHS: &[&str] = &[ - "source.geo", - "source.as", - "destination.geo", - "destination.as", - "client.geo", - "client.as", - "event.ingested", - "log.file.path", - // VRL's parse_user_agent (woothee/ua-parser) classifies differently - // from elastic's user_agent processor ("Mac OSX" vs "Mac OS X", - // device "pc" vs "Mac"). Engine-library parity, not a port defect; - // user_agent.original stays asserted. - "user_agent.name", - "user_agent.os", - "user_agent.device", - "user_agent.version", - "user_agent.major", - "user_agent.minor", - "user_agent.patch", -]; - -fn ignored(path: &str) -> bool { - IGNORED_PATHS - .iter() - .any(|p| path == *p || path.starts_with(&format!("{p}."))) -} - -/// Narrow per-line divergences from the goldens, each with a cause that is -/// library parity or an elastic-integration feature the 2.1 templates -/// never had (see pipelines/filebeat/README.md, Limitations). -pub const ACCEPTED_MISMATCHES: &[(&str, usize, &str)] = &[ - // VRL's public-suffix data includes the PSL private section - // (*.servicebus.windows.net); elastic's registered_domain processor - // uses the ICANN section only. - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 6, - "dns.question.registered_domain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 6, - "dns.question.subdomain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 6, - "dns.question.top_level_domain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 9, - "dns.question.registered_domain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 9, - "dns.question.subdomain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 9, - "dns.question.top_level_domain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 10, - "dns.question.registered_domain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 10, - "dns.question.subdomain", - ), - ( - "cisco_umbrella/log/test-umbrella-dnslogs.log", - 10, - "dns.question.top_level_domain", - ), - // VRL parse_url normalises an empty path to "/"; elastic keeps "". - ( - "cisco_umbrella/log/test-umbrella-dlplogs.log", - 0, - "url.path", - ), - // VRL is_nullish(" ") is true, so the template skips user_agent - // entirely for whitespace-only UA strings; elastic still emits - // user_agent.original " " with name "Other". Library semantics. - ( - "cisco_umbrella/log/test-umbrella-proxylogs.log", - 7, - "user_agent", - ), - ( - "cisco_umbrella/log/test-umbrella-proxylogs.log", - 8, - "user_agent", - ), - ( - "cisco_umbrella/log/test-umbrella-proxylogs.log", - 9, - "user_agent", - ), - ( - "cisco_umbrella/log/test-umbrella-proxylogs.log", - 10, - "user_agent", - ), - ( - "cisco_umbrella/log/test-umbrella-proxylogs.log", - 11, - "user_agent", - ), -]; - -fn accepted_mismatch(file: &str, idx: usize, path: &str) -> bool { - ACCEPTED_MISMATCHES - .iter() - .any(|(f, i, p)| *f == file && *i == idx && *p == path) -} - -/// Whether a corpus file's timestamps may differ in the year alone. -/// -/// `cisco_ios` only: its classic "Feb 8 04:00:48" format carries no year, so -/// the parser infers the CURRENT year while the goldens embed their -/// generation year. -pub fn allows_year_swap(log_name: &str) -> bool { - log_name.starts_with("cisco_ios/") -} - -/// Timestamps match at millisecond precision, with the year-swap tolerance -/// applied when the corpus file has no year to parse. -fn timestamps_match(ours: &str, exp: &str, allow_year_swap: bool) -> bool { - use chrono::{DateTime, Datelike, SubsecRound}; - let (Ok(a), Ok(b)) = ( - DateTime::parse_from_rfc3339(ours), - DateTime::parse_from_rfc3339(exp), - ) else { - return false; - }; - let a = a.trunc_subsecs(3); - let b = b.trunc_subsecs(3); - if a == b { - return true; - } - if !allow_year_swap { - return false; - } - let (Some(a), Some(b)) = (a.with_year(2000), b.with_year(2000)) else { - return false; - }; - a == b -} - -/// Canonical query-string compare: same key -> multiset of values, -/// independent of parameter order (the 2.1 template rebuilds url.query -/// from a VRL object, which alphabetises; elastic keeps raw order). -fn query_params(q: &str) -> BTreeMap> { - let mut map: BTreeMap> = BTreeMap::new(); - for pair in q.split('&').filter(|s| !s.is_empty()) { - let (k, v) = pair.split_once('=').unwrap_or((pair, "")); - map.entry(k.to_string()).or_default().push(v.to_string()); - } - for values in map.values_mut() { - values.sort(); - } - map -} - -/// Compare one transformed event against its elastic golden: every leaf in -/// `expected` (bar the documented ignores) must be present and equal in -/// ours. Extra fields on our side (DFE 2.1 event carriage like `_conf`, -/// `timestamp`) are fine. The root `tags` array compares as a set (the -/// 2.1 template seeds tags in a different order). +/// Compare one transformed event against a case's `expect`: every leaf in +/// `expected` must be present and equal in `ours`. Extra fields on our side +/// (DFE 2.1 event carriage like `_conf`, `timestamp`) are fine. The root +/// `tags` array compares as a set, because the 2.1 template seeds tags in a +/// fixed but arbitrary order. pub fn diff_expected( path: &str, - exp: &serde_json::Value, + expected: &serde_json::Value, ours: &serde_json::Value, - allow_year_swap: bool, problems: &mut Vec, ) { - if ignored(path) { - return; - } - match exp { + match expected { serde_json::Value::Object(map) => { for (k, v) in map { let child = if path.is_empty() { @@ -381,85 +153,45 @@ pub fn diff_expected( format!("{path}.{k}") }; match ours.get(k) { - Some(o) => diff_expected(&child, v, o, allow_year_swap, problems), - None => { - if !ignored(&child) { - problems.push(format!("{child}: missing (expected {v})")); - } - } + Some(o) => diff_expected(&child, v, o, problems), + None => problems.push(format!("{child}: missing (expected {v})")), } } } serde_json::Value::Array(items) if path == "tags" => { - let ours_set: Vec<&serde_json::Value> = ours - .as_array() - .map(|a| a.iter().collect()) - .unwrap_or_default(); + let have = ours.as_array().map(Vec::as_slice).unwrap_or_default(); for item in items { - if !ours_set.contains(&item) { + if !have.contains(item) { problems.push(format!("tags: missing {item}")); } } } serde_json::Value::Array(items) => match ours.as_array() { - Some(ours_arr) if ours_arr.len() == items.len() => { - for (i, (e, o)) in items.iter().zip(ours_arr).enumerate() { - diff_expected(&format!("{path}[{i}]"), e, o, allow_year_swap, problems); + Some(have) if have.len() == items.len() => { + for (i, (e, o)) in items.iter().zip(have).enumerate() { + diff_expected(&format!("{path}[{i}]"), e, o, problems); } } - _ => problems.push(format!("{path}: expected array {exp}, got {ours}")), + _ => problems.push(format!("{path}: expected array {expected}, got {ours}")), }, leaf => { - if path == "@timestamp" || path.ends_with(".@timestamp") { - let ours_s = ours.as_str().unwrap_or_default(); - let exp_s = leaf.as_str().unwrap_or_default(); - if !timestamps_match(ours_s, exp_s, allow_year_swap) { - problems.push(format!("{path}: expected {exp_s}, got {ours_s}")); - } - } else if path == "url.query" { - let ours_s = ours.as_str().unwrap_or_default(); - let exp_s = leaf.as_str().unwrap_or_default(); - if query_params(ours_s) != query_params(exp_s) { - problems.push(format!( - "{path}: params differ, expected {exp_s}, got {ours_s}" - )); - } - } else if ours != leaf { + if ours != leaf { problems.push(format!("{path}: expected {leaf}, got {ours}")); } } } } -/// Everything wrong with `ours` as the transform of `log_name[idx]`, after -/// the documented ignores and that line's accepted divergences are removed. -/// Empty means it agrees with the golden. -pub fn problems_against_golden( - log_name: &str, - idx: usize, - expected: &serde_json::Value, - ours: &serde_json::Value, -) -> Vec { +/// Everything wrong with `ours` as the transform of `case`. Empty means it +/// agrees. +pub fn problems_for(case: &Case, ours: &serde_json::Value) -> Vec { let mut problems = Vec::new(); - diff_expected( - "", - expected, - ours, - allows_year_swap(log_name), - &mut problems, - ); + diff_expected("", &case.expect, ours, &mut problems); problems - .into_iter() - .filter(|p| { - let path = p.split(':').next().unwrap_or_default(); - !accepted_mismatch(log_name, idx, path) - }) - .collect() } /// Whether the pipeline routed this event to the shape-unmatched catch-all. -/// Every corpus line belongs to a known module, so this must never fire on -/// one. +/// Every case belongs to a known module, so this must never fire on one. pub fn is_unmatched(event: &serde_json::Value) -> bool { event .get("tags") diff --git a/tests/e2e.rs b/tests/e2e.rs index 678472e..6dbaef1 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -24,7 +24,7 @@ mod common; #[allow(clippy::unwrap_used, clippy::expect_used)] #[path = "common/filebeat.rs"] -mod filebeat_corpus; +mod filebeat_cases; #[allow(clippy::expect_used, clippy::panic)] #[path = "e2e/filebeat_kafka.rs"] diff --git a/tests/e2e/filebeat_kafka.rs b/tests/e2e/filebeat_kafka.rs index d0d1181..c3b97b2 100644 --- a/tests/e2e/filebeat_kafka.rs +++ b/tests/e2e/filebeat_kafka.rs @@ -1,15 +1,15 @@ // Project: dfe-transform-vrl // File: tests/e2e/filebeat_kafka.rs -// Purpose: The filebeat corpus through the app and a broker +// Purpose: The filebeat cases through the app and a broker // Language: Rust // // License: BUSL-1.1 // Copyright: (c) 2026 HYPERI PTY LIMITED -//! The transform hop of the filebeat chain, proved on real data. +//! The transform hop of the filebeat chain, proved on synthetic cases. //! //! ```text -//! filebeat corpus -> filebeat_land -> dfe-transform-vrl -> filebeat_load +//! filebeat cases -> filebeat_land -> dfe-transform-vrl -> filebeat_load //! ``` //! //! What makes this an acceptance test rather than another unit test: @@ -20,9 +20,9 @@ //! - The topics are the real source-bound names. `filebeat_land` in and //! `filebeat_load` out is what a `filebeat` Source compiles to, and the //! app derives the source name back out of the input topic. -//! - The data is the elastic/integrations corpus, and the output is graded -//! against its golden events by the same comparison the in-process suite -//! uses (`tests/common/filebeat.rs`). +//! - The data is `tests/fixtures/filebeat/cases.yaml`, and each output is +//! graded against the fields its case names by the same comparison the +//! in-process suite uses (`tests/common/filebeat.rs`). //! //! The broker is a container this test starts and drops, never a live one: //! `filebeat_land` on a shared broker is somebody else's data. @@ -37,7 +37,7 @@ use scalo::transport::kafka::{KafkaAdmin, KafkaConfig, KafkaProfile, KafkaTransp use scalo::transport::{TransportBase, TransportReceiver, TransportSender}; use crate::common::{self, KafkaTestConfig}; -use crate::filebeat_corpus as fb; +use crate::filebeat_cases as fb; /// The source this instance is bound to, and the topics that follow from it. const SOURCE: &str = "filebeat"; @@ -52,41 +52,26 @@ const UNMATCHED_PROBE: &str = "ws21 probe: a line no filebeat module claims"; /// consumer group. Debug builds spend most of it in the VRL compiler. const READY_TIMEOUT: Duration = Duration::from_secs(240); -/// How long the corpus gets to arrive on the sink topic once the app is ready. +/// How long the cases get to arrive on the sink topic once the app is ready. const DRAIN_TIMEOUT: Duration = Duration::from_secs(180); -/// One golden event, and the corpus line it grades. -struct Golden { - log: &'static str, - idx: usize, - event: serde_json::Value, -} - -/// Every corpus line as the app will read it off `filebeat_land`, paired with -/// the goldens they must produce. -fn corpus_workload() -> (Vec, Vec) { - let mut inputs = Vec::new(); - let mut goldens = Vec::new(); - - for log in fb::all_logs() { - let lines = fb::corpus_lines(log); - let expected = fb::corpus_expected(&format!("{log}-expected.json")); - assert_eq!( - lines.len(), - expected.len(), - "{log}: {} input lines but {} golden events", - lines.len(), - expected.len() - ); - let conf = fb::corpus_conf(log); - for (idx, (line, event)) in lines.iter().zip(expected).enumerate() { - inputs.push(fb::input_event(line, conf.as_ref())); - goldens.push(Golden { log, idx, event }); - } - } - +/// Every case as the app will read it off `filebeat_land`, plus the shape- +/// unmatched probe. +/// +/// Each case event carries its name in a root `_case` field. The pipeline keeps +/// root fields on every branch, so the name comes out the far side and pairs +/// an output with its case whatever order the broker returns them in. +fn workload() -> Vec { + let mut inputs: Vec = fb::cases() + .iter() + .map(|case| { + let mut event = fb::case_event(case); + event["_case"] = serde_json::json!(case.name); + event + }) + .collect(); inputs.push(fb::input_event(UNMATCHED_PROBE, None)); - (inputs, goldens) + inputs } fn kafka_config(kf: &KafkaTestConfig, topics: &[&str], group: &str) -> KafkaConfig { @@ -212,70 +197,43 @@ async fn drain( out } -/// Match every golden to an output event, consuming each output once. -/// -/// The sink topic has one partition, so the output usually arrives in the -/// order it was produced; that position is tried first and the scan is the -/// fallback. Returns the goldens that nothing produced, each with the -/// closest candidate's problems. -fn match_goldens(goldens: &[Golden], outputs: &[serde_json::Value]) -> Vec { - let mut used = vec![false; outputs.len()]; - let mut unmatched = Vec::new(); - - for (position, golden) in goldens.iter().enumerate() { - let mut order: Vec = Vec::with_capacity(outputs.len()); - if position < outputs.len() { - order.push(position); - } - order.extend((0..outputs.len()).filter(|i| *i != position)); - - let mut closest: Option> = None; - let mut matched = false; - for candidate in order { - if used[candidate] { - continue; +/// Grade every case against the output event that carries its name. +/// Returns one line per case that has no output or whose output disagrees. +fn grade_cases(outputs: &[serde_json::Value]) -> Vec { + let mut failures = Vec::new(); + for case in fb::cases() { + let found = outputs + .iter() + .find(|e| e.get("_case").and_then(serde_json::Value::as_str) == Some(&case.name)); + match found { + None => failures.push(format!("{}: no output event carried its name", case.name)), + Some(event) if fb::is_unmatched(event) => { + failures.push(format!("{}: routed to filebeat_unmatched", case.name)); } - let problems = fb::problems_against_golden( - golden.log, - golden.idx, - &golden.event, - &outputs[candidate], - ); - if problems.is_empty() { - used[candidate] = true; - matched = true; - break; - } - if closest.as_ref().is_none_or(|c| problems.len() < c.len()) { - closest = Some(problems); + Some(event) => { + let problems = fb::problems_for(case, event); + if !problems.is_empty() { + failures.push(format!("{}: {}", case.name, problems.join("; "))); + } } } - - if !matched { - let detail = closest.map_or_else( - || "no output event was left to compare against".to_string(), - |problems| problems.join("; "), - ); - unmatched.push(format!("{}[{}]: {detail}", golden.log, golden.idx)); - } } - unmatched + failures } #[tokio::test] -async fn filebeat_corpus_round_trips_through_kafka() { - let Some(env) = common::KafkaTestEnv::hermetic("filebeat-corpus-round-trip").await else { +async fn filebeat_cases_round_trip_through_kafka() { + let Some(env) = common::KafkaTestEnv::hermetic("filebeat-cases-round-trip").await else { common::require_service_in_ci("Kafka", "no Docker to start a broker container"); eprintln!("SKIP: no Docker, so this test cannot own a broker."); return; }; let kf = env.config(); - let (inputs, goldens) = corpus_workload(); + let inputs = workload(); eprintln!( - "replaying {} corpus lines from {} logs through {LAND_TOPIC} -> {LOAD_TOPIC}", - inputs.len(), - fb::all_logs().len() + "replaying {} events through {LAND_TOPIC} -> {LOAD_TOPIC}", + inputs.len() ); let admin = @@ -382,17 +340,14 @@ async fn filebeat_corpus_round_trips_through_kafka() { outputs[probe_at] ); - // Assertion 3 -- every corpus line agrees with its elastic golden, after - // the port's documented divergences. - let mut corpus_outputs = outputs; - corpus_outputs.remove(probe_at); - let unmatched = match_goldens(&goldens, &corpus_outputs); + // Assertion 3 -- every case agrees with the fields it names. + let failures = grade_cases(&outputs); assert!( - unmatched.is_empty(), - "{} of {} corpus events did not match their golden:\n{}", - unmatched.len(), - goldens.len(), - unmatched.join("\n") + failures.is_empty(), + "{} of {} cases did not match:\n{}", + failures.len(), + fb::cases().len(), + failures.join("\n") ); // Assertion 4 -- all three module branches actually ran. A pipeline that @@ -400,7 +355,7 @@ async fn filebeat_corpus_round_trips_through_kafka() { // everything through unmatched. let branch_ran = |module: &str, probe: &dyn Fn(&serde_json::Value) -> bool| { assert!( - corpus_outputs.iter().any(probe), + outputs.iter().any(probe), "no output event came from the {module} branch" ); }; diff --git a/tests/fixtures/filebeat/README.md b/tests/fixtures/filebeat/README.md index c903900..29ef5c8 100644 --- a/tests/fixtures/filebeat/README.md +++ b/tests/fixtures/filebeat/README.md @@ -1,61 +1,15 @@ -# Filebeat pipeline test corpus +# Filebeat pipeline test cases -Real filebeat/Elastic-integration pipeline test data used by -`tests/integration/filebeat_pipeline.rs` to exercise the bundled INTERIM -filebeat-compat pipeline (`pipelines/filebeat/filebeat.vrl`). +`cases.yaml` holds the synthetic log lines that `tests/integration/filebeat_pipeline.rs`, `tests/e2e/filebeat_kafka.rs` and the PGO workload (`src/bin/pgo_driver.rs`) replay through the bundled filebeat-compat pipeline (`pipelines/filebeat/filebeat.vrl`). The lines are written for this repo. Hosts, addresses and names come from the documentation ranges (`192.0.2.0/24`, `198.51.100.0/24`, `203.0.113.0/24`, `2001:db8::/32`, `example.*`). Earlier revisions replayed third-party sample data, which is no longer part of the repo. -## Contents +Each case has: -`filebeat-testdata.tar.gz` (read in-memory at test time, never unpacked to -the working tree): +- `name`: unique, and carried through the broker in the e2e suite to pair an output with its case. +- `module`: `meraki`, `ios` or `umbrella`. +- `conf`: optional `_conf` object the DFE 2.1 deployment injected (`tz_offset`). +- `line`: the raw log line, which becomes `message`. +- `expect`: the fields the output must carry. Every leaf named must be present and equal. Fields not named are not checked, and the root `tags` array compares as a set. -- `cisco_umbrella/log/` -- raw Cisco Umbrella export samples for all seven - log types (audit, cloud firewall, dlp, dns, intrusion, ip, proxy) plus - the upstream `-expected.json` golden outputs and `-config.yml` knobs. -- `cisco_ios/log/` -- Cisco IOS syslog samples (with and without syslog - priority headers, several date formats and tz offsets) plus goldens. -- `cisco_meraki/log/` -- Cisco Meraki syslog samples (flows, events, - airmarshal, security, urls, ip-flows) plus goldens. -- `cisco_*/manifest.yml` -- upstream package manifests (package versions, - provenance). -- `LICENSE.txt` -- upstream root licence pointer. -- `licenses/Elastic-2.0.txt` -- the full Elastic License 2.0 text the - pointer refers to (redistribution requires shipping the terms). +## Adding a case -## Provenance - -- Source repo: `github.com/elastic/integrations` -- Commit: `c7bc53023288c7d34afd631c0c3e3552f6219419` (pinned in - `scripts/filebeat/fetch_testdata.py` -- the same vintage the DFE 2.1 VRL - templates were generated against) -- Paths: `packages/cisco_{umbrella,ios,meraki}/data_stream/log/_dev/test/pipeline/` -- Licence: Elastic License 2.0 (test data redistributed unmodified for - compatibility testing, see `LICENSE.txt` inside the archive) - -## Refreshing - -Re-fetch at a new pin with the gh CLI (contents API, read-only), then -re-create the archive: - -```bash -python3 scripts/filebeat/fetch_testdata.py /tmp/filebeat-fetch # edit SHA first -tar --sort=name --owner=0 --group=0 --numeric-owner \ - --mtime='2026-08-18 00:00:00 UTC' \ - -czf tests/fixtures/filebeat/filebeat-testdata.tar.gz \ - -C /tmp/filebeat-fetch \ - cisco_umbrella cisco_ios cisco_meraki LICENSE.txt licenses -``` - -Those tar flags make the archive byte-reproducible, so a refresh only churns -the blob when content actually changed. Re-validate the -`ACCEPTED_MISMATCHES` line indices in `tests/common/filebeat.rs` after any -refresh - they are keyed by line position within each log file. - -## Golden-output caveats - -The `-expected.json` files are the outputs of the ELASTIC ingest pipelines -at that pin. This repo's VRL is the DFE 2.1 port, which deliberately -differs: geoip enrichment is stripped (dfe-loader owns geoip), `related.*` -/ ECS versions may lag the newest integration revisions, and events carry -DFE 2.1 conventions (`_conf.tz_offset`, tag seeding). Tests therefore -assert on stable parsed fields, not whole-object equality. +Write a line for the module, run it through the pipeline, read the output, and put the fields you have checked by hand into `expect`. A Cisco IOS line with no year in its timestamp gets the current year from the parser, so leave `@timestamp` out of `expect` for those. diff --git a/tests/fixtures/filebeat/cases.yaml b/tests/fixtures/filebeat/cases.yaml new file mode 100644 index 0000000..f7d6eee --- /dev/null +++ b/tests/fixtures/filebeat/cases.yaml @@ -0,0 +1,715 @@ +# Synthetic filebeat cases, replayed by the integration and e2e suites and the PGO workload. +# See README.md in this directory for the shape of a case. +- name: umbrella-dns-allowed + module: umbrella + line: |- + "2026-03-04 09:15:22","alice.example","alice.example,Laptop 17","192.0.2.10","198.51.100.7","Allowed","1 (A)","NOERROR","www.example.net.","Business Services,Search Engines","Office Policy","AD Users, Roaming Computers","" + expect: + '@timestamp': '2026-03-04T09:15:22Z' + observer: + vendor: Cisco + product: Umbrella + type: dns + log: + file: + path: /filebeat/cisco_umbrella/dnslogs + event: + action: dns-request-Allowed + category: network + kind: event + type: + - allowed + - connection + dns: + type: query + response_code: NOERROR + question: + name: www.example.net + registered_domain: example.net + subdomain: www + top_level_domain: net + type: 1 (A) + source: + ip: 192.0.2.10 + nat: + ip: 198.51.100.7 + cisco: + umbrella: + identity: alice.example + tags: + - preserve_original_event + +- name: umbrella-dns-blocked-aaaa + module: umbrella + line: |- + "2026-03-04 09:16:02","alice.example","alice.example,Laptop 17","192.0.2.10","198.51.100.8","Blocked","28 (AAAA)","NOERROR","ads.tracker.example.org.","Advertisements,Block List","Office Policy","Roaming Computers","BlockedCategories" + expect: + '@timestamp': '2026-03-04T09:16:02Z' + event: + action: dns-request-Blocked + type: + - denied + - connection + dns: + question: + name: ads.tracker.example.org + registered_domain: example.org + subdomain: ads.tracker + top_level_domain: org + type: 28 (AAAA) + cisco: + umbrella: + categories: + - Advertisements + - Block List + blocked_categories: + - BlockedCategories + +- name: umbrella-dns-nxdomain-mx + module: umbrella + line: |- + "2026-03-04 09:17:40","Site Router","Site Router","192.0.2.1","203.0.113.53","Allowed","15 (MX)","NXDOMAIN","mail.corp.example.co.uk.","Email","Default Policy","Networks","" + expect: + '@timestamp': '2026-03-04T09:17:40Z' + dns: + response_code: NXDOMAIN + question: + name: mail.corp.example.co.uk + registered_domain: example.co.uk + subdomain: mail.corp + top_level_domain: co.uk + type: 15 (MX) + source: + ip: 192.0.2.1 + +- name: umbrella-dns-txt + module: umbrella + line: |- + "2026-03-05 21:02:11","bob.example","bob.example,Desktop 4","192.0.2.44","198.51.100.9","Allowed","16 (TXT)","NOERROR","_dmarc.example.com.","Infrastructure","Office Policy","AD Users","" + expect: + '@timestamp': '2026-03-05T21:02:11Z' + dns: + question: + name: _dmarc.example.com + registered_domain: example.com + subdomain: _dmarc + type: 16 (TXT) + +- name: umbrella-proxy-allowed-chrome + module: umbrella + line: |- + "2026-03-04 10:01:15","Branch Network","192.0.2.77","198.51.100.31","203.0.113.80","","ALLOWED","https://files.example.net/reports/q1.pdf","https://portal.example.net/home","Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36","200","612","20480","","","","","","","","","","Networks" + expect: + '@timestamp': '2026-03-04T10:01:15Z' + observer: + type: proxy + event: + type: + - allowed + - connection + source: + ip: 192.0.2.77 + nat: + ip: 198.51.100.31 + destination: + ip: 203.0.113.80 + http: + request: + bytes: 612 + referrer: https://portal.example.net/home + response: + bytes: 20480 + status_code: 200 + url: + original: https://files.example.net/reports/q1.pdf + scheme: https + domain: files.example.net + port: 443 + path: /reports/q1.pdf + extension: pdf + user_agent: + original: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 + name: Chrome + version: 122.0.0.0 + +- name: umbrella-proxy-blocked-query + module: umbrella + line: |- + "2026-03-04 10:02:49","Branch Network","192.0.2.78","198.51.100.31","203.0.113.81","","BLOCKED","http://download.example.org/setup.exe?src=mail&id=42","","curl/8.5.0","403","310","0","","","","","","","","","","Networks" + expect: + '@timestamp': '2026-03-04T10:02:49Z' + event: + type: + - denied + - connection + http: + response: + status_code: 403 + url: + scheme: http + domain: download.example.org + port: 80 + path: /setup.exe + extension: exe + query: id=42&src=mail + user_agent: + original: curl/8.5.0 + +- name: umbrella-ip-allowed + module: umbrella + line: |- + "2026-03-04 11:20:46","alice.example","192.0.2.10","51234","198.51.100.60","8443","Business Services" + expect: + '@timestamp': '2026-03-04T11:20:46Z' + observer: + type: firewall + event: + type: + - connection + source: + ip: 192.0.2.10 + port: 51234 + destination: + ip: 198.51.100.60 + port: 8443 + +- name: umbrella-ip-zero-ports + module: umbrella + line: |- + "2026-03-04 11:20:47","alice.example","192.0.2.10","0","198.51.100.61","0","Infrastructure" + expect: + source: + port: 0 + destination: + ip: 198.51.100.61 + port: 0 + +- name: umbrella-audit-update + module: umbrella + line: |- + "7000000001","2026-02-11 14:30:05","admin@example.com","Admin User", "policies", "update","198.51.100.200","enabled: false","enabled: true" + expect: + '@timestamp': '2026-02-11T14:30:05Z' + event: + id: '7000000001' + action: update + category: configuration + type: + - change + user: + email: admin@example.com + name: Admin User + source: + ip: 198.51.100.200 + cisco: + umbrella: + audit: + type: policies + before_values: + enabled: 'false' + after_values: + enabled: 'true' + +- name: umbrella-cloudfirewall-plain + module: umbrella + line: |- + 2026-03-04 12:00:01,[900000001],Branch Tunnel,Office Tunnel Device,OUTBOUND,6,512,192.0.2.20,,198.51.100.90,,lhr1.edc,14,ALLOW + expect: + '@timestamp': '2026-03-04T12:00:01Z' + observer: + type: firewall + event: + action: fw-connection-ALLOW + type: + - allowed + - connection + network: + direction: outbound + transport: '6' + source: + ip: 192.0.2.20 + bytes: 512 + destination: + ip: 198.51.100.90 + rule: + id: '14' + cisco: + umbrella: + datacenter: lhr1.edc + +- name: umbrella-cloudfirewall-quoted + module: umbrella + line: |- + "2026-03-04 12:00:09","[900000001]","Branch Tunnel", "Office Tunnel Device","INBOUND","17","96","192.0.2.21","","198.51.100.91", "","lhr1.edc","9","BLOCK","example.com,example.net","12,34" + expect: + '@timestamp': '2026-03-04T12:00:09Z' + event: + action: fw-connection-BLOCK + type: + - denied + - connection + network: + direction: inbound + transport: '17' + rule: + id: '9' + related: + hosts: + - example.com + - example.net + cisco: + umbrella: + fqdns: + - example.com + - example.net + destination_lists_id: 12,34 + +- name: umbrella-intrusion + module: umbrella + line: |- + "2026-03-04 12:30:00","Branch Firewall","Network Tunnels","1","21000","EXAMPLE-SERVER overflow attempt against test service","4455","HIGH","Attempted Administrator Privilege Gain","cve-2020-0001","TCP","40000","198.51.100.5","41234","192.0.2.30","443","Would Block" + expect: + '@timestamp': '2026-03-04T12:30:00Z' + observer: + type: idps + event: + kind: alert + action: ips-Would-Block + category: + - network + - intrusion_detection + network: + protocol: TCP + source: + ip: 198.51.100.5 + port: 41234 + destination: + ip: 192.0.2.30 + port: 443 + cisco: + umbrella: + gid: '1' + sid: '21000' + severity: HIGH + +- name: umbrella-dlp + module: umbrella + line: |- + "2026-03-04 12:45:30","Real Time","aaaaaaaa-0000-4000-8000-000000000001_bbbbbbbb-0000-4000-8000-000000000002","CRITICAL","Network1","","budget.xlsx","ExampleDrive","https://drive.example.net/upload","BLOCK","rule-7","classification-9","classifier-9.1","text/html","512","0000000000000000000000000000000000000000000000000000000000000000", "Confidential" + expect: + '@timestamp': '2026-03-04T12:45:30Z' + observer: + type: dlp + event: + action: dlp-BLOCK + category: + - network + - file + file: + name: budget.xlsx + size: 512 + mime_type: text/html + rule: + name: rule-7 + network: + application: ExampleDrive + url: + domain: drive.example.net + path: /upload + cisco: + umbrella: + severity: CRITICAL + +- name: ios-acl-host-header-tcp + module: ios + conf: {"tz_offset": "UTC"} + line: |- + Mar 5 02:11:08 192.0.2.1 12345: Mar 5 02:11:07.512: %SEC-6-IPACCESSLOGP: list 120 denied tcp 198.51.100.20(40012) -> 203.0.113.9(443), 3 packets + expect: + observer: + vendor: Cisco + product: IOS + type: firewall + cisco: + ios: + access_list: '120' + facility: SEC + sequence: '12345' + event: + action: deny + code: IPACCESSLOGP + severity: 6 + timezone: UTC + type: + - info + - denied + log: + level: informational + network: + transport: tcp + type: ipv4 + packets: 3 + source: + ip: 198.51.100.20 + port: 40012 + packets: 3 + destination: + ip: 203.0.113.9 + port: 443 + message: list 120 denied tcp 198.51.100.20(40012) -> 203.0.113.9(443), 3 packets + tags: + - preserve_original_event + +- name: ios-acl-named-icmp + module: ios + conf: {"tz_offset": "UTC"} + line: |- + Mar 6 02:11:08 192.0.2.1 12346: Mar 6 02:11:07.512: %SEC-6-IPACCESSLOGDP: list EDGE-IN denied icmp 198.51.100.21 -> 192.0.2.1 (8/0), 2 packets + expect: + cisco: + ios: + access_list: EDGE-IN + facility: SEC + event: + action: deny + code: IPACCESSLOGDP + icmp: + type: '8' + code: '0' + network: + transport: icmp + packets: 2 + source: + ip: 198.51.100.21 + destination: + ip: 192.0.2.1 + +- name: ios-acl-ipv6 + module: ios + conf: {"tz_offset": "UTC"} + line: |- + Mar 7 08:00:00 192.0.2.1 12350: Mar 7 08:00:00.001: %IPV6-6-ACCESSLOGP: list ACL-V6-IN/10 permitted tcp 2001:db8:10::5(51000) -> 2001:db8:20::7(22), 4 packets + expect: + cisco: + ios: + access_list: ACL-V6-IN/10 + facility: IPV6 + event: + action: allow + code: ACCESSLOGP + network: + transport: tcp + type: ipv6 + source: + ip: 2001:db8:10::5 + port: 51000 + destination: + ip: 2001:db8:20::7 + port: 22 + +- name: ios-config-change + module: ios + conf: {"tz_offset": "UTC"} + line: |- + <189>4021: Mar 5 2026 10:22:09.113: %SYS-5-CONFIG_I: Configured from console by admin on vty1 (192.0.2.55) + expect: + '@timestamp': '2026-03-05T10:22:09.113Z' + cisco: + ios: + facility: SYS + message_count: 4021 + event: + code: CONFIG_I + sequence: 4021 + severity: 5 + log: + level: notification + syslog: + priority: 189 + message: Configured from console by admin on vty1 (192.0.2.55) + +- name: ios-link-down-hostname + module: ios + conf: {"tz_offset": "UTC"} + line: |- + <190>: edge-sw02: Mar 5 2026 10:30:00.004: %LINK-3-UPDOWN: Interface GigabitEthernet0/1, changed state to down + expect: + '@timestamp': '2026-03-05T10:30:00.004Z' + cisco: + ios: + facility: LINK + event: + code: UPDOWN + severity: 3 + log: + level: error + syslog: + hostname: edge-sw02 + priority: 190 + message: Interface GigabitEthernet0/1, changed state to down + +- name: ios-ntp-tz-abbreviation + module: ios + conf: {"tz_offset": "UTC"} + line: |- + <191>7001: core-rt1: Mar 5 2026 10:31:12.000 CEST: NTP Core (NOTICE): Clock synchronization lost. + expect: + event: + sequence: 7001 + log: + syslog: + hostname: core-rt1 + priority: 191 + message: 'NTP Core (NOTICE): Clock synchronization lost.' + +- name: ios-date-no-fraction-offset + module: ios + conf: {"tz_offset": "-0600"} + line: |- + <190>7002: sw01: Mar 8 2026 22:11:43: %FOO-6-BAR: Synthetic message for the date format. + expect: + '@timestamp': '2026-03-09T04:11:43Z' + event: + code: BAR + timezone: '-0600' + log: + syslog: + hostname: sw01 + message: Synthetic message for the date format. + +- name: ios-md5-badauth + module: ios + conf: {"tz_offset": "UTC"} + line: |- + <190>9100: Mar 9 20:07:53.081: %TCP-6-BADAUTH: No MD5 digest from 192.0.2.50(60001) to 198.51.100.18(179) (RST) tableid - 0 + expect: + cisco: + ios: + facility: TCP + tableid: '0' + event: + code: BADAUTH + sequence: 9100 + source: + ip: 192.0.2.50 + port: 60001 + destination: + ip: 198.51.100.18 + port: 179 + +- name: meraki-flows-allow + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000000.100000001 MX_branch1 flows allow src=192.0.2.15 dst=198.51.100.44 mac=AA:BB:CC:00:11:22 protocol=udp sport=51000 dport=5060 + expect: + '@timestamp': '2026-02-25T06:13:20.100000001Z' + cisco_meraki: + event_type: flows + event_subtype: flow_allowed + event: + action: layer3-firewall-allowed-flow + category: + - network + network: + protocol: udp + observer: + hostname: MX_branch1 + source: + ip: 192.0.2.15 + mac: AA-BB-CC-00-11-22 + port: 51000 + destination: + ip: 198.51.100.44 + port: 5060 + tags: + - preserve_original_event + - forwarded + +- name: meraki-flows-pattern + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000001.200000002 MX_branch1 flows src=192.0.2.16 dst=198.51.100.45 protocol=tcp sport=44000 dport=8080 pattern: 1 all + expect: + '@timestamp': '2026-02-25T06:13:21.200000002Z' + cisco_meraki: + event_subtype: ip_session_initiated + firewall: + pattern: 1 all + event: + action: ip-session-initiated + network: + protocol: tcp + source: + ip: 192.0.2.16 + port: 44000 + destination: + ip: 198.51.100.45 + port: 8080 + +- name: meraki-flows-icmp6 + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000002.300000003 AP_lobby flows allow src=fe80::aaaa:bbbb:cccc:1 dst=ff02::1:ff00:2 mac=AA:BB:CC:00:11:33 protocol=icmp6 type=135 + expect: + network: + protocol: icmp6 + observer: + hostname: AP_lobby + source: + ip: fe80::aaaa:bbbb:cccc:1 + mac: AA-BB-CC-00-11-33 + destination: + ip: ff02::1:ff00:2 + +- name: meraki-urls-agent + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000010.400000004 MX_branch1 urls src=192.0.2.15:52000 dst=198.51.100.46:443 mac=AA:BB:CC:00:11:22 agent='Mozilla/5.0 (X11; Linux x86_64; rv:123.0) Gecko/20100101 Firefox/123.0' request: GET https://www.example.net/index.html + expect: + cisco_meraki: + event_type: urls + event_subtype: http_access + event: + action: http-access + category: + - network + - web + http: + request: + method: GET + url: + original: https://www.example.net/index.html + scheme: https + domain: www.example.net + port: 443 + path: /index.html + extension: html + registered_domain: example.net + subdomain: www + user_agent: + name: Firefox + version: '123.0' + +- name: meraki-urls-no-agent + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000011.500000005 MX_branch1 urls src=192.0.2.15:52001 dst=198.51.100.47:80 mac=AA:BB:CC:00:11:22 request: GET http://www.example.org/robots.txt + expect: + source: + ip: 192.0.2.15 + port: 52001 + url: + original: http://www.example.org/robots.txt + scheme: http + port: 80 + extension: txt + cisco_meraki: + urls: + mac: AA-BB-CC-00-11-22 + +- name: meraki-events-wpa-auth + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000020.600000006 MR_floor2 events type=wpa_auth radio='1' vap='2' client_mac='AA:BB:CC:00:11:44' aid='1000000001' + expect: + cisco_meraki: + event_type: events + event_subtype: wpa_auth + wpa_auth: + client_mac: AA:BB:CC:00:11:44 + aid: '1000000001' + event: + action: wifi-wpa-authentication + category: + - network + - authentication + log: + syslog: + priority: 134 + observer: + hostname: MR_floor2 + +- name: meraki-events-dfs + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000021.700000007 MR_floor2 events type=dfs_event channel='100' radio='1' + expect: + cisco_meraki: + event_subtype: dfs_event + dfs_event: + channel: '100' + radio: '1' + event: + action: dynamic-frequency-selection-detected + +- name: meraki-ip-flow-start + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000030.800000008 MX100 ip_flow_start src=192.0.2.15 dst=198.51.100.48 protocol=tcp sport=34000 dport=443 translated_src_ip=203.0.113.2 translated_port=34000 + expect: + cisco_meraki: + event_type: ip_flow_start + network: + protocol: tcp + source: + port: 34000 + destination: + ip: 198.51.100.48 + port: 443 + +- name: meraki-security-ids + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000040.900000009 MX84 security_event ids_alerted signature=1:99999:1 priority=2 timestamp=1772000040.890000 direction=ingress protocol=tcp/ip src=198.51.100.49:80 dst=192.0.2.15:56000 decision=allowed message: EXAMPLE-WEB synthetic signature for testing + expect: + cisco_meraki: + event_type: security_event + event_subtype: ids_alerted + security: + signature: 1:99999:1 + decision: allowed + event: + action: ids-signature-matched + category: + - network + - intrusion_detection + network: + direction: ingress + protocol: tcp/ip + source: + ip: 198.51.100.49 + port: 80 + destination: + ip: 192.0.2.15 + port: 56000 + message: EXAMPLE-WEB synthetic signature for testing + +- name: meraki-airmarshal-rogue + module: meraki + conf: {"tz_offset": "UTC"} + line: |- + <134>1 1772000050.000000010 MX84_5 airmarshal_events type=rogue_ssid_detected ssid='' bssid='AA:BB:CC:DD:EE:01' src='AA:BB:CC:DD:EE:01' dst='FF:FF:FF:FF:FF:FF' wired_mac='AA:BB:CC:DD:EE:02' vlan_id='0' channel='6' rssi='30' fc_type='0' fc_subtype='8' + expect: + cisco_meraki: + event_type: airmarshal_events + event_subtype: rogue_ssid_detected + bssid: AA:BB:CC:DD:EE:01 + channel: '6' + event: + action: rogue-ssid-detected + source: + mac: AA-BB-CC-DD-EE-01 + destination: + mac: FF-FF-FF-FF-FF-FF + network: + vlan: + id: '0' diff --git a/tests/fixtures/filebeat/filebeat-testdata.tar.gz b/tests/fixtures/filebeat/filebeat-testdata.tar.gz deleted file mode 100644 index 001a9b2..0000000 Binary files a/tests/fixtures/filebeat/filebeat-testdata.tar.gz and /dev/null differ diff --git a/tests/integration.rs b/tests/integration.rs index 6ab34c7..17db4f0 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -37,12 +37,12 @@ mod config; #[path = "integration/enrichment.rs"] mod enrichment; -// Shared with the e2e binary, which grades the same corpus through a real +// Shared with the e2e binary, which grades the same cases through a real // broker. Declared here rather than under `common/mod.rs` so the integration // binary does not pull in the container helpers it never uses. #[allow(clippy::unwrap_used, clippy::expect_used)] #[path = "common/filebeat.rs"] -mod filebeat_corpus; +mod filebeat_cases; #[allow(clippy::unwrap_used, clippy::expect_used)] #[path = "integration/enrichment_contract.rs"] diff --git a/tests/integration/direct_transport.rs b/tests/integration/direct_transport.rs index 3b92ee1..14a3246 100644 --- a/tests/integration/direct_transport.rs +++ b/tests/integration/direct_transport.rs @@ -37,7 +37,7 @@ use scalo::worker::{AdaptiveWorkerPool, BatchEngine, WorkerPoolConfig}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; -use crate::filebeat_corpus as fb; +use crate::filebeat_cases as fb; use crate::offline::binary; use crate::ports; diff --git a/tests/integration/filebeat_pipeline.rs b/tests/integration/filebeat_pipeline.rs index 51afdc1..26731d6 100644 --- a/tests/integration/filebeat_pipeline.rs +++ b/tests/integration/filebeat_pipeline.rs @@ -10,14 +10,13 @@ //! (`pipelines/filebeat/filebeat.vrl`), ported from the DFE 2.1 Vector //! templates (dfe-vector-templates 14x family). //! -//! The corpus tests drive the pipeline with REAL filebeat test data from -//! elastic/integrations (pinned SHA, see -//! `tests/fixtures/filebeat/README.md`), read in memory from the committed -//! tar.gz at test time. +//! The case tests drive the pipeline with the synthetic lines in +//! `tests/fixtures/filebeat/cases.yaml` and check each output against the +//! subset of fields its case names. //! //! This runs the pipeline IN PROCESS. `tests/e2e/filebeat_kafka.rs` runs the -//! same corpus through the app binary and a real broker; both grade against -//! the shared comparison in `tests/common/filebeat.rs`. +//! same cases through the app binary and a real broker; both grade with the +//! shared comparison in `tests/common/filebeat.rs`. use std::sync::OnceLock; @@ -28,7 +27,7 @@ use dfe_transform_vrl::enrichment::EnrichmentRegistry; use vrl::compiler::Program; use vrl::value::Value; -use crate::filebeat_corpus as fb; +use crate::filebeat_cases as fb; /// Compile the bundled pipeline with the bundled timezones table. /// @@ -91,23 +90,27 @@ fn bundled_filebeat_pipeline_compiles() { } // ========================================================================= -// Module detection smoke tests (single real lines per module) +// Module detection smoke tests (one line per module) // ========================================================================= -const MERAKI_LINE: &str = "<134>1 1647478988.289402144 MX84_4 flows allow src=10.0.2.170 \ - dst=10.0.0.34 mac=00:7C:2D:BD:76:F2 protocol=udp sport=54841 dport=15600"; +fn meraki_line() -> &'static str { + &fb::case("meraki-flows-allow").line +} -const IOS_LINE: &str = "Feb 8 04:00:48 192.168.100.2 585917: Feb 8 04:00:47.272: \ - %SEC-6-IPACCESSLOGRP: list 177 denied igmp 192.168.100.197 -> 224.0.0.22, 1 packet"; +fn ios_line() -> &'static str { + &fb::case("ios-acl-host-header-tcp").line +} -const UMBRELLA_LINE: &str = r#""2020-07-23 23:49:54","elasticuser","elasticuser,Elastic Machine","192.168.1.1","81.2.69.144","Allowed","1 (A)","NOERROR","www.elastic.co.","Software/Technology,Business Services,Application","Test Policy Name","AD Users, Roaming Computers","""#; +fn umbrella_line() -> &'static str { + &fb::case("umbrella-dns-allowed").line +} const UNMATCHED_LINE: &str = "completely unrecognisable payload"; #[test] fn detects_and_parses_meraki_flows_line() { let program = compile_filebeat(); - let mut value = event(MERAKI_LINE, Some("UTC")); + let mut value = event(meraki_line(), Some("UTC")); run_vrl(program, &mut value).expect("meraki flows line must transform"); assert_eq!( str_field(&value, "cisco_meraki.event_type").as_deref(), @@ -124,7 +127,7 @@ fn detects_and_parses_meraki_flows_line() { #[test] fn detects_and_parses_cisco_ios_line() { let program = compile_filebeat(); - let mut value = event(IOS_LINE, Some("UTC")); + let mut value = event(ios_line(), Some("UTC")); run_vrl(program, &mut value).expect("cisco ios line must transform"); assert_eq!( str_field(&value, "observer.product").as_deref(), @@ -136,7 +139,7 @@ fn detects_and_parses_cisco_ios_line() { #[test] fn detects_and_parses_umbrella_dns_line() { let program = compile_filebeat(); - let mut value = event(UMBRELLA_LINE, None); + let mut value = event(umbrella_line(), None); run_vrl(program, &mut value).expect("umbrella dns line must transform"); assert_eq!( str_field(&value, "observer.product").as_deref(), @@ -172,23 +175,25 @@ fn unmatched_event_is_tagged_and_passed_through() { // Event carriage // ========================================================================= -/// One real line per branch, for tests that care which branch ran rather than +/// One line per branch, for tests that care which branch ran rather than /// what it produced. -const BRANCH_LINES: &[(&str, &str)] = &[ - ("meraki", MERAKI_LINE), - ("ios", IOS_LINE), - ("umbrella", UMBRELLA_LINE), - ("unmatched", UNMATCHED_LINE), -]; +fn branch_lines() -> [(&'static str, &'static str); 4] { + [ + ("meraki", meraki_line()), + ("ios", ios_line()), + ("umbrella", umbrella_line()), + ("unmatched", UNMATCHED_LINE), + ] +} #[test] fn a_root_field_survives_every_branch_but_a_tag_does_not() { - // A corpus replay marks its own events so a shared cluster's older rows + // A case replay marks its own events so a shared cluster's older rows // are not counted as this run's. `tags` cannot carry that marker: every // module branch assigns `.tags` outright before it does anything else, so // a root field is the only carriage that comes out the far side. let program = compile_filebeat(); - for (branch, line) in BRANCH_LINES { + for (branch, line) in branch_lines() { let mut input = fb::input_event(line, Some(&serde_json::json!({ "tz_offset": "UTC" }))); input["_e2e"] = serde_json::json!({ "run": "ws21" }); input["tags"] = serde_json::json!(["ws21"]); @@ -200,7 +205,7 @@ fn a_root_field_survives_every_branch_but_a_tag_does_not() { Some("ws21"), "the {branch} branch dropped the root marker: {value:?}" ); - if *branch != "unmatched" { + if branch != "unmatched" { let kept_the_tag = field(&value, "tags") .and_then(Value::as_array) .is_some_and(|tags| tags.contains(&Value::from("ws21"))); @@ -213,101 +218,70 @@ fn a_root_field_survives_every_branch_but_a_tag_does_not() { } // ========================================================================= -// Real-data corpus (elastic/integrations pipeline test data, pinned; see -// tests/fixtures/filebeat/README.md) +// Synthetic cases (tests/fixtures/filebeat/cases.yaml) // ========================================================================= -/// Run every line of a corpus log through the pipeline and compare against -/// the goldens. Returns human-readable problem strings (empty = pass). -fn run_corpus_file(log_name: &str) -> Vec { +/// Run every case of one module through the pipeline and compare against +/// what each names. Returns human-readable problem strings (empty = pass). +fn run_cases(module: fb::Module) -> Vec { let program = compile_filebeat(); - let lines = fb::corpus_lines(log_name); - let expected = fb::corpus_expected(&format!("{log_name}-expected.json")); - let conf = fb::corpus_conf(log_name); let mut problems = Vec::new(); - if lines.len() != expected.len() { - problems.push(format!( - "{log_name}: {} input lines but {} golden events", - lines.len(), - expected.len() - )); - } - - for (idx, (line, exp)) in lines.iter().zip(&expected).enumerate() { - let mut value = event_with_conf(line, conf.as_ref()); + for case in fb::cases_for(module) { + let mut value = Value::from(fb::case_event(case)); if let Err(e) = run_vrl(program, &mut value) { - problems.push(format!("{log_name}[{idx}]: pipeline failed: {e}")); + problems.push(format!("{}: pipeline failed: {e}", case.name)); continue; } let ours = serde_json::to_value(&value).expect("event serialises"); if fb::is_unmatched(&ours) { - problems.push(format!("{log_name}[{idx}]: routed to filebeat_unmatched")); + problems.push(format!("{}: routed to filebeat_unmatched", case.name)); continue; } problems.extend( - fb::problems_against_golden(log_name, idx, exp, &ours) + fb::problems_for(case, &ours) .into_iter() - .map(|p| format!("{log_name}[{idx}]: {p}")), + .map(|p| format!("{}: {p}", case.name)), ); } problems } -fn assert_corpus_clean(log_names: &[&str]) { - let mut problems = Vec::new(); - for name in log_names { - problems.extend(run_corpus_file(name)); +#[test] +fn cases_cover_each_module_and_say_what_they_expect() { + for module in [fb::Module::Meraki, fb::Module::Ios, fb::Module::Umbrella] { + assert!(!fb::cases_for(module).is_empty(), "no cases for {module:?}"); } + for case in fb::cases() { + assert!( + case.expect.as_object().is_some_and(|o| !o.is_empty()), + "case {} expects nothing, so it can never fail", + case.name + ); + } +} + +fn assert_cases_clean(module: fb::Module) { + let problems = run_cases(module); assert!( problems.is_empty(), - "corpus mismatches ({}):\n{}", + "case mismatches ({}):\n{}", problems.len(), problems.join("\n") ); } #[test] -fn corpus_umbrella_dnslogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-dnslogs.log"]); -} - -#[test] -fn corpus_umbrella_proxylogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-proxylogs.log"]); -} - -#[test] -fn corpus_umbrella_auditlogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-auditlogs.log"]); -} - -#[test] -fn corpus_umbrella_iplogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-iplogs.log"]); -} - -#[test] -fn corpus_umbrella_dlplogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-dlplogs.log"]); -} - -#[test] -fn corpus_umbrella_intrusionlogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-intrusionlogs.log"]); -} - -#[test] -fn corpus_umbrella_cloudfirewalllogs() { - assert_corpus_clean(&["cisco_umbrella/log/test-umbrella-cloudfirewalllogs.log"]); +fn cases_umbrella() { + assert_cases_clean(fb::Module::Umbrella); } #[test] -fn corpus_cisco_ios() { - assert_corpus_clean(fb::IOS_LOGS); +fn cases_cisco_ios() { + assert_cases_clean(fb::Module::Ios); } #[test] -fn corpus_cisco_meraki() { - assert_corpus_clean(fb::MERAKI_LOGS); +fn cases_cisco_meraki() { + assert_cases_clean(fb::Module::Meraki); }