From ddaf967e34c796c93011ef603a993c5252518488 Mon Sep 17 00:00:00 2001 From: lunarthegrey Date: Sat, 10 Oct 2026 22:04:55 -0500 Subject: [PATCH 1/2] feat(flow-export): path-mtu sizes datagrams for a tunnel to the collectors Datagrams were a fixed 1,400 bytes: right for a 1500-byte path, too big for the tunnel the production collectors will most likely sit behind. Tailscale's MTU is 1280, so every full datagram would leave fragmented, and losing either fragment loses the datagram: whole batches of samples, or IPFIX records the cache has already let go. `path-mtu ` makes each datagram that MTU less 48 (IPv6 and UDP headers, so either family fits). It is one value for the section rather than per collector: the sFlow exporter encodes each datagram once for every sFlow collector under one agent sequence, and IPFIX once per privacy profile, and the collectors one router reaches share a path. Unset, nothing changes. 576 is the floor because it still holds a sample of the most header-bytes allows in sFlow's expanded encoding with an IPv6 agent, and the IPFIX announcement with its records. It is hot: each tick encodes with the configuration in place. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + conf/example.conf | 6 +- crates/common/src/config.rs | 50 ++++++++++++++- crates/modules/flow-export/src/cfg.rs | 15 ++++- crates/modules/flow-export/src/ipfix_out.rs | 61 +++++++++++++++--- crates/modules/flow-export/src/sflow_out.rs | 68 ++++++++++++++------- crates/modules/flow-export/src/worker.rs | 39 +++++++++++- docs/runbooks/flow-export.md | 18 ++++++ 8 files changed, 222 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 66f7a3b4..3fd9f988 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,6 +54,7 @@ shipped; that work is part of 0.5.0. - A reload that changes `sample-rate` or `header-bytes` succeeds only once fast-path's sampler runs at the new values. One that cannot be applied changes nothing, fails the reload, and leaves the `sampling` row Degraded with the reason. VPP's plugin takes the new values from the next `desired.conf`, and the `vpp` row reads Degraded until it has. - Status shows a row per path (`xdp`, `tc`, `vpp`) with each port's coverage (starting, covered, degraded on loss, uncovered on heavy loss or a busy port that yields no samples). A VPP port is also uncovered while the plugin is unavailable or its interface is missing from VPP, and the `vpp` row leads with the plugin's own state: its epoch, the generation it applied of the one asked, and why it is not healthy. There is also a `sampling` row (over budget below 1:1000), and a row per collector, which reports local submission only: receipt is the collector's to confirm. A collector reads Degraded while its sends fail or while datagrams exceed its per-tick send budget. Metrics are under `packetframe_flow_export_`. `worker_up` reads 0 once the export worker stalls or panics, and the coverage gauges then read 0 too. - **VPP.** With `module vpp-offload` configured, flow-export owns the plugin's `desired.conf` while it runs, asking for every VPP port. `packetframe sampler configure` and `clear` are refused meanwhile, as is a second `sampler watch`. A VPP interface index means something only to the VPP process that assigned it, so a sample is read only through the port list of the process that wrote it. A sample from a restarted VPP that its new port list does not yet cover is counted as unmapped, never attributed to a port by guess. Like samples lost to full rings, unmapped samples degrade VPP's coverage and ride the next VPP sample's `drops`. A VPP port reads `starting` for 60 s rather than 30 s, since its interface appears only once VPP is up. A `desired.conf` that a daemon which died left behind is flow-export's to take over: vpp-offload's attach leaves it, so VPP keeps sampling until flow-export writes its own, and removes it only when flow-export is not configured. Generations keep rising across a restart that keeps VPP. A stop removes `desired.conf`, and the plugin then reports generation 0, so flow-export records the newest it wrote for each VPP process in `/flow-export-vpp-generation.json`. Without the record the next start wrote generation 1 again, and a sample the previous run left queued under a reused number was read through the new run's port list rather than counted unmapped. + - **`path-mtu `** (hot, 576–9000) fits every datagram to the path to the collectors: each is that MTU less 48 bytes for the IPv6 and UDP headers. Unset, datagrams are 1,400 bytes, inside any 1500-byte path. Set it when the collectors sit across a tunnel (Tailscale's MTU is 1280), where larger datagrams leave fragmented and one lost fragment loses the whole datagram. - Requires `module fast-path` listed before it, and `module vpp-offload` too when present. A failure to start degrades the module and never stops forwarding. A panic in the export worker stops both samplers and says why in the `worker` row. Stopping the module (including `systemctl stop`) stops both samplers, and so does `packetframe detach --all`. - `packetframe sampler status` and `watch` judge VPP coverage Degraded while samples are lost to full rings (`status` over a quarter second, `watch` over each report), or while more than half the ring slots are queued unread. - **IPFIX.** `collector ipfix :` sends flow records (RFC 7011) instead of samples. diff --git a/conf/example.conf b/conf/example.conf index cadce3a7..88685875 100644 --- a/conf/example.conf +++ b/conf/example.conf @@ -1199,7 +1199,7 @@ module fast-path # rate is accepted (down to 1:100) and reported over budget. # # RESTART-ONLY: `source-address`, `kernel-sample`. HOT (SIGHUP): `sample-rate`, -# `header-bytes`, `collector`, `flow-cache`, `privacy-local-prefix`. +# `header-bytes`, `path-mtu`, `collector`, `flow-cache`, `privacy-local-prefix`. # # module flow-export # source-address 192.0.2.1 # required: the sFlow agent address and @@ -1207,6 +1207,10 @@ module fast-path # # be an address of this host # sample-rate 1000 # 1 in N packets (100..16777216) # header-bytes 128 # leading packet bytes per sample (64..256) +# path-mtu 1280 # the path to the collectors (576..9000): +# # datagrams are this less 48, so a tunnel +# # (Tailscale: 1280) fragments none. Unset: +# # 1400-byte datagrams, inside any 1500 path # collector fnm sflow 198.51.100.10:6343 kind ddos # collector akv ipfix 198.51.100.11:4739 kind stats profile truncate # # name, format (sflow or ipfix), : diff --git a/crates/common/src/config.rs b/crates/common/src/config.rs index 91c20a23..4c7e507d 100644 --- a/crates/common/src/config.rs +++ b/crates/common/src/config.rs @@ -860,6 +860,15 @@ pub enum ModuleDirective { bytes: u32, line: usize, }, + /// `path-mtu ` — the MTU of the path to the collectors, within + /// [`FLOW_PATH_MTU_RANGE`]: every datagram fits it with room for IPv6 + /// and UDP headers ([`FLOW_DATAGRAM_OVERHEAD`]), so a tunnel's smaller + /// MTU (Tailscale's is 1280) fragments none. Unset, datagrams are + /// [`FLOW_DEFAULT_DATAGRAM`] bytes, inside any 1500-byte path. Hot. + FlowPathMtu { + mtu: u32, + line: usize, + }, /// `collector sflow|ipfix : [kind stats|ddos]` — a /// collector, the format it takes, and what it is for. `kind ddos` /// takes sFlow only. Hot. @@ -2028,6 +2037,15 @@ pub const FLOW_DEFAULT_HEADER_BYTES: u32 = 128; /// more than the VPP sampler's slot carries. pub const FLOW_HEADER_BYTES_RANGE: (u32, u32) = (64, 256); pub const FLOW_MAX_COLLECTORS: usize = 8; +/// A datagram's bytes with no `path-mtu`: inside a 1500-byte path with +/// room for the IP and UDP headers. +pub const FLOW_DEFAULT_DATAGRAM: usize = 1400; +/// What a datagram leaves of `path-mtu` for the IP and UDP headers: IPv6's +/// and UDP's, so either family fits. +pub const FLOW_DATAGRAM_OVERHEAD: u32 = 48; +/// `path-mtu` bounds: IPv4's minimum, which still holds a sample of the +/// most `header-bytes` allows; and a jumbo frame. +pub const FLOW_PATH_MTU_RANGE: (u32, u32) = (576, 9000); /// The IPFIX flow cache: flows per observation domain, and its bounds. pub const FLOW_CACHE_ENTRIES_DEFAULT: usize = 65_536; pub const FLOW_CACHE_ENTRIES_RANGE: (usize, usize) = (1024, 1 << 20); @@ -3383,7 +3401,8 @@ impl Config { message: format!("module flow-export: {msg}"), }; let mut source: Option<(IpAddr, usize)> = None; - let (mut rate_line, mut header_line, mut cache_line) = (None, None, None); + let (mut rate_line, mut header_line, mut cache_line, mut mtu_line) = + (None, None, None, None); let mut kernel: Vec<(String, usize)> = Vec::new(); let mut collectors: Vec<(&str, std::net::SocketAddr, usize)> = Vec::new(); for d in &self.modules[pos].directives { @@ -3413,6 +3432,14 @@ impl Config { )); } } + ModuleDirective::FlowPathMtu { line, .. } => { + if let Some(prev) = mtu_line.replace(*line) { + return Err(err( + *line, + format!("`path-mtu` given twice (first on line {prev})"), + )); + } + } ModuleDirective::FlowKernelSample { iface, line } => { if let Some((_, prev)) = kernel.iter().find(|(i, _)| i == iface) { return Err(err( @@ -4938,6 +4965,14 @@ fn parse_module_directive(line: usize, s: &str) -> Result parse_single_arg(line, rest, "path-mtu", |t| { + let mtu: u32 = t.parse().map_err(|e| format!("bad integer `{t}`: {e}"))?; + let (lo, hi) = FLOW_PATH_MTU_RANGE; + if !(lo..=hi).contains(&mtu) { + return Err(format!("must be between {lo} and {hi}")); + } + Ok(ModuleDirective::FlowPathMtu { mtu, line }) + }), "collector" => parse_flow_collector(line, rest), "flow-cache" => parse_flow_cache(line, rest), "kernel-sample" => parse_single_arg(line, rest, "kernel-sample", |t| { @@ -10459,6 +10494,12 @@ module fast-path assert!(fe .directives .contains(&ModuleDirective::FlowHeaderBytes { bytes: 96, line: 6 })); + let tunnel = format!("{FLOW_OK} path-mtu 1280\n"); + let c = Config::parse(&tunnel).unwrap(); + c.validate_flow_export().unwrap(); + assert!(c.modules[1] + .directives + .contains(&ModuleDirective::FlowPathMtu { mtu: 1280, line: 9 })); assert!(fe.directives.contains(&ModuleDirective::FlowCollector { name: "fnm".into(), format: CollectorFormat::Sflow, @@ -10488,6 +10529,9 @@ module fast-path ("sample-rate 16777217", "between 100 and"), ("header-bytes 63", "between 64 and 256"), ("header-bytes 257", "between 64 and 256"), + ("path-mtu 575", "between 576 and 9000"), + ("path-mtu 9001", "between 576 and 9000"), + ("path-mtu jumbo", "bad integer"), ("source-address 0.0.0.0", "cannot be an exporter"), ("source-address host", "not an IP address"), ("collector c netflow 198.51.100.1:2055", "unknown format"), @@ -10601,6 +10645,10 @@ module fast-path ); refuse(" flow-cache active 0\n", "active must be"); refuse(" flow-cache entries 2048 entries 4096\n", "given twice"); + refuse( + " path-mtu 1280\n path-mtu 1500\n collector a sflow 198.51.100.1:6343\n", + "`path-mtu` given twice", + ); refuse( " flow-cache active 60\n flow-cache inactive 15\n collector a ipfix 198.51.100.1:4739\n", "`flow-cache` given twice", diff --git a/crates/modules/flow-export/src/cfg.rs b/crates/modules/flow-export/src/cfg.rs index 5c13b5a3..964c194e 100644 --- a/crates/modules/flow-export/src/cfg.rs +++ b/crates/modules/flow-export/src/cfg.rs @@ -6,7 +6,7 @@ use std::net::{IpAddr, SocketAddr}; use packetframe_common::config::{ CollectorFormat, CollectorKind, CollectorProfile, ModuleDirective, FLOW_BUDGET_RATE, - FLOW_DEFAULT_HEADER_BYTES, FLOW_DEFAULT_RATE, + FLOW_DATAGRAM_OVERHEAD, FLOW_DEFAULT_DATAGRAM, FLOW_DEFAULT_HEADER_BYTES, FLOW_DEFAULT_RATE, }; use packetframe_common::fib::IpPrefix; @@ -30,6 +30,9 @@ pub struct FlowExportConfig { pub rate: u32, pub header_bytes: u32, pub collectors: Vec, + /// The most bytes a datagram holds: what `path-mtu` leaves for them, + /// else [`FLOW_DEFAULT_DATAGRAM`]. + pub datagram: usize, /// The IPFIX flow cache's bounds. pub cache: Limits, /// `privacy-local-prefix`: empty means `local_default`. @@ -46,6 +49,7 @@ impl FlowExportConfig { let mut source = None; let mut rate = FLOW_DEFAULT_RATE; let mut header_bytes = FLOW_DEFAULT_HEADER_BYTES; + let mut datagram = FLOW_DEFAULT_DATAGRAM; let mut collectors = Vec::new(); let mut cache = Limits::default(); let mut local = Vec::new(); @@ -55,6 +59,9 @@ impl FlowExportConfig { ModuleDirective::FlowSourceAddress { addr, .. } => source = Some(*addr), ModuleDirective::FlowSampleRate { rate: r, .. } => rate = *r, ModuleDirective::FlowHeaderBytes { bytes, .. } => header_bytes = *bytes, + ModuleDirective::FlowPathMtu { mtu, .. } => { + datagram = (*mtu - FLOW_DATAGRAM_OVERHEAD) as usize + } ModuleDirective::FlowCollector { name, addr, @@ -95,6 +102,7 @@ impl FlowExportConfig { rate, header_bytes, collectors, + datagram, cache, local, kernel, @@ -175,6 +183,11 @@ mod tests { assert_eq!((c.rate, c.header_bytes), (100, 64)); assert!(c.over_budget()); assert_eq!(c.collectors[0].kind, CollectorKind::Ddos); + assert_eq!(c.datagram, FLOW_DEFAULT_DATAGRAM, "no path-mtu"); + let c = parse( + " source-address 192.0.2.1\n path-mtu 1280\n collector a sflow 198.51.100.1:6343\n", + ); + assert_eq!(c.datagram, 1232, "Tailscale's MTU, less IPv6 and UDP"); } #[test] diff --git a/crates/modules/flow-export/src/ipfix_out.rs b/crates/modules/flow-export/src/ipfix_out.rs index 716a5890..5a42e7a8 100644 --- a/crates/modules/flow-export/src/ipfix_out.rs +++ b/crates/modules/flow-export/src/ipfix_out.rs @@ -28,7 +28,6 @@ use packetframe_flow_encode::{ }; use crate::flows::{Domain, Export}; -use crate::sflow_out::MAX_DATAGRAM; pub const SELECTOR_TEMPLATE: u16 = 258; /// Templates and the selector, again at least this often while records @@ -182,7 +181,9 @@ impl IpfixOut { } } - /// Messages for one domain's exports, in order, appended to `out`. + /// Messages of at most `max` bytes (the module's `datagram`) for one + /// domain's exports, in order, appended to `out`. + #[allow(clippy::too_many_arguments)] pub fn encode( &mut self, domain: Domain, @@ -190,6 +191,7 @@ impl IpfixOut { privacy: &Privacy<'_>, now: Instant, export_s: u32, + max: usize, out: &mut Vec>, ) { self.domains.entry(domain).or_insert_with(|| DomainOut { @@ -213,7 +215,7 @@ impl IpfixOut { }; let d = self.domains.get_mut(&domain).expect("inserted above"); if rate != d.rate { - self.flush(domain, &mut pending, now, export_s, out); + self.flush(domain, &mut pending, now, export_s, max, out); let d = self.domains.get_mut(&domain).expect("inserted above"); d.rate = rate; d.next_announce = None; @@ -223,7 +225,7 @@ impl IpfixOut { pending.entry(template).or_default().push(r); } } - self.flush(domain, &mut pending, now, export_s, out); + self.flush(domain, &mut pending, now, export_s, max, out); } fn flush( @@ -232,6 +234,7 @@ impl IpfixOut { pending: &mut BTreeMap>, now: Instant, export_s: u32, + max: usize, out: &mut Vec>, ) { if pending.values().all(Vec::is_empty) { @@ -265,10 +268,10 @@ impl IpfixOut { } else { Announce::default() }; - let mut buf = Vec::with_capacity(MAX_DATAGRAM); + let mut buf = Vec::with_capacity(max); match d .exporter - .encode_flows(&mut buf, export_s, &a, template, rest, MAX_DATAGRAM) + .encode_flows(&mut buf, export_s, &a, template, rest, max) { Ok(n) => { if announce { @@ -283,9 +286,9 @@ impl IpfixOut { // The announcement and a record do not fit together: // the announcement goes alone, then the records. Err(EncodeError::TooLarge) if announce => { - let mut alone = Vec::with_capacity(MAX_DATAGRAM); + let mut alone = Vec::with_capacity(max); if d.exporter - .encode_flows(&mut alone, export_s, &a, template, &[], MAX_DATAGRAM) + .encode_flows(&mut alone, export_s, &a, template, &[], max) .is_ok() { out.push(alone); @@ -311,6 +314,9 @@ impl IpfixOut { mod tests { use super::*; use crate::flows::{FlowCache, Limits, Now, Packet, Sampled}; + use packetframe_common::config::{ + FLOW_DATAGRAM_OVERHEAD, FLOW_DEFAULT_DATAGRAM, FLOW_PATH_MTU_RANGE, + }; const TEMPLATE_V4: u16 = 256; const TEMPLATE_V6: u16 = 257; @@ -383,11 +389,22 @@ mod tests { x: &mut IpfixOut, privacy: &Privacy<'_>, now: Instant, + ) -> Vec> { + encode_all_within(cache, x, privacy, now, FLOW_DEFAULT_DATAGRAM) + } + + /// [`encode_all`] into messages of at most `max` bytes. + fn encode_all_within( + cache: &mut FlowCache, + x: &mut IpfixOut, + privacy: &Privacy<'_>, + now: Instant, + max: usize, ) -> Vec> { let mut out = Vec::new(); for d in cache.pending() { let items = cache.take(d, RECORDS_PER_TICK); - x.encode(d, &items, privacy, now, 0, &mut out); + x.encode(d, &items, privacy, now, 0, max, &mut out); } out } @@ -445,6 +462,7 @@ mod tests { &NONE, Instant::now(), 0, + FLOW_DEFAULT_DATAGRAM, &mut out, ); assert_eq!(sets(&out[0]).1, Some(1000)); @@ -513,10 +531,33 @@ mod tests { cache.expire(at(15_000), 1000); let out = encode_all(&mut cache, &mut x, &NONE, Instant::now()); assert!(out.len() > 1); - assert!(out.iter().all(|d| d.len() <= MAX_DATAGRAM)); + assert!(out.iter().all(|d| d.len() <= FLOW_DEFAULT_DATAGRAM)); assert_eq!(x.records, 500); } + /// `path-mtu`'s smallest: the announcement and every record still go, + /// in more messages, none larger. + #[test] + fn the_smallest_path_mtu_still_carries_every_record() { + let max = (FLOW_PATH_MTU_RANGE.0 - FLOW_DATAGRAM_OVERHEAD) as usize; + let batch: Vec = (0..500) + .map(|p| sampled("192.0.2.1", "198.51.100.2", p, 1000, 1)) + .collect(); + let mut counts = Vec::new(); + for limit in [FLOW_DEFAULT_DATAGRAM, max] { + let mut cache = FlowCache::new(Limits::default()); + let mut x = IpfixOut::new(CollectorProfile::Full); + ingest(&mut cache, Domain::FastPath, &batch, at(0)); + cache.expire(at(15_000), 1000); + let out = encode_all_within(&mut cache, &mut x, &NONE, Instant::now(), limit); + assert!(out.iter().all(|d| d.len() <= limit), "{limit}"); + assert_eq!(sets(&out[0]).1, Some(1000), "announced first"); + assert_eq!((x.records, x.unencodable), (500, 0), "{limit}"); + counts.push(out.len()); + } + assert!(counts[1] > counts[0], "{counts:?}"); + } + fn record(src: &str, dst: &str) -> FlowRecord { FlowRecord { src: src.parse().unwrap(), diff --git a/crates/modules/flow-export/src/sflow_out.rs b/crates/modules/flow-export/src/sflow_out.rs index 3563c1a5..35d4dc3f 100644 --- a/crates/modules/flow-export/src/sflow_out.rs +++ b/crates/modules/flow-export/src/sflow_out.rs @@ -7,10 +7,6 @@ use packetframe_fast_path::sample::Sample; use packetframe_flow_encode::sflow::{self, Agent, Encoding, FlowSample, COMPACT_IF_LIMIT}; use packetframe_flow_encode::EncodeError; -/// Datagram size: inside a 1500-byte path with room for the IP and UDP -/// headers, so no collector ever sees a fragment. -pub const MAX_DATAGRAM: usize = 1400; - /// Octets a frame's FCS adds on the wire: sFlow's `frame_length` counts /// them and `stripped` says they are not in the header. pub const FCS: u32 = 4; @@ -102,11 +98,18 @@ impl Exporter { self.agent.encoding == Encoding::Expanded } - /// Encode `ready` into datagrams of at most [`MAX_DATAGRAM`] bytes, - /// appended to `out`. The first ifIndex past the compact encoding's - /// 24 bits moves the agent to the expanded one for good. Returns the - /// samples no datagram could hold (none, at these sizes). - pub fn encode(&mut self, ready: &[Ready], uptime_ms: u32, out: &mut Vec>) -> usize { + /// Encode `ready` into datagrams of at most `max` bytes (the module's + /// `datagram`), appended to `out`. The first ifIndex past the compact + /// encoding's 24 bits moves the agent to the expanded one for good. + /// Returns the samples no datagram could hold (none at the sizes + /// `path-mtu` allows). + pub fn encode( + &mut self, + ready: &[Ready], + uptime_ms: u32, + max: usize, + out: &mut Vec>, + ) -> usize { if self.agent.encoding == Encoding::Compact && ready .iter() @@ -133,15 +136,9 @@ impl Exporter { let mut rest = &samples[..]; let mut unsent = 0; while !rest.is_empty() { - let mut buf = Vec::with_capacity(MAX_DATAGRAM); - match sflow::encode_datagram( - &mut buf, - &self.agent, - self.sequence, - uptime_ms, - rest, - MAX_DATAGRAM, - ) { + let mut buf = Vec::with_capacity(max); + match sflow::encode_datagram(&mut buf, &self.agent, self.sequence, uptime_ms, rest, max) + { Ok(n) => { self.sequence = self.sequence.wrapping_add(1); out.push(buf); @@ -165,6 +162,9 @@ impl Exporter { #[cfg(test)] mod tests { use super::*; + use packetframe_common::config::{ + FLOW_DATAGRAM_OVERHEAD, FLOW_DEFAULT_DATAGRAM, FLOW_HEADER_BYTES_RANGE, FLOW_PATH_MTU_RANGE, + }; use packetframe_fast_path::sample::{Disposition, Path, Vlan}; fn sample<'a>(header: &'a [u8], vlan: Option) -> Sample<'a> { @@ -244,9 +244,9 @@ mod tests { let mut x = Exporter::new("192.0.2.1".parse().unwrap()); let batch: Vec = (0..40).map(|i| ready(i, 3)).collect(); let mut out = Vec::new(); - assert_eq!(x.encode(&batch, 5, &mut out), 0); + assert_eq!(x.encode(&batch, 5, FLOW_DEFAULT_DATAGRAM, &mut out), 0); assert!(out.len() > 1); - assert!(out.iter().all(|d| d.len() <= MAX_DATAGRAM)); + assert!(out.iter().all(|d| d.len() <= FLOW_DEFAULT_DATAGRAM)); // Each datagram's sequence (word 4 after the v4 agent address). let seqs: Vec = out .iter() @@ -261,13 +261,37 @@ mod tests { assert_eq!(n, 40); } + /// `path-mtu`'s smallest still holds a sample of the most + /// `header-bytes` allows, in the expanded encoding with an IPv6 agent: + /// every sample goes, one datagram each at worst. + #[test] + fn the_smallest_path_mtu_holds_the_largest_sample() { + let max = (FLOW_PATH_MTU_RANGE.0 - FLOW_DATAGRAM_OVERHEAD) as usize; + let mut x = Exporter::new("2001:db8::1".parse().unwrap()); + let big = |i| Ready { + header: vec![0xab; FLOW_HEADER_BYTES_RANGE.1 as usize], + ..ready(i, COMPACT_IF_LIMIT) + }; + let batch: Vec = (0..10).map(big).collect(); + let mut out = Vec::new(); + assert_eq!(x.encode(&batch, 5, max, &mut out), 0, "none unencodable"); + assert!(x.expanded()); + assert!(out.iter().all(|d| d.len() <= max)); + assert!(out.len() >= 5, "{} datagrams", out.len()); + } + #[test] fn a_wide_ifindex_moves_the_agent_to_the_expanded_encoding() { let mut x = Exporter::new("192.0.2.1".parse().unwrap()); let mut out = Vec::new(); - x.encode(&[ready(0, 3)], 0, &mut out); + x.encode(&[ready(0, 3)], 0, FLOW_DEFAULT_DATAGRAM, &mut out); assert!(!x.expanded()); - x.encode(&[ready(1, COMPACT_IF_LIMIT)], 0, &mut out); + x.encode( + &[ready(1, COMPACT_IF_LIMIT)], + 0, + FLOW_DEFAULT_DATAGRAM, + &mut out, + ); assert!(x.expanded()); assert_eq!(out.len(), 2); } diff --git a/crates/modules/flow-export/src/worker.rs b/crates/modules/flow-export/src/worker.rs index 6d92b947..8d7369a7 100644 --- a/crates/modules/flow-export/src/worker.rs +++ b/crates/modules/flow-export/src/worker.rs @@ -545,7 +545,9 @@ impl Worker { let mut datagrams = Vec::new(); let uptime_ms = now.saturating_duration_since(self.shared.epoch).as_millis() as u32; self.p.unencodable_total += - self.exporter.encode(&ready, uptime_ms, &mut datagrams) as u64; + self.exporter + .encode(&ready, uptime_ms, self.cfg.datagram, &mut datagrams) + as u64; self.p.datagrams_total += datagrams.len() as u64; send_all( &self.transport, @@ -652,6 +654,7 @@ impl Worker { &privacy, now, (at.wall_ms / 1000) as u32, + self.cfg.datagram, &mut messages, ); self.p.ipfix_messages_total += messages.len() as u64; @@ -1315,6 +1318,7 @@ mod tests { format: CollectorFormat::Sflow, profile: Default::default(), }], + datagram: packetframe_common::config::FLOW_DEFAULT_DATAGRAM, cache: flows::Limits::default(), local: Vec::new(), kernel: Vec::new(), @@ -1387,6 +1391,39 @@ mod tests { u32::from_be_bytes(d[at..at + 4].try_into().unwrap()) } + /// A reload to a smaller `path-mtu` applies from the next tick: the + /// same samples go in more, smaller datagrams. + #[test] + fn a_reloaded_path_mtu_sizes_the_next_datagrams() { + let mut r = rig(1000); + r.w.tick(r.t0); + r.ports.rx.borrow_mut().insert(3, 6_000); + let mut t = r.t0 + PORTS_EVERY; + r.w.tick(t); + let mut sizes = Vec::new(); + for datagram in [packetframe_common::config::FLOW_DEFAULT_DATAGRAM, 528] { + let c = FlowExportConfig { + datagram, + ..cfg(1000) + }; + let answer = reload(&r.shared, c); + r.src + .events + .borrow_mut() + .extend((0..20).map(|_| event(3, 0, 0, 1000, 1))); + r.net.sent.borrow_mut().clear(); + t += TICK; + r.w.tick(t); + assert_eq!(answer.try_recv().unwrap(), Ok(())); + let sent = r.net.sent.borrow(); + assert!(sent.iter().all(|(_, d)| d.len() <= datagram), "{datagram}"); + let samples: u32 = sent.iter().map(|(_, d)| word(d, 24)).sum(); + assert_eq!(samples, 20, "{datagram}"); + sizes.push(sent.len()); + } + assert!(sizes[1] > sizes[0], "{sizes:?}"); + } + #[test] fn samples_reach_every_collector_as_sflow_with_their_pool_and_sequence() { let mut r = rig(1000); diff --git a/docs/runbooks/flow-export.md b/docs/runbooks/flow-export.md index ab428acb..ee1647d7 100644 --- a/docs/runbooks/flow-export.md +++ b/docs/runbooks/flow-export.md @@ -51,6 +51,7 @@ module flow-export source-address 192.0.2.1 # RESTART-ONLY; must be this host's sample-rate 1000 # hot; 100..16777216, default 1000 header-bytes 128 # hot; 64..256 + path-mtu 1280 # hot; 576..9000; unset: 1400-byte datagrams # collectors: hot; at most 8, each in source-address's family collector fnm sflow 198.51.100.10:6343 kind ddos collector akv ipfix 198.51.100.11:4739 kind stats profile truncate @@ -73,6 +74,23 @@ host: exporter on. Keep it stable: changing it is a restart, and to the collector it is a new device. +**Over a tunnel, set `path-mtu`.** Datagrams are 1,400 bytes by default, +which fits any 1500-byte path. A tunnel's MTU is smaller (Tailscale's is +1280, WireGuard's usually 1420), so they would leave fragmented, and one +lost fragment loses the whole datagram. `path-mtu ` makes every +datagram that MTU less 48 (IPv6 and UDP headers), so `path-mtu 1280` +sends 1,232-byte datagrams. It is one value for every collector: they +share the datagrams. + +Over Tailscale: +- give the collector's Tailscale address in each `collector` line, and + this router's own as `source-address`; +- the tailnet policy must allow this router to reach the collector's + UDP ports (tailscaled drops what it does not allow, silently); +- a collector in Docker keeps the sender's address only with Docker's + default (iptables) port publishing: a userland proxy shows every + exporter as the bridge's gateway. + ### Akvorado - **sFlow or IPFIX.** Both scale exactly once From a6d5efdafe453636f26d8491c32c4de15c742bb1 Mon Sep 17 00:00:00 2001 From: lunarthegrey Date: Sat, 10 Oct 2026 22:14:24 -0500 Subject: [PATCH 2/2] fix(flow-export): take no more IPFIX records a tick than the budget sends A tick took a fixed 8,192 records from the flow cache, sized for 1,400-byte messages: 16 IPv6 records each, 512 messages, exactly the per-collector send budget. With path-mtu 1280 a message holds 14, so a full take needs about 586 messages, and the budget dropped the rest after their records had left the cache. A rate change makes a full take certain: it queues every flow at once. In the test, 20,000 IPv6 flows at 1,232-byte datagrams lost 148 messages. At 1,400 the old take had no room for the announcement either. The take is now derived from the datagram size: what the budget's messages carry of the largest record any template sends (85 bytes, IPv6), keeping 64 messages back for announcements and the part-filled message each domain, template and rate change ends a tick with. At 1,400 that is 7,168 records a tick, still 71,680 a second. Co-Authored-By: Claude Opus 5.5 --- crates/flow-encode/src/ipfix.rs | 11 +++ crates/modules/flow-export/src/ipfix_out.rs | 48 ++++++++++-- crates/modules/flow-export/src/worker.rs | 82 ++++++++++++++++++++- 3 files changed, 133 insertions(+), 8 deletions(-) diff --git a/crates/flow-encode/src/ipfix.rs b/crates/flow-encode/src/ipfix.rs index 3063c95d..8ade93a3 100644 --- a/crates/flow-encode/src/ipfix.rs +++ b/crates/flow-encode/src/ipfix.rs @@ -52,6 +52,17 @@ const DATA_LINK_ETHERNET: u16 = 0x0001; const SET_TEMPLATE: u16 = 2; const SET_OPTIONS_TEMPLATE: u16 = 3; +/// A message's bytes before its first set (RFC 7011 §3.1), and a set's +/// before its records (§3.3.2); a set is padded to four bytes after them. +pub const MESSAGE_HEADER_LEN: usize = 16; +pub const SET_HEADER_LEN: usize = 4; +pub const SET_PADDING_MAX: usize = 3; + +/// The bytes one record of `fields` takes in a data set. +pub fn record_len(fields: &[Field]) -> usize { + fields.iter().map(|&f| usize::from(ie_and_len(f).1)).sum() +} + fn ie_and_len(f: Field) -> (u16, u16) { match f { Field::Octets => (ie::OCTET_DELTA_COUNT, 8), diff --git a/crates/modules/flow-export/src/ipfix_out.rs b/crates/modules/flow-export/src/ipfix_out.rs index 5a42e7a8..d30d1fd7 100644 --- a/crates/modules/flow-export/src/ipfix_out.rs +++ b/crates/modules/flow-export/src/ipfix_out.rs @@ -22,11 +22,15 @@ use std::time::{Duration, Instant}; use packetframe_common::config::CollectorProfile; use packetframe_common::fib::asn::AsnTable; use packetframe_common::fib::IpPrefix; -use packetframe_flow_encode::ipfix::{Announce, Exporter, Selector, Template}; +use packetframe_flow_encode::ipfix::{ + record_len, Announce, Exporter, Selector, Template, MESSAGE_HEADER_LEN, SET_HEADER_LEN, + SET_PADDING_MAX, +}; use packetframe_flow_encode::{ flow_fields, EncodeError, Family, Field, FlowRecord, Profile, SamplingSignal, }; +use crate::collector::SEND_BUDGET; use crate::flows::{Domain, Export}; pub const SELECTOR_TEMPLATE: u16 = 258; @@ -34,9 +38,27 @@ pub const SELECTOR_TEMPLATE: u16 = 258; /// flow: a collector that restarted learns them within it /// (requirement 3). pub const ANNOUNCE_EVERY: Duration = Duration::from_secs(10); -/// Records taken from the cache per tick, across domains: about what the -/// send budget carries. -pub const RECORDS_PER_TICK: usize = 8192; +/// Messages of a collector's per-tick send budget kept back from records: +/// for announcements, and the part-filled message each domain, template +/// and rate change ends a tick with. +const BUDGET_SLACK: usize = 64; + +const _: () = assert!(BUDGET_SLACK < SEND_BUDGET); + +/// Records taken from the cache per tick, across domains, for messages of +/// at most `max` bytes: what the rest of the send budget carries of the +/// largest record any template sends. More, and the budget drops messages +/// whose records have already left the cache, which a rate change (every +/// flow queued at once) makes certain. +pub fn records_per_tick(max: usize) -> usize { + let largest = TEMPLATES + .iter() + .map(|(_, f, p)| record_len(&flow_fields(*f, *p, SamplingSignal::Options))) + .max() + .unwrap_or(1); + let room = max.saturating_sub(MESSAGE_HEADER_LEN + SET_HEADER_LEN + SET_PADDING_MAX); + (room / largest).max(1) * (SEND_BUDGET - BUDGET_SLACK) +} /// Every template a profile can use, and its id. const TEMPLATES: [(u16, Family, Profile); 7] = [ @@ -403,7 +425,7 @@ mod tests { ) -> Vec> { let mut out = Vec::new(); for d in cache.pending() { - let items = cache.take(d, RECORDS_PER_TICK); + let items = cache.take(d, records_per_tick(max)); x.encode(d, &items, privacy, now, 0, max, &mut out); } out @@ -535,6 +557,22 @@ mod tests { assert_eq!(x.records, 500); } + /// The take is what the budget's messages carry of an IPv6 record, the + /// largest (85 bytes), less the slack. + #[test] + fn a_ticks_take_shrinks_with_the_datagram() { + let v6 = record_len(&flow_fields( + Family::V6, + Profile::Full, + SamplingSignal::Options, + )); + assert_eq!(v6, 85); + assert_eq!(records_per_tick(FLOW_DEFAULT_DATAGRAM), 16 * 448); + assert_eq!(records_per_tick(1232), 14 * 448); + let floor = (FLOW_PATH_MTU_RANGE.0 - FLOW_DATAGRAM_OVERHEAD) as usize; + assert_eq!(records_per_tick(floor), 5 * 448); + } + /// `path-mtu`'s smallest: the announcement and every record still go, /// in more messages, none larger. #[test] diff --git a/crates/modules/flow-export/src/worker.rs b/crates/modules/flow-export/src/worker.rs index 8d7369a7..73e190b8 100644 --- a/crates/modules/flow-export/src/worker.rs +++ b/crates/modules/flow-export/src/worker.rs @@ -29,7 +29,7 @@ use crate::cfg::FlowExportConfig; use crate::collector::{reconcile, send_all, send_until, CollectorState, Transport}; use crate::coverage::{PortCoverage, State, Window, WINDOW}; use crate::flows::{self, Domain, FlowCache, Now, Sampled}; -use crate::ipfix_out::{IpfixOut, Privacy, RECORDS_PER_TICK}; +use crate::ipfix_out::{records_per_tick, IpfixOut, Privacy}; use crate::pool::Accumulator; use crate::sflow_out::{ip_frame, wire_frame, Exporter, Ready, FCS}; use crate::vpp::{NoVpp, Taken, VppDir, VppHealth, VppSide}; @@ -606,7 +606,8 @@ impl Worker { cache.ingest(*domain, batch); } cache.expire(at, EXPIRE_BUDGET); - self.send_flows(now, at, RECORDS_PER_TICK, None); + let take = records_per_tick(self.cfg.datagram); + self.send_flows(now, at, take, None); } /// The local prefixes: the section's, else fast-path's allowlist as @@ -702,7 +703,8 @@ impl Worker { .is_some_and(|f| !f.cache.pending().is_empty()) && Instant::now() < deadline { - unsent += self.send_flows(now, at, RECORDS_PER_TICK, Some(deadline)); + let take = records_per_tick(self.cfg.datagram); + unsent += self.send_flows(now, at, take, Some(deadline)); } let unencoded = self.flows.as_ref().map_or(0, |f| f.cache.queued()); if unsent > 0 || unencoded > 0 { @@ -1281,6 +1283,29 @@ mod tests { event_with(ingress, 0, 0, rate, generation, &f) } + /// [`ip_event`] for IPv6: its records are the largest a template + /// carries. + fn ip6_event(ingress: u32, dport: u16, rate: u32, generation: u32) -> Vec { + let mut f = vec![0x02; 12]; + f.extend_from_slice(&[0x86, 0xdd, 0x60, 0, 0, 0, 0x03, 0xc6, 6, 64]); + f.extend_from_slice( + &"2001:db8::7" + .parse::() + .unwrap() + .octets(), + ); + f.extend_from_slice( + &"2001:db8::9" + .parse::() + .unwrap() + .octets(), + ); + f.extend_from_slice(&40000u16.to_be_bytes()); + f.extend_from_slice(&dport.to_be_bytes()); + f.resize(96, 0); + event_with(ingress, 0, 0, rate, generation, &f) + } + fn event_with( ingress: u32, egress: u32, @@ -1933,6 +1958,57 @@ mod tests { assert_eq!(sent.last().unwrap().0.port(), 4739, "{sent:?}"); } + /// A tunnel's `path-mtu` makes messages smaller, so a tick takes fewer + /// records from the cache: what the send budget carries of the largest + /// (IPv6) record. A rate change queues every flow at once; each goes, + /// and the budget drops no message whose records already left the + /// cache. + #[test] + fn a_small_path_mtu_takes_no_more_records_than_the_budget_sends() { + for datagram in [packetframe_common::config::FLOW_DEFAULT_DATAGRAM, 1232] { + a_rate_change_backlog_goes_out_whole(datagram); + } + } + + fn a_rate_change_backlog_goes_out_whole(datagram: usize) { + let mut r = rig(1000); + let c = FlowExportConfig { + datagram, + ..with_ipfix(cfg(1000)) + }; + let _ = reload(&r.shared, c); + r.w.tick(r.t0); + let flows = 20_000u16; + r.src + .events + .borrow_mut() + .extend((0..flows).map(|port| ip6_event(3, port, 1000, 1))); + r.w.tick(r.t0 + TICK); + assert_eq!( + r.shared.snapshot().flows.unwrap().active, + usize::from(flows) + ); + // A sample at a new rate: all 20,000 are queued for export. + r.src + .events + .borrow_mut() + .push_back(ip6_event(3, 60_000, 500, 2)); + let mut t = r.t0 + 2 * TICK; + for _ in 0..20 { + r.w.tick(t); + t += TICK; + } + let out = &r.w.flows.as_ref().unwrap().outs[0]; + assert_eq!(out.records, u64::from(flows), "{datagram}"); + assert_eq!(r.w.collectors[1].budget_drops, 0, "{datagram}"); + assert!(r + .net + .sent + .borrow() + .iter() + .all(|(to, d)| to.port() != 4739 || d.len() <= datagram)); + } + /// More flows than one tick's send budget carries all go out at a /// stop: none is dropped by the budget as the worker exits. #[test]