diff --git a/config.example.yaml b/config.example.yaml index 13a69c9..5a7cb41 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -93,7 +93,7 @@ dfe_source: "my_source" # batch: # max_events: 10000 # Vector-level batch size (default 10K) # # max_bytes: 8388608 # optional byte limit, at most 16 MiB on direct, where unset means 8 MiB -# timeout_secs: 1 # flush timeout, at most 1 on direct +# timeout_secs: 1 # flush timeout, at most 1 on direct. On bus the profile's linger.ms replaces it # sasl: # enabled: true # mechanism: "scram_sha_512" diff --git a/docs/config-schema.json b/docs/config-schema.json index 0aa3684..c592cf5 100644 --- a/docs/config-schema.json +++ b/docs/config-schema.json @@ -606,7 +606,7 @@ "default": null }, "timeout_secs": { - "description": "Max batch age (seconds) before flush. Default: 1, and at most 1 on the\ndirect transport.", + "description": "Max batch age (seconds) before flush. Default: 1, and at most 1 on the\ndirect transport. On the bus transport it applies only where no\nlibrdkafka layer sets `linger.ms`, which the production profile does.", "type": "integer", "format": "uint32", "minimum": 0, diff --git a/docs/config-schema.yaml b/docs/config-schema.yaml index f72197e..0ce8701 100644 --- a/docs/config-schema.yaml +++ b/docs/config-schema.yaml @@ -582,7 +582,8 @@ $defs: timeout_secs: description: |- Max batch age (seconds) before flush. Default: 1, and at most 1 on the - direct transport. + direct transport. On the bus transport it applies only where no + librdkafka layer sets `linger.ms`, which the production profile does. type: integer format: uint32 minimum: 0 diff --git a/docs/librdkafka.md b/docs/librdkafka.md index 825dc2d..ec4741b 100644 --- a/docs/librdkafka.md +++ b/docs/librdkafka.md @@ -23,7 +23,7 @@ Vector applies `librdkafka_options` over the settings it writes itself from the | `message.timeout.ms`, `socket.timeout.ms` | `sink.message_timeout_ms` (0, no limit), `sink.socket_timeout_ms` (60000) | | `queue.buffering.max.ms`, `batch.num.messages`, `batch.size` | `sink.batch.timeout_secs`, `.max_events` and, where set, `.max_bytes` | -`linger.ms` is librdkafka's other name for `queue.buffering.max.ms`. rdkafka applies a client's options from a hash map, so with the production profile's `linger.ms` beside Vector's `queue.buffering.max.ms`, which one a producer ends up with is not fixed. +`linger.ms` is librdkafka's other name for `queue.buffering.max.ms`, and a client built from both takes whichever its hash map applied last. Vector always writes `sink.batch.timeout_secs` as `queue.buffering.max.ms`, so the supervisor moves a batching delay that a layer names into `batch.timeout_secs` and leaves both spellings out of `librdkafka_options`. A layer that names either spelling replaces the other spelling from every layer below it, and within one layer the spelling that sorts last wins. A layer therefore outranks `sink.batch.timeout_secs`. The production profile names `linger.ms`, so on the bus transport `timeout_secs` has no effect until a central profile drops that key, and the supervisor logs a warning when `timeout_secs` is not 1 and a layer overrides it. A few keys are derived from the big dials and written after the merge, so `librdkafka_options` cannot reach them: @@ -48,9 +48,9 @@ sink: ## Example -For a producer with no central file, `sink.compression: lz4` and `sink.librdkafka_options: {linger.ms: "20"}`, the supervisor writes `linger.ms=20`, `compression.type=lz4`, `queue.buffering.max.kbytes=262144`, `enable.idempotence=true`, `socket.nagle.disable=true`, `statistics.interval.ms=1000` and `message.max.bytes=16777216`, the last because no layer set it. +For a producer with no central file, `sink.compression: lz4` and `sink.librdkafka_options: {linger.ms: "20"}`, the supervisor writes `batch.timeout_secs: 0.02` on the sink and, in `librdkafka_options`, `compression.type=lz4`, `queue.buffering.max.kbytes=262144`, `enable.idempotence=true`, `socket.nagle.disable=true`, `statistics.interval.ms=1000` and `message.max.bytes=16777216`, the last because no layer set it. -Add a central `producer.production` naming only `linger.ms: "50"` and `socket.nagle.disable` and `statistics.interval.ms` drop out of `librdkafka_options`, because the central profile replaced the constants rather than merging over them. Vector still writes `statistics.interval.ms=1000` itself. `linger.ms` stays `20`: the per-instance option outranks the central file. Drop that option and it becomes `50`. +Add a central `producer.production` naming only `linger.ms: "50"` and `socket.nagle.disable` and `statistics.interval.ms` drop out of `librdkafka_options`, because the central profile replaced the constants rather than merging over them. Vector still writes `statistics.interval.ms=1000` itself. The delay stays 20 ms: the per-instance option outranks the central file. Drop that option and it becomes 50 ms. `dfe-transform-vector --config assemble ` writes `90_sink.yaml` and `00_source.yaml`, so you can read the result for any config. @@ -69,7 +69,7 @@ This app reads `CONSUMER_PRODUCTION` for the source and `PRODUCER_PRODUCTION` fo | Producer setting | Value | librdkafka default | Justification | |---------|-------|--------------------|---------------| -| `linger.ms` | `100` | `5` | Larger batches -- 100 ms of latency for higher throughput. See Layers for how it meets Vector's `queue.buffering.max.ms`. | +| `linger.ms` | `100` | `5` | Larger batches -- 100 ms of latency for higher throughput. This is the delay the sink uses, in place of `sink.batch.timeout_secs`. See Layers. | | `compression.type` | `zstd` | `none` | Best compression ratio with acceptable CPU cost. `sink.compression` replaces it on every sink. | | `socket.nagle.disable` | `true` | `true` | Already librdkafka's default. The profile states it. | | `statistics.interval.ms` | `1000` | `0` | librdkafka metrics. | diff --git a/pipelines/filebeat/filebeat.yaml b/pipelines/filebeat/filebeat.yaml index f4d58e4..789108d 100644 --- a/pipelines/filebeat/filebeat.yaml +++ b/pipelines/filebeat/filebeat.yaml @@ -6,8 +6,8 @@ # License: BUSL-1.1 # Copyright: (c) 2026 HYPERI PTY LIMITED # -# A pre-canned transform file, dropped into `transforms.dir` as it stands: the -# wrapper writes it into Vector's config dir verbatim, so the top-level +# A pre-canned transform file for `transforms.dir`. The wrapper parses it and +# writes it into Vector's config dir with its wired `inputs`, so the top-level # `enrichment_tables` key reaches Vector and the VRL's timezone lookups resolve. # `inputs: [dfe_source]` binds it to whatever Kafka topic `dfe_source` derives, # and the wrapper wires the transform's output to `dfe_sink`. @@ -17,10 +17,10 @@ # path or points `path` at its own copy; startup validation names the path when # the file is not there. # -# The VRL body is byte-identical to dfe-transform-vrl pipelines/filebeat/ -# filebeat.vrl @ e0b7087, which is where it is maintained. Refresh by copying +# The VRL body is dfe-transform-vrl pipelines/filebeat/filebeat.vrl @ e0b7087, +# bar two lines of its header comment, and that repo maintains it. Refresh by copying # that file again rather than editing the copy here, so the two transform apps -# keep grading against the same corpus. +# keep passing the same cases. enrichment_tables: timezones: diff --git a/src/config/generate.rs b/src/config/generate.rs index ea34af7..698201d 100644 --- a/src/config/generate.rs +++ b/src/config/generate.rs @@ -27,7 +27,8 @@ use std::collections::{BTreeMap, HashMap}; use scalo::transport::kafka::MESSAGE_MAX_BYTES; -use serde_json::{Map, Value}; +use serde_json::{Map, Number, Value}; +use tracing::warn; /// A YAML mapping in the generated Vector config. type Mapping = Map; @@ -323,6 +324,7 @@ fn direct_sink_component( build_batch_block( &sink.batch, Some(sink.batch.max_bytes.unwrap_or(DIRECT_BATCH_MAX_BYTES)), + sink.batch.timeout_secs.into(), ), ); @@ -367,10 +369,28 @@ fn kafka_sink_component(sink: &SinkConfig, inputs: &[String], acknowledged: bool Value::Number(sink.socket_timeout_ms.into()), ); - // Vector-level batch + // librdkafka options — production defaults + user overrides + let mut rdkafka = build_sink_librdkafka(sink); + + // Vector always writes `batch.timeout_secs` as `queue.buffering.max.ms`, + // which librdkafka reads as `linger.ms`. A layer that names the delay + // moves into that field, so the sink states it once. + let timeout_secs = match take_batching_delay(&mut rdkafka) { + Some(layered) => { + if sink.batch.timeout_secs != BatchConfig::default().timeout_secs { + warn!( + sink_batch_timeout_secs = sink.batch.timeout_secs, + layered_delay_secs = %layered, + "sink.batch.timeout_secs is ignored: a librdkafka layer sets the batching delay" + ); + } + layered + } + None => sink.batch.timeout_secs.into(), + }; component.insert( key("batch"), - build_batch_block(&sink.batch, sink.batch.max_bytes), + build_batch_block(&sink.batch, sink.batch.max_bytes, timeout_secs), ); // SASL @@ -386,21 +406,35 @@ fn kafka_sink_component(sink: &SinkConfig, inputs: &[String], acknowledged: bool // Buffer component.insert(key("buffer"), build_buffer_block(&sink.buffer)); - // librdkafka options — production defaults + user overrides - let rdkafka = build_sink_librdkafka(sink); component.insert(key("librdkafka_options"), Value::Object(rdkafka)); component } +/// Remove the batching delay from a sink's librdkafka options and return it in +/// seconds, the unit of Vector's `batch.timeout_secs`. +/// +/// `None`, with the options untouched, when no layer names it or the value is +/// not a number of milliseconds. librdkafka refuses such a value at startup. +fn take_batching_delay(rdkafka: &mut Mapping) -> Option { + let name = ["linger.ms", "queue.buffering.max.ms"] + .into_iter() + .find(|name| rdkafka.contains_key(*name))?; + let millis: f64 = rdkafka.get(name)?.as_str()?.trim().parse().ok()?; + if !millis.is_finite() || millis < 0.0 { + return None; + } + let seconds = Number::from_f64(millis / 1000.0)?; + rdkafka.remove("linger.ms"); + rdkafka.remove("queue.buffering.max.ms"); + Some(seconds) +} + /// A sink's Vector-level `batch` block, with `max_bytes` written when given. -fn build_batch_block(batch: &BatchConfig, max_bytes: Option) -> Value { +fn build_batch_block(batch: &BatchConfig, max_bytes: Option, timeout_secs: Number) -> Value { let mut block = Mapping::new(); block.insert(key("max_events"), Value::Number(batch.max_events.into())); - block.insert( - key("timeout_secs"), - Value::Number(batch.timeout_secs.into()), - ); + block.insert(key("timeout_secs"), Value::Number(timeout_secs)); if let Some(max_bytes) = max_bytes { block.insert(key("max_bytes"), Value::Number(max_bytes.into())); } @@ -1352,7 +1386,6 @@ mod tests { assert!(text.contains("codec: json")); assert!(text.contains("compression: zstd")); // Production librdkafka defaults (from scalo kafka_config) - assert!(text.contains("linger.ms: \"100\"")); assert!(text.contains("compression.type: zstd")); assert!(text.contains("socket.nagle.disable: \"true\"")); assert!(text.contains("statistics.interval.ms: \"1000\"")); @@ -1374,7 +1407,10 @@ mod tests { assert!(text.contains("socket_timeout_ms: 60000")); // Vector batch assert!(text.contains("max_events: 10000")); - assert!(text.contains("timeout_secs: 1")); + // The production profile's `linger.ms` is the batching delay, in + // Vector's field. + assert!(text.contains("timeout_secs: 0.1")); + assert!(!text.contains("linger.ms")); } #[test] @@ -1430,7 +1466,65 @@ mod tests { assert!(text.contains("max_events: 5000")); assert!(text.contains("max_bytes: 8388608")); - assert!(text.contains("timeout_secs: 2")); + // The production profile's `linger.ms` replaces `timeout_secs` on the bus. + assert!(text.contains("timeout_secs: 0.1")); + } + + /// A Kafka sink states its batching delay once, as `batch.timeout_secs`. + /// Vector writes that as `queue.buffering.max.ms`, which librdkafka reads + /// as `linger.ms`, so a second statement leaves the producer with whichever + /// one the client applied last. + #[test] + fn a_kafka_sink_states_its_batching_delay_once() { + let with_option = |k: &str, v: &str| SinkConfig { + librdkafka_options: [(k.to_string(), v.to_string())].into(), + ..Default::default() + }; + + for (case, sink, seconds) in [ + ("profile only", SinkConfig::default(), 0.1), + ("user linger.ms", with_option("linger.ms", "20"), 0.02), + ( + "user queue.buffering.max.ms", + with_option("queue.buffering.max.ms", "5"), + 0.005, + ), + ] { + let component = kafka_sink_component(&sink, &["src".into()], true); + let options = component["librdkafka_options"].as_object().unwrap(); + assert!(!options.contains_key("linger.ms"), "{case}"); + assert!(!options.contains_key("queue.buffering.max.ms"), "{case}"); + assert_eq!(component["batch"]["timeout_secs"], seconds, "{case}"); + } + } + + /// The delay leaves the librdkafka options for Vector's own field, and a + /// value that is not a number of milliseconds stays where librdkafka + /// refuses it. + #[test] + fn the_batching_delay_moves_into_the_batch_timeout() { + let options = |pairs: &[(&str, &str)]| -> Mapping { + pairs.iter().map(|(k, v)| (k.to_string(), val(v))).collect() + }; + + for (case, pairs, seconds, left) in [ + ("none", &[("fetch.min.bytes", "1")][..], None, 1), + ("linger", &[("linger.ms", "100")][..], Some(0.1), 0), + ( + "alias", + &[("queue.buffering.max.ms", "1000")][..], + Some(1.0), + 0, + ), + ("zero", &[("linger.ms", "0")][..], Some(0.0), 0), + ("not a number", &[("linger.ms", "soon")][..], None, 1), + ("negative", &[("linger.ms", "-5")][..], None, 1), + ] { + let mut rdkafka = options(pairs); + let delay = take_batching_delay(&mut rdkafka).map(|n| n.as_f64().unwrap()); + assert_eq!(delay, seconds, "{case}"); + assert_eq!(rdkafka.len(), left, "{case}"); + } } /// The stack's producer compression is zstd, so a sink that names none diff --git a/src/config/kafka_defaults.rs b/src/config/kafka_defaults.rs index 2fd37db..eab8526 100644 --- a/src/config/kafka_defaults.rs +++ b/src/config/kafka_defaults.rs @@ -180,11 +180,39 @@ pub fn producer_profile(profile_name: &str) -> HashMap { } } +/// Settings librdkafka accepts under two spellings and stores as one. A client +/// built from a map applies its entries in hash order, so a map holding both +/// spellings ends up with whichever was applied last. +const ALIASES: &[(&str, &str)] = &[("linger.ms", "queue.buffering.max.ms")]; + +/// The other spelling of `key`, when librdkafka has one. +fn alias_of(key: &str) -> Option<&'static str> { + ALIASES.iter().find_map(|(a, b)| { + if key == *a { + Some(*b) + } else if key == *b { + Some(*a) + } else { + None + } + }) +} + +/// Write `key`, dropping the other spelling of the same setting. +fn insert_unaliased(map: &mut HashMap, key: &str, value: &str) { + if let Some(twin) = alias_of(key) { + map.remove(twin); + } + map.insert(key.to_string(), value.to_string()); +} + /// Merge a base profile with user overrides. /// /// Applied in order, so the last writer wins: the central config (or its scalo /// fallback) baseline, then service-specific overrides, then the user's -/// `librdkafka_options` from config YAML. +/// `librdkafka_options` from config YAML. A layer that names a setting under +/// one spelling replaces the other spelling from every layer below it, and +/// within one layer the spelling that sorts last wins. #[must_use] #[allow( clippy::implicit_hasher, @@ -195,14 +223,22 @@ pub fn merge_layers( service_overrides: &[(&str, &str)], user_overrides: &HashMap, ) -> HashMap { - let mut merged = base.clone(); + let mut merged = HashMap::with_capacity(base.len()); + + let mut base_entries: Vec<(&String, &String)> = base.iter().collect(); + base_entries.sort(); + for (k, v) in base_entries { + insert_unaliased(&mut merged, k, v); + } for (k, v) in service_overrides { - merged.insert((*k).to_string(), (*v).to_string()); + insert_unaliased(&mut merged, k, v); } - for (k, v) in user_overrides { - merged.insert(k.clone(), v.clone()); + let mut user_entries: Vec<(&String, &String)> = user_overrides.iter().collect(); + user_entries.sort(); + for (k, v) in user_entries { + insert_unaliased(&mut merged, k, v); } merged @@ -275,6 +311,68 @@ mod tests { assert_eq!(merged["queue.buffering.max.kbytes"], "524288"); } + /// The setting under its other spelling never survives a higher layer, so + /// the merged map names it exactly once. + #[test] + fn merge_layers_keeps_one_spelling_of_an_aliased_setting() { + type Pairs = &'static [(&'static str, &'static str)]; + let map = |pairs: &[(&str, &str)]| -> HashMap { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect() + }; + let cases: [(Pairs, Pairs, Pairs, &str, &str); 6] = [ + // (base, service, user, surviving key, surviving value) + (&[("linger.ms", "100")], &[], &[], "linger.ms", "100"), + ( + &[("linger.ms", "100")], + &[], + &[("queue.buffering.max.ms", "5")], + "queue.buffering.max.ms", + "5", + ), + ( + &[("queue.buffering.max.ms", "5")], + &[], + &[("linger.ms", "20")], + "linger.ms", + "20", + ), + ( + &[("linger.ms", "100")], + &[("queue.buffering.max.ms", "7")], + &[], + "queue.buffering.max.ms", + "7", + ), + ( + &[("linger.ms", "100"), ("queue.buffering.max.ms", "5")], + &[], + &[], + "queue.buffering.max.ms", + "5", + ), + ( + &[], + &[], + &[("linger.ms", "20"), ("queue.buffering.max.ms", "5")], + "queue.buffering.max.ms", + "5", + ), + ]; + + for (base, service, user, key, value) in cases { + let merged = merge_layers(&map(base), service, &map(user)); + let spellings: Vec<&str> = ["linger.ms", "queue.buffering.max.ms"] + .into_iter() + .filter(|k| merged.contains_key(*k)) + .collect(); + assert_eq!(spellings, vec![key], "base {base:?} user {user:?}"); + assert_eq!(merged[key], value, "base {base:?} user {user:?}"); + } + } + #[test] fn parse_profile_map_handles_types() { let mut mapping = serde_json::Map::new(); diff --git a/src/config/loader.rs b/src/config/loader.rs index 288a6b2..d8f81e0 100644 --- a/src/config/loader.rs +++ b/src/config/loader.rs @@ -506,7 +506,8 @@ pub struct BatchConfig { /// transport at most 16 MiB, and 8 MiB when unset. pub max_bytes: Option, /// Max batch age (seconds) before flush. Default: 1, and at most 1 on the - /// direct transport. + /// direct transport. On the bus transport it applies only where no + /// librdkafka layer sets `linger.ms`, which the production profile does. pub timeout_secs: u32, } diff --git a/src/config/validate.rs b/src/config/validate.rs index 915ceec..8093471 100644 --- a/src/config/validate.rs +++ b/src/config/validate.rs @@ -453,7 +453,7 @@ pub fn warn_if_disk_buffer_is_ephemeral(config_dir: &Path, vector_config: &Vecto (writable layer or emptyDir). Buffered events will be LOST on \ restart or reschedule -- the durability a disk buffer exists to \ provide is absent, and nothing will error when it happens. \ - Mount external storage at that path (Helm: persistence.enabled=true; \ + Mount external storage at that path (Helm: writablePaths.data.persistence.enabled=true; \ docker: -v), or switch to buffer.type=memory so the trade is explicit.", data_dir.display() ); diff --git a/src/deployment.rs b/src/deployment.rs index f4ec420..1c12616 100644 --- a/src/deployment.rs +++ b/src/deployment.rs @@ -313,7 +313,7 @@ const SHA256_ARM64_PLACEHOLDER: &str = "@VECTOR_SHA256_ARM64@"; /// dfe-transform-vector ships TWO binaries in its runtime image — its own /// Rust wrapper (autobuilt by cargo, handled by scalo's contract) AND /// the upstream `vector` binary (downloaded inside the build at the -/// version pinned by [`VECTOR_VERSION`]). The other five DFE Rust apps +/// version pinned by [`VECTOR_VERSION`]). The other DFE Rust apps /// are single-binary images; this consumer-side override exists because /// scalo's deployment contract has no slot for an add-on native binary. /// diff --git a/src/metrics.rs b/src/metrics.rs index 3164dfe..3b67feb 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -126,8 +126,8 @@ impl WrapperMetrics { "config_validation_errors_total", "Total config validation errors", ); - // App-segment name: the manager's namespace prepends the platform - // prefix, giving `dfe_transform_vector_scrape_failures_total`. + // Registered as written, `transform_vector_scrape_failures_total`: names + // are bare unless `metrics.namespace` is set. let scrape_failures_total = manager.counter( "transform_vector_scrape_failures_total", "Failed scrapes of Vector's prometheus_exporter", @@ -140,7 +140,7 @@ impl WrapperMetrics { // Describe the labelled metrics (recorded via macros in // set_lifecycle_state / record_config_reload). Names are BARE -- - // the manager's namespace adds the app prefix once. + // the manager's `metrics.namespace`, when set, prefixes them. metrics::describe_gauge!( "transform_vector_lifecycle_state", "Current lifecycle state (1=active)"