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;