feat(flow-export): path-mtu sizes datagrams for a tunnel to the collectors

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 `<state-dir>/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 <bytes>`** (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 <name> ipfix <ip>:<port>` sends flow records (RFC 7011) instead of samples.

conf/example.conf

@@ -1199,14 +1199,18 @@ 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
 #                                 # the source every collector sees; must
 #                                 # 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), <ip>:<port>

crates/common/src/config.rs

@@ -860,6 +860,15 @@ pub enum ModuleDirective {
         bytes: u32,
         line: usize,
     },
+    /// `path-mtu <bytes>` — 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 <name> sflow|ipfix <ip>:<port> [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<ModuleDirective, Confi
             }
             Ok(ModuleDirective::FlowHeaderBytes { bytes, line })
         }),
+        "path-mtu" => 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",

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),

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<Collector>,
+    /// 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]