fix(tests): replace the filebeat test archive with synthetic cases

.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''',
 ]

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 = [

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`

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

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"

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"

scripts/pgo-workload.sh

@@ -10,8 +10,8 @@
 # Usage:
 #   scripts/pgo-workload.sh <path-to-dfe-transform-vrl-binary>
 #
-# 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

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<T: std::str::FromStr>(key: &str, default: T) -> T {
 }
 
 // ===========================================================================
-// Corpus
+// Cases
 // ===========================================================================
 
-/// Every corpus line as the input event the bundled pipeline reads.
-fn corpus_events() -> Vec<Vec<u8>> {
-    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<Vec<u8>> {
+    let events: Vec<Vec<u8>> = 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,

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`.