From 0d3c6d2c42a5d6d4d931052c2a6e6e7dab5d3993 Mon Sep 17 00:00:00 2001 From: Madison Grubb Date: Thu, 22 Oct 2026 16:02:06 -0400 Subject: [PATCH] feat(mermin): decapsulate vxlan and geneve flows in ebpf Keying the flow map on the inner 5-tuple in eBPF means both tunnel directions and the veth copy of a flow share one FlowStats entry, so userspace no longer has to dedupe by deleting map entries. It also gives tunneled flows real TCP state, flags, timeouts, and inner byte counts, and keeps distinct inner flows apart when they share an outer UDP port. The outer headers and VNI are stored in FlowStats as tunnel metadata. FlowStats grows to 240 bytes, so the map schema version moves to 2. --- docs/configuration/default/config.hcl | 2 +- .../reference/flow-processing-pipeline.md | 2 +- docs/getting-started/quickstart-guide.md | 5 +- ...ce-visibility-and-traffic-decapsulation.md | 2 +- mermin-common/src/lib.rs | 38 +- mermin-ebpf/src/main.rs | 348 ++++++++++++-- mermin/src/filter/source.rs | 7 + mermin/src/main.rs | 23 +- mermin/src/packet.rs | 19 - mermin/src/packet/parser.rs | 426 ------------------ mermin/src/packet/types.rs | 103 ----- mermin/src/runtime/conf.rs | 6 +- mermin/src/span/direction.rs | 7 + mermin/src/span/producer.rs | 131 ++---- mermin/tests/e2e/cni/flannel-vxlan.hcl | 41 ++ mermin/tests/e2e/cni/test.sh | 32 +- mermin/tests/e2e/common/setup.sh | 6 +- mermin/tests/integration.rs | 42 ++ 18 files changed, 532 insertions(+), 708 deletions(-) delete mode 100644 mermin/src/packet.rs delete mode 100644 mermin/src/packet/parser.rs delete mode 100644 mermin/src/packet/types.rs create mode 100644 mermin/tests/e2e/cni/flannel-vxlan.hcl diff --git a/docs/configuration/default/config.hcl b/docs/configuration/default/config.hcl index aef2dc1db..4500e2c44 100644 --- a/docs/configuration/default/config.hcl +++ b/docs/configuration/default/config.hcl @@ -32,7 +32,7 @@ pipeline { parser { # Tunnel port detection (IANA defaults shown) geneve_port = 6081 # IANA default for Geneve - vxlan_port = 4789 # IANA default for VXLAN + vxlan_port = 4789 # IANA default for VXLAN. Linux Flannel uses 8472 wireguard_port = 51820 # IANA default for WireGuard } diff --git a/docs/configuration/reference/flow-processing-pipeline.md b/docs/configuration/reference/flow-processing-pipeline.md index 7604a508f..724ba2699 100644 --- a/docs/configuration/reference/flow-processing-pipeline.md +++ b/docs/configuration/reference/flow-processing-pipeline.md @@ -18,7 +18,7 @@ Configures eBPF-level flow tracking. - `flow_stats_capacity` attribute Max entries in the `FLOW_STATS` eBPF map. The map is created at this fixed size at program load time and does not resize at runtime. Uses `BPF_F_NO_PREALLOC`: - memory grows on demand as flows are tracked, up to this limit — so actual kernel memory equals the number of active concurrent flows × ~270 bytes (FlowStats: 192B + FlowKey: 40B aligned + htab_elem overhead: ~38B). + memory grows on demand as flows are tracked, up to this limit — so actual kernel memory equals the number of active concurrent flows × ~318 bytes (FlowStats: 240B + FlowKey: 40B aligned + htab_elem overhead: ~38B). Once full, new flows are dropped. **Type:** Integer diff --git a/docs/getting-started/quickstart-guide.md b/docs/getting-started/quickstart-guide.md index 2cde77471..627fefeeb 100644 --- a/docs/getting-started/quickstart-guide.md +++ b/docs/getting-started/quickstart-guide.md @@ -270,14 +270,13 @@ Default: **Use cases**: Fine-tuning for specific CNI setups, reducing monitored interface count {% hint style="info" %} -Mermin's goal is to show you pod-to-pod traffic which is exposed by Virtual Ethernet Devices, which match patterns like `"veth*", "gke*", "cali*"`. Currently, bridge interfaces like `"tun*"` or `flannel*` are ignored, -because Mermin does not support parsing tunneled/encapsulated traffic. This feature will come very soon. +Mermin's goal is to show you pod-to-pod traffic. Use `veth*` for same-node traffic and tunnel interfaces (`flannel*`, `tunl*`, `vxlan*`) or node uplinks (`eth*`) for inter-node overlay traffic. Mermin parses VXLAN and Geneve encapsulation automatically. Set `parser.vxlan_port` to match your CNI. Linux Flannel uses `8472` instead of the IANA default `4789`. {% endhint %} **Physical Interfaces Only:** {% hint style="warning" %} -Most of the traffic on the physical interfaces will be ignored, because Mermin currently lacks support for tunneled/encapsulated traffic. +On node uplinks (`eth*`), only tunneled flows (VXLAN/Geneve UDP ports) are deep-parsed into inner 5-tuples. Plain traffic on those interfaces is still tracked, but avoid monitoring both `veth*` and `eth*` for the same paths unless you need uplink-level tunnel metadata. {% endhint %} Monitor only physical network interfaces for inter-node traffic: diff --git a/docs/troubleshooting/interface-visibility-and-traffic-decapsulation.md b/docs/troubleshooting/interface-visibility-and-traffic-decapsulation.md index 88accb0f9..44511bd71 100644 --- a/docs/troubleshooting/interface-visibility-and-traffic-decapsulation.md +++ b/docs/troubleshooting/interface-visibility-and-traffic-decapsulation.md @@ -277,7 +277,7 @@ discovery "instrument" { parser { geneve_port = 6081 - vxlan_port = 4789 + vxlan_port = 8472 # Linux Flannel. Use 4789 for IANA-default VXLAN wireguard_port = 51820 } ``` diff --git a/mermin-common/src/lib.rs b/mermin-common/src/lib.rs index 65ce7b668..5f9322dda 100644 --- a/mermin-common/src/lib.rs +++ b/mermin-common/src/lib.rs @@ -38,7 +38,7 @@ //! eBPF/userspace compatibility. //! //! - [`FlowKey`]: Normalized 5-tuple for bidirectional flow aggregation (Community ID compatible) -//! - [`FlowStats`]: Per-flow counters and metadata stored in eBPF maps (192 bytes) +//! - [`FlowStats`]: Per-flow counters and metadata stored in eBPF maps (240 bytes) //! - [`FlowEvent`]: New-flow ring buffer event carrying the outermost key plus unparsed packet bytes (234 bytes) pub mod ah; @@ -354,6 +354,10 @@ pub struct FlowStats { pub src_ip: [u8; 16], /// Original destination IP pub dst_ip: [u8; 16], + /// Outer source IP of the creating packet when the flow was decapsulated (VXLAN/Geneve) + pub tunnel_src_ip: [u8; 16], + /// Outer destination IP of the creating packet when the flow was decapsulated + pub tunnel_dst_ip: [u8; 16], // === 4-byte aligned (u32) - 20 bytes === /// Network interface index where flow was observed @@ -366,6 +370,8 @@ pub struct FlowStats { pub tcp_txn_count: u32, /// Average jitter in nanoseconds (only meaningful when protocol == TCP) pub tcp_jitter_avg_ns: u32, + /// VXLAN/Geneve VNI (only meaningful when tunnel_type != None) + pub tunnel_id: u32, // === 2-byte aligned (u16) - 6 bytes === /// EtherType @@ -374,6 +380,10 @@ pub struct FlowStats { pub src_port: u16, /// Original destination port pub dst_port: u16, + /// Outer UDP source port of the creating packet (only meaningful when tunnel_type != None) + pub tunnel_src_port: u16, + /// Outer UDP destination port of the creating packet (only meaningful when tunnel_type != None) + pub tunnel_dst_port: u16, // === 6-byte array (MAC address) - 6 bytes === /// Source MAC address @@ -418,6 +428,10 @@ pub struct FlowStats { pub reverse_icmp_type: u8, /// ICMP message code for reverse direction (only meaningful when protocol == ICMP or ICMPv6) pub reverse_icmp_code: u8, + /// Encapsulation eBPF removed to find the flow key (None for plain traffic) + pub tunnel_type: TunnelType, + /// Outer IP version (only meaningful when tunnel_type != None) + pub tunnel_ip_version: IpVersion, // === Process metadata - 20 bytes === /// Process ID that created/connected this flow (0 if unknown) @@ -456,7 +470,7 @@ impl FlowStats { /// # Examples /// /// ``` - /// # use mermin_common::{FlowStats, ConnectionState, Direction, IpVersion}; + /// # use mermin_common::{FlowStats, ConnectionState, Direction, IpVersion, TunnelType}; /// # use mermin_common::eth::EtherType; /// # use mermin_common::ip::IpProto; /// # let mut stats = FlowStats { @@ -473,7 +487,9 @@ impl FlowStats { /// # tcp_state: ConnectionState::default(), forward_tcp_flags: 0, /// # reverse_tcp_flags: 0, icmp_type: 0, icmp_code: 0, /// # reverse_icmp_type: 0, reverse_icmp_code: 0, - /// # pid: 0, comm: [0; 16], + /// # pid: 0, comm: [0; 16], tunnel_src_ip: [0; 16], tunnel_dst_ip: [0; 16], + /// # tunnel_id: 0, tunnel_src_port: 0, tunnel_dst_port: 0, + /// # tunnel_type: TunnelType::None, tunnel_ip_version: IpVersion::Unknown, /// # }; /// stats.tcp_flags = FlowStats::TCP_FLAG_SYN | FlowStats::TCP_FLAG_ACK; /// @@ -820,6 +836,13 @@ mod tests { reverse_icmp_code: 0, pid: 0, comm: [0; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: TunnelType::None, + tunnel_ip_version: IpVersion::Unknown, } } @@ -1141,13 +1164,12 @@ mod tests { /// 2. Alignment is consistent #[test] fn test_flow_stats_memory_layout() { - // Verify optimized memory layout (192 bytes) - // 88 (11 u64) + 32 (2 IP arrays) + 20 (5 u32) + 6 (3 u16) + 6 (MAC) - // + 19 (u8: 11 base + 4 TCP + 4 ICMP) + 1 (pad) + 4 (pid) + 16 (comm) = 192 + // 88 (11 u64) + 64 (4 IP arrays) + 24 (6 u32) + 10 (5 u16) + 6 (MAC) + // + 21 (u8: 11 base + 4 TCP + 4 ICMP + 2 tunnel) + 3 (pad) + 4 (pid) + 16 (comm) + 4 (tail pad) = 240 assert_eq!( size_of::(), - 192, - "FlowStats size MUST be 192 bytes for eBPF/userspace compatibility" + 240, + "FlowStats size MUST be 240 bytes for eBPF/userspace compatibility" ); // Verify alignment (critical for correct memory access in eBPF) diff --git a/mermin-ebpf/src/main.rs b/mermin-ebpf/src/main.rs index 95ecf9570..9417c8688 100644 --- a/mermin-ebpf/src/main.rs +++ b/mermin-ebpf/src/main.rs @@ -94,7 +94,7 @@ use aya_ebpf::{ bindings::TC_ACT_UNSPEC, helpers::bpf_ktime_get_boot_ns, macros::{classifier, fexit, lsm, map}, - maps::{HashMap, PerCpuArray, RingBuf}, + maps::{Array, HashMap, PerCpuArray, RingBuf}, programs::{LsmContext, TcContext}, }; #[cfg(not(feature = "test"))] @@ -102,11 +102,11 @@ use mermin_common::FlowEvent; #[cfg(not(feature = "test"))] use mermin_common::LogEntry; use mermin_common::{ - ConnectionState, Direction, EbpfError, FlowKey, FlowStats, IpVersion, eth, + ConnectionState, Direction, EbpfError, FlowKey, FlowStats, IpVersion, TunnelType, eth, eth::EtherType, - icmp, + geneve, icmp, ip::{IpProto, ipv4, ipv6}, - tcp, udp, + tcp, udp, vxlan, }; /// Size of unparsed packet data captured per flow event (in bytes). @@ -184,6 +184,17 @@ static mut FLOW_EVENT_SCRATCH: PerCpuArray = PerCpuArray::with_max_en #[map] static mut FLOW_KEY_SCRATCH: PerCpuArray = PerCpuArray::with_max_entries(1, 0); +/// Outer key of a decapsulated packet, kept off the stack until the new-flow path copies it. +#[cfg(not(feature = "test"))] +#[map] +static mut TUNNEL_KEY_SCRATCH: PerCpuArray = PerCpuArray::with_max_entries(1, 0); + +/// UDP ports to decapsulate, written by userspace from the `parser` config. +/// Index 0 is VXLAN and index 1 is Geneve. A zero port is disabled. +#[cfg(not(feature = "test"))] +#[map] +static mut TUNNEL_PORTS: Array = Array::with_max_entries(2, 0); + /// Map to track listening ports for client/server direction inference. /// Key: ListeningPortKey (port + protocol), Value: 1 (presence marker) /// Max entries: 1024 — covers dense nodes (~500 pods × 2 listening ports each). @@ -611,10 +622,41 @@ fn try_flow_stats(ctx: &TcContext, direction: Direction) -> Result = None; + if flow_key.protocol == IpProto::Udp { + let vxlan_port = unsafe { TUNNEL_PORTS.get(0) }.copied().unwrap_or(0); + let geneve_port = unsafe { TUNNEL_PORTS.get(1) }.copied().unwrap_or(0); + if let Some((tunnel_type, vni, inner_offset)) = + parse_tunnel(ctx, flow_key, l4_offset, vxlan_port, geneve_port) + { + let outer_key_ptr = unsafe { + TUNNEL_KEY_SCRATCH + .get_ptr_mut(0) + .ok_or(EbpfError::ScratchUnavailable)? + }; + unsafe { *outer_key_ptr = *flow_key }; + match parse_flow_key(ctx, flow_key, inner_offset) { + Ok(inner) => { + (eth_type, l4_offset) = inner; + frame_offset = inner_offset; + tunnel = Some((tunnel_type, vni, outer_key_ptr)); + } + // Non-IP inner frames (e.g. ARP) stay accounted on the outer UDP flow. + Err(_) => unsafe { *flow_key = *outer_key_ptr }, + } + } + } + let l3_offset = frame_offset + eth::ETH_LEN; + let frame_len = (ctx.len() as u64).saturating_sub(frame_offset as u64); // Normalize the flow key for map lookup (bidirectional aggregation) let normalized_key = flow_key.normalize(); + let swapped = flow_key.should_normalize(); let timestamp = unsafe { bpf_ktime_get_boot_ns() }; #[allow(static_mut_refs)] @@ -650,18 +692,21 @@ fn try_flow_stats(ctx: &TcContext, direction: Direction) -> Result Result Result Result Result Result<(EtherType, usize), EbpfError> { - if eth::ETH_LEN > ctx.len() as usize { +fn parse_flow_key( + ctx: &TcContext, + key: &mut FlowKey, + frame_offset: usize, +) -> Result<(EtherType, usize), EbpfError> { + if frame_offset + eth::ETH_LEN > ctx.len() as usize { return Err(EbpfError::MalformedHeader); } @@ -820,14 +876,12 @@ fn parse_flow_key(ctx: &TcContext, key: &mut FlowKey) -> Result<(EtherType, usiz key.src_port = 0; key.dst_port = 0; - let mut offset = 0; - let eth_type: EtherType = ctx - .load(eth::ETH_ETHER_TYPE_OFFSET) + .load(frame_offset + eth::ETH_ETHER_TYPE_OFFSET) .map_err(|_| EbpfError::OutOfBounds)?; - offset += eth::ETH_LEN; + let offset = frame_offset + eth::ETH_LEN; - offset = match eth_type { + let offset = match eth_type { EtherType::Ipv4 => { if offset + ipv4::IPV4_LEN > ctx.len() as usize { return Err(EbpfError::OutOfBounds); @@ -931,6 +985,72 @@ fn parse_flow_key(ctx: &TcContext, key: &mut FlowKey) -> Result<(EtherType, usiz Ok((eth_type, offset)) } +/// Stores the outer headers oriented like the normalized key, which is what the span reports +/// as source/destination. `swapped` is set when normalizing reversed the inner packet. +#[inline(always)] +fn record_tunnel( + stats: &mut FlowStats, + tunnel_type: TunnelType, + vni: u32, + outer: &FlowKey, + swapped: bool, +) { + stats.tunnel_type = tunnel_type; + stats.tunnel_id = vni; + stats.tunnel_ip_version = outer.ip_version; + if swapped { + stats.tunnel_src_ip = outer.dst_ip; + stats.tunnel_dst_ip = outer.src_ip; + stats.tunnel_src_port = outer.dst_port; + stats.tunnel_dst_port = outer.src_port; + } else { + stats.tunnel_src_ip = outer.src_ip; + stats.tunnel_dst_ip = outer.dst_ip; + stats.tunnel_src_port = outer.src_port; + stats.tunnel_dst_port = outer.dst_port; + } +} + +/// Detects VXLAN or Geneve on an outer UDP key and returns +/// `(tunnel_type, vni, inner_frame_offset)`. Anything else, including a Geneve payload +/// that is not Ethernet, stays keyed on the outer headers. +#[inline(always)] +fn parse_tunnel( + ctx: &TcContext, + outer: &FlowKey, + l4_offset: usize, + vxlan_port: u16, + geneve_port: u16, +) -> Option<(TunnelType, u32, usize)> { + let hdr = l4_offset + udp::UDP_LEN; + let is_port = |port: u16| port != 0 && (outer.src_port == port || outer.dst_port == port); + + if is_port(vxlan_port) { + let flags: vxlan::Flags = ctx.load(hdr).ok()?; + if !vxlan::flags_i_flag(flags) { + return None; + } + let vni: vxlan::Vni = ctx.load(hdr + 4).ok()?; + return Some((TunnelType::Vxlan, vxlan::vni(vni), hdr + vxlan::VXLAN_LEN)); + } + + if is_port(geneve_port) { + let ver_opt_len: geneve::VerOptLen = ctx.load(hdr).ok()?; + let protocol: u16 = ctx.load(hdr + 2).ok()?; + if geneve::ver(ver_opt_len) != 0 || protocol != EtherType::Ethernet as u16 { + return None; + } + let vni: geneve::Vni = ctx.load(hdr + 4).ok()?; + return Some(( + TunnelType::Geneve, + geneve::vni(vni), + hdr + geneve::total_hdr_len(ver_opt_len), + )); + } + + None +} + /// Capture metadata (DSCP, ECN, TTL, flow label, ICMP type/code) for a specific direction. /// Used both during initial flow creation and when userspace resets metadata flags for re-capture. #[inline(always)] @@ -938,16 +1058,17 @@ fn capture_direction_metadata( ctx: &TcContext, stats: &mut FlowStats, eth_type: EtherType, + l3_offset: usize, l4_offset: usize, is_forward: bool, ) -> Result<(), EbpfError> { match eth_type { EtherType::Ipv4 => { let dscp_ecn: ipv4::DscpEcn = ctx - .load(eth::ETH_LEN + ipv4::IPV4_DSCP_ECN_OFFSET) + .load(l3_offset + ipv4::IPV4_DSCP_ECN_OFFSET) .map_err(|_| EbpfError::OutOfBounds)?; let ttl: ipv4::Ttl = ctx - .load(eth::ETH_LEN + ipv4::IPV4_TTL_OFFSET) + .load(l3_offset + ipv4::IPV4_TTL_OFFSET) .map_err(|_| EbpfError::OutOfBounds)?; if is_forward { @@ -961,9 +1082,9 @@ fn capture_direction_metadata( } } EtherType::Ipv6 => { - let vtcfl: ipv6::Vcf = ctx.load(eth::ETH_LEN).map_err(|_| EbpfError::OutOfBounds)?; + let vtcfl: ipv6::Vcf = ctx.load(l3_offset).map_err(|_| EbpfError::OutOfBounds)?; let hop_limit: ipv6::HopLimit = ctx - .load(eth::ETH_LEN + ipv6::IPV6_HOP_LIMIT_OFFSET) + .load(l3_offset + ipv6::IPV6_HOP_LIMIT_OFFSET) .map_err(|_| EbpfError::OutOfBounds)?; if is_forward { @@ -1466,6 +1587,13 @@ mod tests { reverse_icmp_code: 0, pid: 0, comm: [0; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: TunnelType::None, + tunnel_ip_version: IpVersion::Unknown, } } @@ -1603,12 +1731,132 @@ mod tests { pkt } + // Outer Eth + IPv4 + UDP(54321 -> dst_port) wrapping the plain IPv4 TCP frame. + fn build_tunnel_packet(dst_port: u16, tunnel_hdr: &[u8]) -> Vec { + let mut pkt = Vec::new(); + pkt.extend_from_slice(&[0x02, 0, 0, 0, 0, 0x02, 0x02, 0, 0, 0, 0, 0x01, 0x08, 0x00]); + pkt.extend_from_slice(&[ + 0x45, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 64, 17, 0x00, 0x00, 172, 18, 0, 3, 172, + 18, 0, 4, + ]); + pkt.extend_from_slice(&54321u16.to_be_bytes()); + pkt.extend_from_slice(&dst_port.to_be_bytes()); + pkt.extend_from_slice(&[0x00, 0x00, 0x00, 0x00]); + pkt.extend_from_slice(tunnel_hdr); + pkt.extend_from_slice(&build_ipv4_tcp_packet()); + pkt + } + + const VXLAN_HDR: [u8; 8] = [0x08, 0, 0, 0, 0x12, 0x34, 0x56, 0]; + + fn decap(pkt: Vec) -> Option<(TunnelType, u32, FlowKey, EtherType, usize, usize)> { + let ctx = TcContext::new(pkt); + let mut key = FlowKey::default(); + let (_, l4_offset) = parse_flow_key(&ctx, &mut key, 0).unwrap(); + let (tunnel_type, vni, inner) = parse_tunnel(&ctx, &key, l4_offset, 4789, 6081)?; + let (eth_type, inner_l4) = parse_flow_key(&ctx, &mut key, inner).unwrap(); + Some((tunnel_type, vni, key, eth_type, inner, inner_l4)) + } + + #[test] + fn test_parse_tunnel_vxlan_keys_inner_flow() { + let (tunnel_type, vni, key, eth_type, inner, inner_l4) = + decap(build_tunnel_packet(4789, &VXLAN_HDR)).unwrap(); + + assert_eq!(tunnel_type, TunnelType::Vxlan); + assert_eq!(vni, 0x123456); + assert_eq!(inner, 50); + assert_eq!(inner_l4, 50 + 34); + assert_eq!(eth_type, EtherType::Ipv4); + assert_eq!(key.protocol, IpProto::Tcp); + assert_eq!(key.src_ip[..4], [192, 168, 1, 10]); + assert_eq!(key.dst_ip[..4], [10, 0, 0, 5]); + assert_eq!((key.src_port, key.dst_port), (12345, 443)); + } + + #[test] + fn test_parse_tunnel_inner_metadata_uses_inner_headers() { + let pkt = build_tunnel_packet(4789, &VXLAN_HDR); + let len = pkt.len(); + let (_, _, key, eth_type, inner, inner_l4) = decap(pkt.clone()).unwrap(); + let ctx = TcContext::new(pkt); + let mut stats = create_test_flow_stats(eth_type, key.protocol); + capture_direction_metadata( + &ctx, + &mut stats, + eth_type, + inner + eth::ETH_LEN, + inner_l4, + true, + ) + .unwrap(); + + // The inner TCP header carries DSCP 46 and TTL 64; the outer IPv4 header has 0 and 64. + assert_eq!(stats.ip_dscp, 46); + assert_eq!(len - inner, 60); + } + + #[test] + fn test_parse_tunnel_geneve_skips_options() { + let geneve_hdr = [0x01, 0x00, 0x65, 0x58, 0x00, 0xAB, 0xCD, 0x00, 1, 2, 3, 4]; + let (tunnel_type, vni, key, _, inner, _) = + decap(build_tunnel_packet(6081, &geneve_hdr)).unwrap(); + + assert_eq!(tunnel_type, TunnelType::Geneve); + assert_eq!(vni, 0xABCD); + assert_eq!(inner, 42 + 12); + assert_eq!((key.src_port, key.dst_port), (12345, 443)); + } + + #[test] + fn test_parse_tunnel_rejects_non_tunnel_traffic() { + // Wrong port, VXLAN without the I flag, and Geneve carrying IPv4 instead of Ethernet. + assert!(decap(build_tunnel_packet(5000, &VXLAN_HDR)).is_none()); + assert!(decap(build_tunnel_packet(4789, &[0, 0, 0, 0, 0, 0, 1, 0])).is_none()); + assert!(decap(build_tunnel_packet(6081, &[0, 0, 0x08, 0x00, 0, 0, 1, 0])).is_none()); + } + + #[test] + fn test_record_tunnel_follows_normalized_orientation() { + let pkt = build_tunnel_packet(4789, &VXLAN_HDR); + let ctx = TcContext::new(pkt); + let mut outer = FlowKey::default(); + parse_flow_key(&ctx, &mut outer, 0).unwrap(); + let mut stats = create_test_flow_stats(EtherType::Ipv4, IpProto::Tcp); + + // Inner 192.168.1.10 -> 10.0.0.5 normalizes to 10.0.0.5 -> 192.168.1.10, + // so the outer endpoints swap with it. + record_tunnel(&mut stats, TunnelType::Vxlan, 1, &outer, true); + assert_eq!(stats.tunnel_src_ip[..4], [172, 18, 0, 4]); + assert_eq!(stats.tunnel_dst_ip[..4], [172, 18, 0, 3]); + assert_eq!( + (stats.tunnel_src_port, stats.tunnel_dst_port), + (4789, 54321) + ); + + record_tunnel(&mut stats, TunnelType::Vxlan, 1, &outer, false); + assert_eq!(stats.tunnel_src_ip[..4], [172, 18, 0, 3]); + assert_eq!( + (stats.tunnel_src_port, stats.tunnel_dst_port), + (54321, 4789) + ); + } + + #[test] + fn test_parse_tunnel_disabled_port_is_ignored() { + let pkt = build_tunnel_packet(4789, &VXLAN_HDR); + let ctx = TcContext::new(pkt); + let mut key = FlowKey::default(); + let (_, l4_offset) = parse_flow_key(&ctx, &mut key, 0).unwrap(); + assert!(parse_tunnel(&ctx, &key, l4_offset, 0, 0).is_none()); + } + #[test] fn test_parse_flow_key_ipv4_tcp() { let pkt = build_ipv4_tcp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); let (ether_type, _l4_offset) = result.unwrap(); let expected_src = [192, 168, 1, 10, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; let expected_dst = [10, 0, 0, 5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]; @@ -1627,7 +1875,7 @@ mod tests { let pkt = build_ipv4_udp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); assert_eq!(ether_type, EtherType::Ipv4); assert_eq!(flow_key.protocol, IpProto::Udp); @@ -1640,7 +1888,7 @@ mod tests { let pkt = build_ipv4_icmp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); assert_eq!(ether_type, EtherType::Ipv4); assert_eq!(flow_key.protocol, IpProto::Icmp); @@ -1653,7 +1901,7 @@ mod tests { let pkt = build_ipv6_tcp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, _l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); assert_eq!(ether_type, EtherType::Ipv6); assert_eq!(flow_key.ip_version, IpVersion::V6); @@ -1667,7 +1915,7 @@ mod tests { let pkt = vec![0x00, 0x01, 0x02]; let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); assert!(result.is_err()); assert_eq!(result.unwrap_err(), EbpfError::MalformedHeader); @@ -1682,7 +1930,7 @@ mod tests { pkt.extend_from_slice(&[0x45, 0x00, 0x00, 0x3C, 0x1C, 0x46, 0x40, 0x00, 0x40, 0x06]); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); assert!(result.is_err()); assert_eq!(result.unwrap_err(), EbpfError::OutOfBounds); @@ -1698,7 +1946,7 @@ mod tests { pkt.extend_from_slice(&[0x00; 19]); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); assert!(result.is_err()); assert_eq!(result.unwrap_err(), EbpfError::MalformedHeader); @@ -1712,7 +1960,7 @@ mod tests { pkt.extend_from_slice(&[0x08, 0x06]); // ARP let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); assert!(result.is_err()); assert_eq!(result.unwrap_err(), EbpfError::UnsupportedEtherType); @@ -1723,13 +1971,14 @@ mod tests { let pkt = build_ipv4_tcp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); let mut stats = create_test_flow_stats(ether_type, flow_key.protocol); let result = parse_metadata(&ctx, &mut stats, l4_offset); let parsed_offset = result.unwrap(); - capture_direction_metadata(&ctx, &mut stats, ether_type, l4_offset, true).unwrap(); + capture_direction_metadata(&ctx, &mut stats, ether_type, eth::ETH_LEN, l4_offset, true) + .unwrap(); assert!(result.is_ok()); assert_eq!(stats.src_mac, [0x11, 0x22, 0x33, 0x44, 0x55, 0x66]); @@ -1745,13 +1994,14 @@ mod tests { let pkt = build_ipv6_tcp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); let mut stats = create_test_flow_stats(ether_type, flow_key.protocol); let result = parse_metadata(&ctx, &mut stats, l4_offset); let parsed_offset = result.unwrap(); - capture_direction_metadata(&ctx, &mut stats, ether_type, l4_offset, true).unwrap(); + capture_direction_metadata(&ctx, &mut stats, ether_type, eth::ETH_LEN, l4_offset, true) + .unwrap(); assert!(result.is_ok()); assert_eq!(stats.src_mac, [0x11, 0x22, 0x33, 0x44, 0x55, 0x66]); @@ -1767,13 +2017,14 @@ mod tests { let pkt = build_ipv4_udp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); let mut stats = create_test_flow_stats(ether_type, flow_key.protocol); let result = parse_metadata(&ctx, &mut stats, l4_offset); let parsed_offset = result.unwrap(); - capture_direction_metadata(&ctx, &mut stats, ether_type, l4_offset, true).unwrap(); + capture_direction_metadata(&ctx, &mut stats, ether_type, eth::ETH_LEN, l4_offset, true) + .unwrap(); assert!(result.is_ok()); assert_eq!(stats.ip_ttl, 128); @@ -1785,13 +2036,14 @@ mod tests { let pkt = build_ipv4_icmp_packet(); let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); let mut stats = create_test_flow_stats(ether_type, flow_key.protocol); let result = parse_metadata(&ctx, &mut stats, l4_offset); let parsed_offset = result.unwrap(); - capture_direction_metadata(&ctx, &mut stats, ether_type, l4_offset, true).unwrap(); + capture_direction_metadata(&ctx, &mut stats, ether_type, eth::ETH_LEN, l4_offset, true) + .unwrap(); assert!(result.is_ok()); assert_eq!(stats.icmp_type, 8); @@ -1823,7 +2075,7 @@ mod tests { let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key).unwrap(); + let (ether_type, l4_offset) = parse_flow_key(&ctx, &mut flow_key, 0).unwrap(); let mut stats = create_test_flow_stats(ether_type, flow_key.protocol); let result = parse_metadata(&ctx, &mut stats, l4_offset); @@ -1837,7 +2089,8 @@ mod tests { let pkt = build_ipv4_tcp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv4, IpProto::Tcp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, 34, true); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, eth::ETH_LEN, 34, true); assert!(result.is_ok()); assert_eq!(stats.ip_dscp, 46); @@ -1855,7 +2108,8 @@ mod tests { let pkt = build_ipv4_tcp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv4, IpProto::Tcp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, 34, false); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, eth::ETH_LEN, 34, false); assert!(result.is_ok()); assert_eq!(stats.ip_dscp, 0); @@ -1871,7 +2125,8 @@ mod tests { let pkt = build_ipv6_tcp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv6, IpProto::Tcp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv6, 54, true); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv6, eth::ETH_LEN, 54, true); assert!(result.is_ok()); assert_eq!(stats.ip_dscp, 3); @@ -1889,7 +2144,8 @@ mod tests { let pkt = build_ipv6_tcp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv6, IpProto::Tcp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv6, 54, false); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv6, eth::ETH_LEN, 54, false); assert!(result.is_ok()); assert_eq!(stats.ip_dscp, 0); @@ -1907,7 +2163,8 @@ mod tests { let pkt = build_ipv4_icmp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv4, IpProto::Icmp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, 34, true); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, eth::ETH_LEN, 34, true); assert!(result.is_ok()); assert_eq!(stats.icmp_type, 8); // Echo Request @@ -1921,7 +2178,8 @@ mod tests { let pkt = build_ipv4_icmp_packet(); let ctx = TcContext::new(pkt); let mut stats = create_test_flow_stats(EtherType::Ipv4, IpProto::Icmp); - let result = capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, 34, false); + let result = + capture_direction_metadata(&ctx, &mut stats, EtherType::Ipv4, eth::ETH_LEN, 34, false); assert!(result.is_ok()); assert_eq!(stats.icmp_type, 0); @@ -1953,7 +2211,7 @@ mod tests { let ctx = TcContext::new(pkt); let mut flow_key = FlowKey::default(); - let result = parse_flow_key(&ctx, &mut flow_key); + let result = parse_flow_key(&ctx, &mut flow_key, 0); assert!(result.is_err()); assert_eq!(result.unwrap_err(), EbpfError::UnsupportedProtocol); diff --git a/mermin/src/filter/source.rs b/mermin/src/filter/source.rs index 3e214c92a..9402fb45f 100644 --- a/mermin/src/filter/source.rs +++ b/mermin/src/filter/source.rs @@ -622,6 +622,13 @@ mod tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, } } diff --git a/mermin/src/main.rs b/mermin/src/main.rs index 9b91d469a..57e7f1344 100644 --- a/mermin/src/main.rs +++ b/mermin/src/main.rs @@ -7,7 +7,6 @@ mod k8s; mod listening_ports; mod metrics; mod otlp; -mod packet; mod runtime; mod span; @@ -139,10 +138,11 @@ fn main() { /// - Map key/value types /// /// History: -/// - v1: Initial schema version (current) +/// - v1: Initial schema version +/// - v2: FlowStats gains VXLAN/Geneve tunnel fields (current) /// /// CI check in place: .github/workflows/ci.yaml (schema_version_check job) -const EBPF_MAP_SCHEMA_VERSION: u8 = 1; +const EBPF_MAP_SCHEMA_VERSION: u8 = 2; /// Timeout for individual export operations (in seconds). /// If an export takes longer than this, it's considered timed out and the span may be lost. @@ -802,7 +802,7 @@ async fn start_pipeline( exporter: Arc, ) -> Result { let EbpfResources { - ebpf, + mut ebpf, flow_events_ringbuf, log_events_ringbuf, flow_stats_map, @@ -811,6 +811,21 @@ async fn start_pipeline( host_netns, } = ebpf_resources; + // Written on every pipeline start so a reloaded `parser` config takes effect. + let mut tunnel_ports: aya::maps::Array<_, u16> = aya::maps::Array::try_from( + ebpf.map_mut("TUNNEL_PORTS") + .ok_or_else(|| MerminError::internal("TUNNEL_PORTS not found in eBPF object"))?, + ) + .map_err(|e| MerminError::internal(format!("failed to convert TUNNEL_PORTS: {e}")))?; + for (index, port) in [conf.parser.vxlan_port, conf.parser.geneve_port] + .into_iter() + .enumerate() + { + tunnel_ports + .set(index as u32, port, 0) + .map_err(|e| MerminError::internal(format!("failed to set TUNNEL_PORTS: {e}")))?; + } + let mut pipeline_components = ComponentManager::new(); // Spawn eBPF log consumer with return channel for ring buffer recovery. diff --git a/mermin/src/packet.rs b/mermin/src/packet.rs deleted file mode 100644 index 02a728244..000000000 --- a/mermin/src/packet.rs +++ /dev/null @@ -1,19 +0,0 @@ -//! Deep packet parsing module for extracting innermost 5-tuples from tunneled traffic. -//! -//! This module provides functionality to parse raw packet bytes (captured from eBPF) -//! and extract: -//! - Innermost 5-tuple for correct Community ID calculation -//! - Tunnel metadata (VNI, type, outer headers) -//! - L2/L3 metadata (MACs, DSCP, TTL, etc.) -//! -//! **Key Optimization**: eBPF already parsed outer headers into FlowStats, so we only -//! parse the UNPARSED portion (tunnels) and skip redundant parsing for plain traffic. -//! -//! Supports: -//! - VXLAN (UDP port 4789) -//! - Geneve (UDP port 6081) -//! - GRE (IP protocol 47) - TODO -//! - Plain (non-tunneled) packets (fast path - no parsing!) - -pub mod parser; -pub mod types; diff --git a/mermin/src/packet/parser.rs b/mermin/src/packet/parser.rs deleted file mode 100644 index f60c76359..000000000 --- a/mermin/src/packet/parser.rs +++ /dev/null @@ -1,426 +0,0 @@ -//! Deep packet parser for extracting innermost 5-tuples and tunnel metadata. -//! -//! Handles VXLAN (port 4789), Geneve (port 6081), GRE (proto 47), and plain packets. -//! -//! The parser extracts: -//! 1. Innermost 5-tuple (for correct Community ID calculation) -//! 2. Outermost headers (tunnel transport) -//! 3. Tunnel metadata (VNI, type, etc.) - -use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; - -use mermin_common::{FlowKey, TunnelType, eth::EtherType, ip::IpProto}; - -use crate::packet::types::{ - FiveTuple, InnerHeaders, IpMetadata, L2Metadata, OuterHeaders, ParseError, ParsedPacket, - TunnelInfo, -}; - -/// Returns true if the flow is likely tunneled. Fast check to avoid deep parsing for plain traffic. -pub fn is_tunnel( - flow_key: &FlowKey, - vxlan_port: u16, - geneve_port: u16, - wireguard_port: u16, -) -> bool { - match flow_key.protocol { - IpProto::Tcp | IpProto::Udp => { - let src = flow_key.src_port; - let dst = flow_key.dst_port; - src == vxlan_port - || dst == vxlan_port - || src == geneve_port - || dst == geneve_port - || src == wireguard_port - || dst == wireguard_port - } - IpProto::Gre => true, - _ => false, - } -} - -/// Parse packet data where eBPF has already consumed the outer headers. -/// For tunnels, `data` starts at the inner Ethernet header. -pub fn parse_packet_from_offset( - data: &[u8], - _parsed_offset: u16, -) -> Result { - parse_packet_deep(data) -} - -/// Parse packet from raw bytes (up to 256 bytes captured) -pub fn parse_packet_deep(data: &[u8]) -> Result { - let mut offset = 0; - - let (src_mac, dst_mac, ether_type) = parse_ethernet(data, &mut offset)?; - - let l2_meta = L2Metadata { - src_mac, - dst_mac, - ether_type, - }; - - let (outer_src_ip, outer_dst_ip, outer_protocol, ip_meta) = match ether_type { - EtherType::Ipv4 => parse_ipv4(data, &mut offset)?, - EtherType::Ipv6 => parse_ipv6(data, &mut offset)?, - _ => return Err(ParseError::UnsupportedEtherType), - }; - - match outer_protocol { - IpProto::Udp => { - let (outer_src_port, outer_dst_port) = parse_l4_ports(data, offset)?; - - if outer_dst_port == 4789 || outer_src_port == 4789 { - return parse_vxlan( - data, - &mut offset, - outer_src_ip, - outer_dst_ip, - outer_src_port, - outer_dst_port, - src_mac, - dst_mac, - ); - } else if outer_dst_port == 6081 || outer_src_port == 6081 { - return parse_geneve( - data, - &mut offset, - outer_src_ip, - outer_dst_ip, - outer_src_port, - outer_dst_port, - src_mac, - dst_mac, - ); - } - - Ok(ParsedPacket::Direct { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port: outer_src_port, - dst_port: outer_dst_port, - protocol: IpProto::Udp, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - l2_metadata: l2_meta, - ip_metadata: ip_meta, - }) - } - IpProto::Tcp => { - let (src_port, dst_port) = parse_l4_ports(data, offset)?; - Ok(ParsedPacket::Direct { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port, - dst_port, - protocol: IpProto::Tcp, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - l2_metadata: l2_meta, - ip_metadata: ip_meta, - }) - } - IpProto::Icmp => { - let (type_code, _) = parse_icmp_type_code(data, offset)?; - Ok(ParsedPacket::Direct { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port: type_code, - dst_port: 0, - protocol: outer_protocol, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - l2_metadata: l2_meta, - ip_metadata: ip_meta, - }) - } - IpProto::Gre => { - // TODO: GRE parsing - Err(ParseError::UnsupportedProtocol) - } - _ => Ok(ParsedPacket::Direct { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port: 0, - dst_port: 0, - protocol: outer_protocol, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - l2_metadata: l2_meta, - ip_metadata: ip_meta, - }), - } -} - -fn parse_ethernet( - data: &[u8], - offset: &mut usize, -) -> Result<([u8; 6], [u8; 6], EtherType), ParseError> { - if data.len() < *offset + 14 { - return Err(ParseError::TooShort); - } - - let mut dst_mac = [0u8; 6]; - let mut src_mac = [0u8; 6]; - dst_mac.copy_from_slice(&data[*offset..*offset + 6]); - src_mac.copy_from_slice(&data[*offset + 6..*offset + 12]); - - let ether_type_raw = u16::from_be_bytes([data[*offset + 12], data[*offset + 13]]); - let ether_type = EtherType::try_from(ether_type_raw).unwrap_or(EtherType::Reserved); - - *offset += 14; - Ok((src_mac, dst_mac, ether_type)) -} - -fn parse_ipv4( - data: &[u8], - offset: &mut usize, -) -> Result<(IpAddr, IpAddr, IpProto, IpMetadata), ParseError> { - if data.len() < *offset + 20 { - return Err(ParseError::TooShort); - } - - let ihl = ((data[*offset] & 0x0F) * 4) as usize; - let protocol_num = data[*offset + 9]; - let protocol = match protocol_num { - 6 => IpProto::Tcp, - 17 => IpProto::Udp, - 1 => IpProto::Icmp, - 47 => IpProto::Gre, - _ => IpProto::Reserved, - }; - let src_ip = Ipv4Addr::new( - data[*offset + 12], - data[*offset + 13], - data[*offset + 14], - data[*offset + 15], - ); - let dst_ip = Ipv4Addr::new( - data[*offset + 16], - data[*offset + 17], - data[*offset + 18], - data[*offset + 19], - ); - - let dscp = (data[*offset + 1] >> 2) & 0x3F; - let ecn = data[*offset + 1] & 0x03; - let ttl = data[*offset + 8]; - - *offset += ihl; - - Ok(( - IpAddr::V4(src_ip), - IpAddr::V4(dst_ip), - protocol, - IpMetadata { - dscp, - ecn, - ttl, - ..Default::default() - }, - )) -} - -fn parse_ipv6( - data: &[u8], - offset: &mut usize, -) -> Result<(IpAddr, IpAddr, IpProto, IpMetadata), ParseError> { - if data.len() < *offset + 40 { - return Err(ParseError::TooShort); - } - - let next_hdr_num = data[*offset + 6]; - let next_hdr = match next_hdr_num { - 6 => IpProto::Tcp, - 17 => IpProto::Udp, - 58 => IpProto::Icmp, // ICMPv6 uses protocol 58 - 47 => IpProto::Gre, - _ => IpProto::Reserved, - }; - - let mut src_ip_bytes = [0u8; 16]; - let mut dst_ip_bytes = [0u8; 16]; - src_ip_bytes.copy_from_slice(&data[*offset + 8..*offset + 24]); - dst_ip_bytes.copy_from_slice(&data[*offset + 24..*offset + 40]); - - let src_ip = Ipv6Addr::from(src_ip_bytes); - let dst_ip = Ipv6Addr::from(dst_ip_bytes); - - let dscp = (data[*offset] & 0x0F) << 2 | (data[*offset + 1] >> 6); - let ecn = (data[*offset + 1] >> 4) & 0x03; - let ttl = data[*offset + 7]; - let flow_label = u32::from_be_bytes([ - 0, - data[*offset + 1] & 0x0F, - data[*offset + 2], - data[*offset + 3], - ]) & 0xFFFFF; - - *offset += 40; - - Ok(( - IpAddr::V6(src_ip), - IpAddr::V6(dst_ip), - next_hdr, - IpMetadata { - dscp, - ecn, - ttl, - flow_label: Some(flow_label), - ..Default::default() - }, - )) -} - -fn parse_l4_ports(data: &[u8], offset: usize) -> Result<(u16, u16), ParseError> { - if data.len() < offset + 4 { - return Err(ParseError::TooShort); - } - let src_port = u16::from_be_bytes([data[offset], data[offset + 1]]); - let dst_port = u16::from_be_bytes([data[offset + 2], data[offset + 3]]); - Ok((src_port, dst_port)) -} - -fn parse_icmp_type_code(data: &[u8], offset: usize) -> Result<(u16, u16), ParseError> { - if data.len() < offset + 2 { - return Err(ParseError::TooShort); - } - let icmp_type = data[offset] as u16; - let icmp_code = data[offset + 1] as u16; - // Encode as port field: (type << 8 | code) - let type_code_port = (icmp_type << 8) | icmp_code; - Ok((type_code_port, 0)) -} - -#[allow(clippy::too_many_arguments)] -fn parse_vxlan( - data: &[u8], - offset: &mut usize, - outer_src_ip: IpAddr, - outer_dst_ip: IpAddr, - outer_src_port: u16, - outer_dst_port: u16, - outer_src_mac: [u8; 6], - outer_dst_mac: [u8; 6], -) -> Result { - *offset += 8; // skip UDP header - - // VXLAN header (8 bytes): flags(1) + reserved(3) + VNI(3) + reserved(1) - if data.len() < *offset + 8 { - return Err(ParseError::TooShort); - } - - let vni = u32::from_be_bytes([0, data[*offset + 4], data[*offset + 5], data[*offset + 6]]); - *offset += 8; - - let (inner_src_mac, inner_dst_mac, inner_ether_type) = parse_ethernet(data, offset)?; - - let (inner_src_ip, inner_dst_ip, inner_protocol, _) = match inner_ether_type { - EtherType::Ipv4 => parse_ipv4(data, offset)?, - EtherType::Ipv6 => parse_ipv6(data, offset)?, - _ => return Err(ParseError::UnsupportedEtherType), - }; - - let (inner_src_port, inner_dst_port) = match inner_protocol { - IpProto::Tcp | IpProto::Udp => parse_l4_ports(data, *offset)?, - IpProto::Icmp => parse_icmp_type_code(data, *offset)?, - _ => (0, 0), - }; - - Ok(ParsedPacket::Tunneled { - outer: OuterHeaders { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port: outer_src_port, - dst_port: outer_dst_port, - protocol: IpProto::Udp, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - src_mac: outer_src_mac, - dst_mac: outer_dst_mac, - }, - inner: InnerHeaders { - src_ip: inner_src_ip, - dst_ip: inner_dst_ip, - src_port: inner_src_port, - dst_port: inner_dst_port, - protocol: inner_protocol, - src_mac: inner_src_mac, - dst_mac: inner_dst_mac, - }, - tunnel_info: TunnelInfo { - tunnel_type: TunnelType::Vxlan, - vni, - }, - }) -} - -#[allow(clippy::too_many_arguments)] -fn parse_geneve( - data: &[u8], - offset: &mut usize, - outer_src_ip: IpAddr, - outer_dst_ip: IpAddr, - outer_src_port: u16, - outer_dst_port: u16, - outer_src_mac: [u8; 6], - outer_dst_mac: [u8; 6], -) -> Result { - *offset += 8; // skip UDP header - - // Geneve header (8+ bytes): ver/opt_len(1) + flags(1) + protocol(2) + VNI(3) + reserved(1) - if data.len() < *offset + 8 { - return Err(ParseError::TooShort); - } - - let opt_len = (data[*offset] & 0x3F) as usize * 4; // options length in bytes - let vni = u32::from_be_bytes([0, data[*offset + 4], data[*offset + 5], data[*offset + 6]]); - *offset += 8 + opt_len; - - let (inner_src_mac, inner_dst_mac, inner_ether_type) = parse_ethernet(data, offset)?; - - let (inner_src_ip, inner_dst_ip, inner_protocol, _) = match inner_ether_type { - EtherType::Ipv4 => parse_ipv4(data, offset)?, - EtherType::Ipv6 => parse_ipv6(data, offset)?, - _ => return Err(ParseError::UnsupportedEtherType), - }; - - let (inner_src_port, inner_dst_port) = match inner_protocol { - IpProto::Tcp | IpProto::Udp => parse_l4_ports(data, *offset)?, - IpProto::Icmp => parse_icmp_type_code(data, *offset)?, - _ => (0, 0), - }; - - Ok(ParsedPacket::Tunneled { - outer: OuterHeaders { - five_tuple: FiveTuple { - src_ip: outer_src_ip, - dst_ip: outer_dst_ip, - src_port: outer_src_port, - dst_port: outer_dst_port, - protocol: IpProto::Udp, - ip_version: if outer_src_ip.is_ipv4() { 4 } else { 6 }, - }, - src_mac: outer_src_mac, - dst_mac: outer_dst_mac, - }, - inner: InnerHeaders { - src_ip: inner_src_ip, - dst_ip: inner_dst_ip, - src_port: inner_src_port, - dst_port: inner_dst_port, - protocol: inner_protocol, - src_mac: inner_src_mac, - dst_mac: inner_dst_mac, - }, - tunnel_info: TunnelInfo { - tunnel_type: TunnelType::Geneve, - vni, - }, - }) -} diff --git a/mermin/src/packet/types.rs b/mermin/src/packet/types.rs deleted file mode 100644 index 40e2045a3..000000000 --- a/mermin/src/packet/types.rs +++ /dev/null @@ -1,103 +0,0 @@ -//! Packet parsing types for deep inspection of captured packet data. -//! -//! These types support extracting innermost 5-tuples from tunneled/encapsulated traffic -//! and identifying tunnel metadata (VXLAN VNI, Geneve, GRE, etc.) - -use std::net::IpAddr; - -use mermin_common::{TunnelType, eth::EtherType, ip::IpProto}; - -#[derive(Debug, Clone)] -#[allow(dead_code)] -pub enum ParsedPacket { - /// No tunnel/encapsulation detected - Direct { - five_tuple: FiveTuple, - l2_metadata: L2Metadata, - ip_metadata: IpMetadata, - }, - /// Tunneled/encapsulated traffic (VXLAN, Geneve, GRE, etc.) - Tunneled { - outer: OuterHeaders, - inner: InnerHeaders, - tunnel_info: TunnelInfo, - }, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct FiveTuple { - pub src_ip: IpAddr, - pub dst_ip: IpAddr, - pub src_port: u16, - pub dst_port: u16, - pub protocol: IpProto, - pub ip_version: u8, -} - -#[derive(Debug, Clone, Copy)] -#[allow(dead_code)] -pub struct TunnelInfo { - pub tunnel_type: TunnelType, - pub vni: u32, // VNI for VXLAN/Geneve, Key for GRE -} - -#[derive(Debug, Clone, Copy)] -#[allow(dead_code)] -pub struct OuterHeaders { - pub five_tuple: FiveTuple, - pub src_mac: [u8; 6], - pub dst_mac: [u8; 6], -} - -#[derive(Debug, Clone, Copy)] -#[allow(dead_code)] -pub struct InnerHeaders { - pub src_ip: IpAddr, - pub dst_ip: IpAddr, - pub src_port: u16, - pub dst_port: u16, - pub protocol: IpProto, - pub src_mac: [u8; 6], - pub dst_mac: [u8; 6], -} - -#[derive(Debug, Clone, Copy)] -#[allow(dead_code)] -pub struct L2Metadata { - pub src_mac: [u8; 6], - pub dst_mac: [u8; 6], - pub ether_type: EtherType, -} - -#[derive(Debug, Clone, Copy, Default)] -#[allow(dead_code)] -pub struct IpMetadata { - pub dscp: u8, - pub ecn: u8, - pub ttl: u8, - pub flow_label: Option, // IPv6 only - pub ipsec_ah_spi: Option, - pub ipsec_esp_spi: Option, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -#[allow(dead_code)] -pub enum ParseError { - TooShort, - InvalidHeader, - UnsupportedEtherType, - UnsupportedProtocol, -} - -impl std::fmt::Display for ParseError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - ParseError::TooShort => write!(f, "packet too short"), - ParseError::InvalidHeader => write!(f, "invalid header"), - ParseError::UnsupportedEtherType => write!(f, "unsupported ether type"), - ParseError::UnsupportedProtocol => write!(f, "unsupported protocol"), - } - } -} - -impl std::error::Error for ParseError {} diff --git a/mermin/src/runtime/conf.rs b/mermin/src/runtime/conf.rs index d0da893ba..f5dd87baf 100644 --- a/mermin/src/runtime/conf.rs +++ b/mermin/src/runtime/conf.rs @@ -806,10 +806,10 @@ pub struct FlowCapture { /// Uses `BPF_F_NO_PREALLOC`: memory grows on demand as flows are tracked, up to this /// limit. Once full, new flow insert attempts are dropped. /// - /// Per-entry cost: ~270 bytes (FlowStats: 192 + FlowKey: 40 aligned + htab_elem: ~38). - /// Maximum kernel memory when fully populated: capacity × ~270 bytes. + /// Per-entry cost: ~318 bytes (FlowStats: 240 + FlowKey: 40 aligned + htab_elem: ~38). + /// Maximum kernel memory when fully populated: capacity × ~318 bytes. /// - /// Default (16,384) targets typical deployments: ~4.4 MB maximum kernel memory. + /// Default (16,384) targets typical deployments: ~5.2 MB maximum kernel memory. /// This covers ~3,200 flow records/s at the default 5s poll interval. /// /// Formula: concurrent_flows = flows_per_sec × avg_flow_lifetime_seconds diff --git a/mermin/src/span/direction.rs b/mermin/src/span/direction.rs index 19eb94e9a..19df8cde2 100644 --- a/mermin/src/span/direction.rs +++ b/mermin/src/span/direction.rs @@ -423,6 +423,13 @@ mod tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, } } diff --git a/mermin/src/span/producer.rs b/mermin/src/span/producer.rs index 6bc1b9d21..eafc3ba97 100644 --- a/mermin/src/span/producer.rs +++ b/mermin/src/span/producer.rs @@ -1,5 +1,6 @@ use std::{ collections::HashMap, + net::{Ipv4Addr, Ipv6Addr}, os::fd::AsRawFd, sync::Arc, time::{Duration, UNIX_EPOCH}, @@ -10,7 +11,7 @@ use aya::maps::{HashMap as EbpfHashMap, RingBuf}; use dashmap::DashMap; use fxhash::FxBuildHasher; use mermin_common::{ - FlowEvent, FlowKey, FlowStats, ListeningPortKey, MapUnit, + FlowEvent, FlowKey, FlowStats, IpVersion, ListeningPortKey, MapUnit, TunnelType, eth::EtherType, ip::{IpDscp, IpEcn, IpProto}, tcp::{TCP_FLAG_FIN, TCP_FLAG_RST}, @@ -36,10 +37,6 @@ use crate::{ ProcessingStage, }, }, - packet::{ - parser::{is_tunnel, parse_packet_from_offset}, - types::ParsedPacket, - }, runtime::{conf::Conf, memory::ShrinkPolicy}, span::{ community_id::CommunityIdGenerator, @@ -147,9 +144,6 @@ pub struct FlowSpanProducer { workers: usize, flow_store_poll_interval: Duration, filter: Option>, - vxlan_port: u16, - geneve_port: u16, - wireguard_port: u16, flow_events_ringbuf: RingBuf, components: Arc, community_id_generator: CommunityIdGenerator, @@ -227,9 +221,6 @@ impl FlowSpanProducer { workers, flow_store_poll_interval: conf.pipeline.flow_producer.flow_store_poll_interval, filter, - vxlan_port: conf.parser.vxlan_port, - geneve_port: conf.parser.geneve_port, - wireguard_port: conf.parser.wireguard_port, flow_events_ringbuf, components, community_id_generator, @@ -271,9 +262,6 @@ impl FlowSpanProducer { Arc::clone(&components.flow_stats_map), worker_rx, self.filter.clone(), - self.vxlan_port, - self.geneve_port, - self.wireguard_port, Arc::clone(&self.direction_inferrer), ); @@ -620,9 +608,6 @@ pub struct FlowWorker { flow_stats_map: Arc>>, flow_event_rx: mpsc::Receiver, filter: Option>, - vxlan_port: u16, - geneve_port: u16, - wireguard_port: u16, direction_inferrer: Arc, metrics: FlowWorkerMetrics, } @@ -640,9 +625,6 @@ impl FlowWorker { flow_stats_map: Arc>>, flow_event_rx: mpsc::Receiver, filter: Option>, - vxlan_port: u16, - geneve_port: u16, - wireguard_port: u16, direction_inferrer: Arc, ) -> Self { Self { @@ -661,9 +643,6 @@ impl FlowWorker { flow_stats_map, flow_event_rx, filter, - vxlan_port, - geneve_port, - wireguard_port, direction_inferrer, metrics: FlowWorkerMetrics::new(), } @@ -692,35 +671,10 @@ impl FlowWorker { // TODO: Review starting here - /// Process a new flow event from eBPF. - /// This is the core of the event-driven architecture: - /// 1. Check if tunneled (fast path vs slow path) - /// 2. For direct traffic: Use FlowStats directly (no parsing!) - /// 3. For tunneled traffic: Deep parse packet_data to extract innermost 5-tuple + /// Process a new flow event from eBPF. VXLAN/Geneve are decapsulated in eBPF, so every + /// event already carries the flow's own 5-tuple and needs no userspace parsing. async fn process_new_flow(&self, event: FlowEvent) -> Result<(), Error> { - if !is_tunnel( - &event.flow_key, - self.vxlan_port, - self.geneve_port, - self.wireguard_port, - ) { - return self.create_direct_flow(event).await; - } - - // TODO: Skip tunneled flow processing for now - untested and needs refactoring - debug!( - event.name = "tunneled_flow.skipped", - worker.id = self.worker_id, - protocol = ?event.flow_key.protocol, - src_ip = ?event.flow_key.src_ip, - dst_ip = ?event.flow_key.dst_ip, - "skipping tunneled flow processing - not yet implemented" - ); - - self.remove_ebpf_map_entry(&self.flow_stats_map, &event.flow_key) - .await; - - Ok(()) + self.create_direct_flow(event).await } async fn remove_ebpf_map_entry( @@ -838,39 +792,6 @@ impl FlowWorker { } } - /// Slow path for tunneled traffic. - /// Parses packet_data to extract innermost 5-tuple and tunnel metadata. - #[allow(dead_code)] - async fn create_tunneled_flow(&self, event: FlowEvent) -> Result<(), Error> { - // Parse only the UNPARSED portion (inner headers for tunnels) - let unparsed_data = &event.packet_data[..]; - let parsed = parse_packet_from_offset(unparsed_data, event.parsed_offset) - .map_err(|_| Error::UnknownIpAddrType)?; - - // Extract innermost 5-tuple for Community ID - match parsed { - ParsedPacket::Tunneled { inner, .. } => { - // TODO: Generate Community ID from innermost 5-tuple - // TODO: Map eBPF FlowKey → Community ID - // TODO: Create FlowSpan with tunnel metadata - - warn!( - event.name = "tunneled_flow.not_implemented", - worker.id = self.worker_id, - inner.src_ip = %inner.src_ip, - inner.dst_ip = %inner.dst_ip, - "tunneled flow processing not yet fully implemented" - ); - } - ParsedPacket::Direct { .. } => { - // This shouldn't happen - we already checked is_likely_tunnel - warn!("expected tunneled packet but got plain, treating as error"); - } - } - - Ok(()) - } - /// Determine if a flow should be processed based on filtering rules. fn should_process_flow(&self, flow_key: &FlowKey, stats: &FlowStats) -> bool { if let Some(filter) = &self.filter { @@ -921,7 +842,7 @@ impl FlowWorker { let flow_direction = FlowDirection::from(span_kind.clone()); - let span = FlowSpan { + let mut span = FlowSpan { trace_id: Some(self.id_generator.new_trace_id()), start_time: UNIX_EPOCH + Duration::from_nanos(start_time_nanos), end_time: UNIX_EPOCH + Duration::from_nanos(end_time_nanos), @@ -1062,6 +983,7 @@ impl FlowWorker { timeout_duration: Duration::from_secs(0), }; + apply_tunnel_attrs(&mut span, stats); self.insert_flow(community_id, span, timeout); let interface_name = iface_name.as_deref().unwrap_or("unknown"); @@ -1516,6 +1438,34 @@ fn icmp_code_name( /// Hash a string for flow partitioning across pollers /// Record a flow span by reading from eBPF map, updating counters, and sending to exporter. /// Returns true if the flow should continue being tracked, false if export channel is closed. +/// Copies the outer VXLAN/Geneve headers eBPF recorded into the span's `tunnel.*` attributes. +/// Runs on every record because a flow first seen on a veth can be decapsulated later. +fn apply_tunnel_attrs(flow_span: &mut FlowSpan, stats: &FlowStats) { + if stats.tunnel_type == TunnelType::None || flow_span.attributes.tunnel_type.is_some() { + return; + } + let addr = |ip: [u8; 16]| -> Option> { + match stats.tunnel_ip_version { + IpVersion::V4 => Some(Ipv4Addr::new(ip[0], ip[1], ip[2], ip[3]).to_string().into()), + IpVersion::V6 => Some(Ipv6Addr::from(ip).to_string().into()), + IpVersion::Unknown => None, + } + }; + let attrs = flow_span.attrs_mut(); + attrs.tunnel_type = Some(stats.tunnel_type); + attrs.tunnel_id = Some(stats.tunnel_id); + attrs.tunnel_network_type = Some(match stats.tunnel_ip_version { + IpVersion::V6 => EtherType::Ipv6, + _ => EtherType::Ipv4, + }); + attrs.tunnel_network_transport = Some(IpProto::Udp); + attrs.tunnel_network_interface_mac = attrs.network_interface_mac.clone(); + attrs.tunnel_source_address = addr(stats.tunnel_src_ip); + attrs.tunnel_destination_address = addr(stats.tunnel_dst_ip); + attrs.tunnel_source_port = Some(stats.tunnel_src_port); + attrs.tunnel_destination_port = Some(stats.tunnel_dst_port); +} + async fn record_flow( community_id: &str, flow_store: &FlowStore, @@ -1709,6 +1659,7 @@ async fn record_flow( }; let flow_span = &mut entry_ref.flow_span; + apply_tunnel_attrs(flow_span, &stats); let is_tcp = stats.protocol == IpProto::Tcp; let attrs = flow_span.attrs_mut(); attrs.flow_connection_state = is_tcp.then_some(stats.tcp_state); @@ -2317,6 +2268,13 @@ mod tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: TunnelType::None, + tunnel_ip_version: IpVersion::Unknown, } } @@ -2366,9 +2324,6 @@ mod tests { flow_stats_map, flow_event_rx, filter: None, - vxlan_port: 4789, - geneve_port: 6081, - wireguard_port: 51820, direction_inferrer, metrics: FlowWorkerMetrics::new(), } diff --git a/mermin/tests/e2e/cni/flannel-vxlan.hcl b/mermin/tests/e2e/cni/flannel-vxlan.hcl new file mode 100644 index 000000000..50cc9a979 --- /dev/null +++ b/mermin/tests/e2e/cni/flannel-vxlan.hcl @@ -0,0 +1,41 @@ +export "traces" { + stdout = { + format = "text_indent" # text, text_indent(*new), json, json_indent + } + + # otlp = { + # endpoint = "http://otel-collector:4317" # Uncomment to export to an OTel Demo Collector + + # # Authentication config + # # auth = { + # # basic = { + # # user = "USERNAME" + # # pass = "PASSWORD" + # # } + # # } + + # # TLS config + # # tls = { + # # insecure_skip_verify = false # Skip verifying the OTLP receiver certificate + # # ca_cert = "/etc/certs/ca.crt" # Path to the receiver certificate Certificate Authority + # # client_cert = "/etc/certs/cert.crt" # Client TLS certificate (mTLS) + # # client_key = "/etc/certs/cert.key" # Client TLS key (mTLS) + # # } + # } +} + +# Test configuration optimized for fast CI runs +# Reduce flow export intervals for faster test feedback +span { + max_record_interval = "10s" # Export active flows every 10s (default: 60s) + icmp_timeout = "5s" # ICMP flows timeout after 5s (default: 10s) +} + +# Watch only the node uplink so inter-node pod traffic is seen VXLAN-encapsulated. +discovery "instrument" { + interfaces = ["eth0"] +} + +parser { + vxlan_port = 8472 # Linux/Flannel VXLAN port +} diff --git a/mermin/tests/e2e/cni/test.sh b/mermin/tests/e2e/cni/test.sh index eceb5742d..7f8e1db8a 100755 --- a/mermin/tests/e2e/cni/test.sh +++ b/mermin/tests/e2e/cni/test.sh @@ -28,9 +28,20 @@ verify_deployment() { HELM_NAMESPACE="$NAMESPACE" } +# Pin a pod to a node so the pinger and receiver sit on different nodes and the +# traffic crosses the CNI overlay. +run_on_node() { + local name=$1 node=$2 + shift 2 + kubectl -n "${NAMESPACE}" run "$name" --grace-period=1 --image=alpine \ + --overrides="{\"spec\":{\"nodeName\":\"${node}\"}}" --command -- "$@" +} + verify_agent_logs() { - echo "Creating flows generator (pinger)..." - kubectl -n "${NAMESPACE}" run ping-receiver --grace-period=1 --image=alpine --command -- sleep 3600 + local nodes + read -r -a nodes <<<"$(kubectl get nodes -o 'jsonpath={.items[*].metadata.name}')" + echo "Creating flows generator (pinger) across nodes: ${nodes[0]} -> ${nodes[1]}..." + run_on_node ping-receiver "${nodes[0]}" sleep 3600 echo "Waiting for ping-receiver pod to be ready..." kubectl wait --for=condition=ready pod/ping-receiver -n "${NAMESPACE}" --timeout=60s || { echo "ERROR: ping-receiver pod failed to become ready" @@ -54,7 +65,7 @@ verify_agent_logs() { return 1 fi - kubectl -n "${NAMESPACE}" run pinger --grace-period=1 --image=alpine --command -- ping "${ping_receiver_ip}" + run_on_node pinger "${nodes[1]}" ping "${ping_receiver_ip}" echo "Waiting for pinger pod to be ready..." kubectl wait --for=condition=ready pod/pinger -n "${NAMESPACE}" --timeout=60s || { echo "WARNING: pinger pod failed to become ready. Continuing..." @@ -65,6 +76,9 @@ verify_agent_logs() { echo "Verifying mermin agent logs are enriching data..." export NAMESPACE RELEASE_NAME + # With TUNNEL_TYPE set, the enriched span must also carry the decapsulated tunnel metadata. + local pattern='(source\.k8s\.pod\.name.*String\((Owned|Static|RefCounted)\("(pinger|ping-receiver)"\)|destination\.k8s\.pod\.name.*String\((Owned|Static|RefCounted)\("(pinger|ping-receiver)"\))' + local tunnel_pattern="tunnel\\.type.*\"${TUNNEL_TYPE:-}\"" local pods=() while IFS= read -r pod_name; do if [[ -n "$pod_name" ]]; then @@ -81,7 +95,9 @@ verify_agent_logs() { local counter=0 while [ $counter -lt 60 ]; do # examples: https://regex101.com/r/rYbX7m/1 - if kubectl logs -n "${NAMESPACE}" "$pod" --tail=10000 2>/dev/null | grep --color=never -E '(source\.k8s\.pod\.name.*String\((Owned|Static|RefCounted)\("(pinger|ping-receiver)"\)|destination\.k8s\.pod\.name.*String\((Owned|Static|RefCounted)\("(pinger|ping-receiver)"\))' >/dev/null; then + local logs + logs=$(kubectl logs -n "${NAMESPACE}" "$pod" --tail=10000 2>/dev/null || true) + if grep -qE "$pattern" <<<"$logs" && { [[ -z "${TUNNEL_TYPE:-}" ]] || grep -qE "$tunnel_pattern" <<<"$logs"; }; then exit 0 fi counter=$((counter + 1)) @@ -124,8 +140,14 @@ echo "===============================" echo "Testing with CNI: $CNI" echo "===============================" +# Flannel uses VXLAN, so watch the uplink and require the decapsulated span to carry tunnel.* attributes. +if [[ "$CNI" == "flannel" ]]; then + MERMIN_CONFIG_PATH=./mermin/tests/e2e/cni/flannel-vxlan.hcl + TUNNEL_TYPE=vxlan +fi + # Export variables needed by setup.sh -export CLUSTER_NAME RELEASE_NAME NAMESPACE DOCKER_REPOSITORY DOCKER_IMAGE_TAG VALUES_FILE CNI +export CLUSTER_NAME RELEASE_NAME NAMESPACE DOCKER_REPOSITORY DOCKER_IMAGE_TAG VALUES_FILE CNI MERMIN_CONFIG_PATH bash ./mermin/tests/e2e/common/setup.sh verify_deployment || { dump_debug_info; exit 1; } diff --git a/mermin/tests/e2e/common/setup.sh b/mermin/tests/e2e/common/setup.sh index 4d70e6eff..c3ba22523 100644 --- a/mermin/tests/e2e/common/setup.sh +++ b/mermin/tests/e2e/common/setup.sh @@ -39,7 +39,11 @@ install_cilium() { sudo tar -C /usr/local/bin -xzvf "cilium-${OS}-${ARCH}.tar.gz" rm "cilium-${OS}-${ARCH}.tar.gz"{,.sha256sum} fi - cilium install --wait + cilium install --wait \ + --set image.pullPolicy=IfNotPresent \ + --set ipam.mode=kubernetes + kubectl rollout status daemonset/cilium -n kube-system --timeout=5m + kubectl rollout status daemonset/cilium-envoy -n kube-system --timeout=5m } diff --git a/mermin/tests/integration.rs b/mermin/tests/integration.rs index 1d944cd65..aa784a4b6 100644 --- a/mermin/tests/integration.rs +++ b/mermin/tests/integration.rs @@ -334,6 +334,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; handler.handle_flow_event(flow_key, stats).await; @@ -400,6 +407,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; handler @@ -462,6 +476,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; // Update stats (simulating eBPF map update) @@ -533,6 +554,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; handler.handle_flow_event(flow_key, stats_t1).await; @@ -585,6 +613,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; handler.handle_flow_event(flow_key, stats_t2).await; @@ -673,6 +708,13 @@ mod event_driven_tests { reverse_icmp_code: 0, pid: 0, comm: [0u8; 16], + tunnel_src_ip: [0; 16], + tunnel_dst_ip: [0; 16], + tunnel_id: 0, + tunnel_src_port: 0, + tunnel_dst_port: 0, + tunnel_type: mermin_common::TunnelType::None, + tunnel_ip_version: mermin_common::IpVersion::Unknown, }; handler_clone.handle_flow_event(key, stats).await;