diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cee8a79..2a24aca 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -67,7 +67,7 @@ jobs: uses: hyperi-io/hyperi-ci/.github/workflows/rust-ci.yml@main with: publish-target: both - # The filebeat corpus test reads the pipeline and corpus from this repo, beside the checkout. + # The filebeat test reads the pipeline and cases from this repo, beside the checkout. sibling-checkouts: hyperi-io/dfe-transform-vrl tag: ${{ inputs.tag || '' }} from-head: ${{ inputs.from-head || '' }} diff --git a/Cargo.lock b/Cargo.lock index e692bd2..0413047 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -551,11 +551,9 @@ version = "1.0.13" dependencies = [ "anyhow", "bytes", - "chrono", "clap", "dotenvy", "figment", - "flate2", "metrics", "metrics-util", "nix", @@ -564,7 +562,6 @@ dependencies = [ "schemars", "serde", "serde_json", - "tar", "tempfile", "thiserror", "tikv-jemallocator", @@ -741,16 +738,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" @@ -3279,17 +3266,6 @@ dependencies = [ "windows", ] -[[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 = "tempfile" version = "3.27.0" @@ -4261,16 +4237,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 = "yansi" version = "1.0.1" diff --git a/Cargo.toml b/Cargo.toml index f73a66b..39e0db8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -102,10 +102,6 @@ pgo-driver = [] [dev-dependencies] tempfile = ">=3.27.0, <4" tokio-test = ">=0.4.6, <0.5" -# Reads dfe-transform-vrl's filebeat corpus tar.gz in memory. -flate2 = ">=1.1.10, <2" -tar = ">=0.4.46, <0.5" -chrono = ">=0.4.45, <0.5" scalo = { version = ">=2.14.30, <3", features = [ "transport-kafka", "transport-kafka-test-support", diff --git a/tests/TESTING.md b/tests/TESTING.md index f4a620a..5352b0b 100644 --- a/tests/TESTING.md +++ b/tests/TESTING.md @@ -67,10 +67,9 @@ bare Vector process against a broker. ## The WS21 acceptance case `e2e::filebeat_kafka` runs in core, on the shared broker, which no live cluster's data shares. -It seeds the elastic/integrations filebeat corpus onto `filebeat_land`, runs the -shipped binary over it, and grades `filebeat_load` against elastic's golden -events. +It seeds the synthetic filebeat cases onto `filebeat_land`, runs the shipped +binary over them, and grades `filebeat_load` against the fields each case names. -The corpus, the bundled pipeline and the list of documented divergences belong +The cases (`tests/fixtures/filebeat/cases.yaml`) and the bundled pipeline belong to dfe-transform-vrl, so the case needs that repo checked out beside this one, or `DFE_TRANSFORM_VRL_DIR` pointing at it. Without one it prints a skip. diff --git a/tests/common/filebeat.rs b/tests/common/filebeat.rs index b6439b6..157683b 100644 --- a/tests/common/filebeat.rs +++ b/tests/common/filebeat.rs @@ -1,29 +1,30 @@ +// 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)] // Project: dfe-transform-vector // File: tests/common/filebeat.rs -// Purpose: The filebeat corpus and its golden comparison, read from dfe-transform-vrl +// Purpose: The filebeat test cases and their comparison, read from dfe-transform-vrl // Language: Rust // // License: BUSL-1.1 // Copyright: (c) 2026 HYPERI PTY LIMITED -//! The elastic/integrations filebeat corpus, and what counts as agreeing -//! with it. +//! The synthetic filebeat cases in dfe-transform-vrl's +//! `tests/fixtures/filebeat/cases.yaml`, and what counts as agreeing with one. //! -//! Both transform apps run the SAME bundled pipeline over the SAME data in -//! WS21, and a parity test only means something if both grade on the same -//! curve. dfe-transform-vrl owns the corpus archive, the bundled VRL and the -//! list of documented divergences from elastic's goldens; this is the reader -//! for them, and it mirrors `dfe-transform-vrl/tests/common/filebeat.rs`. -//! -//! The archive is read into memory and NEVER unpacked to the working tree: -//! it carries Elastic-licensed data whose terms travel with it. +//! Both transform apps run the SAME bundled pipeline over the SAME cases, and a +//! parity test only means something if both grade on the same curve. +//! dfe-transform-vrl owns the cases and the bundled VRL, and this is the reader +//! for them. It mirrors `dfe-transform-vrl/tests/common/filebeat.rs`. -use std::collections::BTreeMap; +use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::OnceLock; -/// The dfe-transform-vrl checkout that owns the corpus and the pipeline. +use serde::Deserialize; + +/// The dfe-transform-vrl checkout that owns the cases and the pipeline. /// /// `DFE_TRANSFORM_VRL_DIR` first, then a `dfe-transform-vrl` sitting beside /// any ancestor of this checkout -- which finds it from a plain clone and from @@ -56,133 +57,62 @@ pub fn timezones_csv() -> Option { vrl_repo().map(|dir| dir.join("pipelines/filebeat/timezones.csv")) } -/// The corpus archive. -pub fn corpus_archive() -> Option { - vrl_repo().map(|dir| dir.join("tests/fixtures/filebeat/filebeat-testdata.tar.gz")) -} - // ========================================================================= -// 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 corpus archive, read once and held in memory. -pub fn corpus() -> &'static BTreeMap> { - static DATA: OnceLock>> = OnceLock::new(); - DATA.get_or_init(|| { - let path = corpus_archive().expect("dfe-transform-vrl checkout must be present"); - let file = std::fs::File::open(&path).expect("corpus archive must be readable"); - let mut archive = tar::Archive::new(flate2::read::GzDecoder::new(file)); - let mut map = BTreeMap::new(); - for entry in archive.entries().expect("read corpus tar") { - let mut entry = entry.expect("corpus 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. Callers check [`vrl_repo`] first. +pub fn cases() -> &'static [Case] { + static CASES: OnceLock> = OnceLock::new(); + CASES.get_or_init(|| { + let path = vrl_repo() + .expect("dfe-transform-vrl checkout must be present") + .join("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(), "corpus 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 `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 -} - /// A DFE 2.1-shaped input event: `{message, tags, timestamp}` plus the /// `_conf` object the 2.1 deployment injected (`tz_offset`, `tz_map`). pub fn input_event(message: &str, conf: Option<&serde_json::Value>) -> serde_json::Value { @@ -197,199 +127,27 @@ pub fn input_event(message: &str, conf: Option<&serde_json::Value>) -> serde_jso obj } -// ========================================================================= -// What counts as agreeing with the goldens -// ========================================================================= - -/// 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 the pipeline's own README in dfe-transform-vrl). -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 +/// The input event for one case. +pub fn case_event(case: &Case) -> serde_json::Value { + input_event(&case.line, case.conf.as_ref()) } -/// 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 -} +// ========================================================================= +// What counts as agreeing with a case +// ========================================================================= -/// 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, and the fields -/// Vector's kafka source adds) 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`, and the fields Vector's +/// kafka source adds) 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() { @@ -398,83 +156,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 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 ffff2b7..19eebdf 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -11,9 +11,9 @@ mod common; -/// The filebeat corpus and its golden comparison, owned by dfe-transform-vrl. +/// The filebeat cases and their comparison, owned by dfe-transform-vrl. #[path = "common/filebeat.rs"] -mod filebeat_corpus; +mod filebeat_cases; #[path = "e2e/mod.rs"] mod e2e; diff --git a/tests/e2e/filebeat_kafka.rs b/tests/e2e/filebeat_kafka.rs index 090e450..fed5918 100644 --- a/tests/e2e/filebeat_kafka.rs +++ b/tests/e2e/filebeat_kafka.rs @@ -1,6 +1,6 @@ // Project: dfe-transform-vector // File: tests/e2e/filebeat_kafka.rs -// Purpose: WS21 parity: the filebeat corpus through the app, Vector and a broker +// Purpose: WS21 parity: the filebeat cases through the app, Vector and a broker // Language: Rust // // License: BUSL-1.1 @@ -9,11 +9,11 @@ //! The transform hop of the WS21 chain, with Vector doing the transforming. //! //! ```text -//! filebeat corpus -> filebeat_land -> dfe-transform-vector -> filebeat_load +//! filebeat cases -> filebeat_land -> dfe-transform-vector -> filebeat_load //! ``` //! -//! This is the parity half of the acceptance case: the same corpus, the same -//! bundled filebeat VRL, the same golden events, and the same assertions as +//! This is the parity half of the acceptance case: the same cases, the same +//! bundled filebeat VRL, and the same assertions as //! `dfe-transform-vrl/tests/e2e/filebeat_kafka.rs`. Only the transform app //! differs, which is what makes WS21 a test of the transform LAYER rather //! than of one app. @@ -30,9 +30,8 @@ //! - The app runs `vector validate` before starting, so a pipeline Vector's //! VRL cannot compile fails here rather than silently passing nothing on. //! -//! The corpus, the pipeline and the divergence list belong to -//! dfe-transform-vrl. Without that checkout there is nothing to run against -//! and the test says so. +//! The cases and the pipeline belong to dfe-transform-vrl. Without that +//! checkout there is nothing to run against and the test says so. //! //! NOT covered here, because it needs a cluster: the receiver hop that picks //! the source, and the loader hop that lands rows in `ClickHouse`. @@ -43,7 +42,7 @@ use scalo::transport::kafka::{KafkaAdmin, KafkaConfig, KafkaTransport}; use scalo::transport::{TransportBase, TransportReceiver, TransportSender}; use crate::common::{self, KafkaFixture}; -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"; @@ -58,41 +57,26 @@ const UNMATCHED_PROBE: &str = "ws21 probe: a line no filebeat module claims"; /// the 5.5K-line pipeline, and start Vector. 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 Vector 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 Vector 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 client_config(base: &KafkaConfig, topics: &[&str], group: &str) -> KafkaConfig { @@ -296,65 +280,39 @@ 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_vector() { +async fn filebeat_cases_round_trip_through_vector() { if fb::vrl_repo().is_none() { common::require_service_in_ci( "dfe-transform-vrl checkout", - "the corpus and the bundled pipeline grade this test, and CI clones the repo beside this one", + "the cases and the bundled pipeline grade this test, and CI clones the repo beside this one", ); eprintln!( - "SKIP: the corpus and the bundled pipeline live in dfe-transform-vrl. \ + "SKIP: the cases and the bundled pipeline live in dfe-transform-vrl. \ Point DFE_TRANSFORM_VRL_DIR at a checkout to run this." ); return; @@ -364,18 +322,17 @@ async fn filebeat_corpus_round_trips_through_vector() { eprintln!("SKIP: no Vector binary (run scripts/fetch-vector.sh)"); return; }; - let Some(fixture) = KafkaFixture::hermetic("filebeat-corpus-round-trip").await else { + let Some(fixture) = KafkaFixture::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 base = fixture.config.clone(); - let (inputs, goldens) = corpus_workload(); + let inputs = workload(); eprintln!( - "WS21: replaying {} corpus lines from {} logs through {LAND_TOPIC} -> {LOAD_TOPIC}", - inputs.len(), - fb::all_logs().len() + "WS21: replaying {} events through {LAND_TOPIC} -> {LOAD_TOPIC}", + inputs.len() ); let admin = @@ -503,23 +460,22 @@ async fn filebeat_corpus_round_trips_through_vector() { 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. + // Assertion 4 -- all three module branches actually ran. A pipeline that + // detected nothing would still satisfy the count above by passing + // 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" ); };