mirror of
https://github.com/ParrotXray/Mantis.git
synced 2026-08-24 19:00:27 +09:00
Fill TCP flags/window/header/payload in eBPF parser; migrate stats maps to per-CPU
- common/src/ebpf/parsing.rs: refactor parse_tcp_port/parse_udp_port into parse_tcp/parse_udp returning (src_port, dst_port, TcpFlags, header_length, window_size); compute payload_length from tot_len/payload_len; write all previously-missing fields (tcp_flags, tcp_window_size, header_length, payload_length, is_forward) into the Event; remove redundant nested unsafe blocks inside unsafe fn bodies - ingress-ebpf/src/action/statistics.rs, egress-ebpf/src/action/statistics.rs: replace LruHashMap with LruPerCpuHashMap (BPF_MAP_TYPE_LRU_PERCPU_HASH) to eliminate multi-CPU races on per-flow counter updates - mantis/src/core/ebpf/statistics.rs: replace AyaHashMap with PerCpuHashMap; aggregate per-CPU values in get_map() (sum bytes/packets, max last_seen) and use max last_seen for expiry check in cleanup() Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0138PxtKH73hqxv7h1oaoSdS
This commit is contained in:
Claude
parent
3cd23ae153
commit
77c427c298
@ -5,7 +5,7 @@ use network_types::tcp::TcpHdr;
|
||||
use network_types::udp::UdpHdr;
|
||||
|
||||
use crate::define::offset::*;
|
||||
use crate::model::event::{Event, IPv4Event, IPv6Event};
|
||||
use crate::model::event::{Event, IPv4Event, IPv6Event, TcpFlags};
|
||||
|
||||
pub fn parse_packet(start: usize, end: usize, target: *mut Event) -> Result<(), ()> {
|
||||
unsafe {
|
||||
@ -24,112 +24,85 @@ pub fn parse_packet(start: usize, end: usize, target: *mut Event) -> Result<(),
|
||||
|
||||
#[inline(always)]
|
||||
unsafe fn parse_ipv4_packet(start: usize, end: usize, target: *mut Event) -> Result<(), ()> {
|
||||
unsafe {
|
||||
if start + IPV4_HEADER_END > end {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
let ipv4 = &*((start + IPV4_HEADER_START) as *const Ipv4Hdr);
|
||||
|
||||
// ihl() returns actual IP header length in bytes (20–60); use it so IP Options
|
||||
// don't shift the L4 header offset and corrupt port extraction.
|
||||
let ip_hdr_len = ipv4.ihl() as usize;
|
||||
if ip_hdr_len < core::mem::size_of::<Ipv4Hdr>() {
|
||||
return Err(());
|
||||
}
|
||||
let l4_start = ETHER_HEADER_END + ip_hdr_len;
|
||||
|
||||
let (source_port, destination_port) = match ipv4.proto {
|
||||
IpProto::Tcp => parse_tcp_port(start, end, l4_start)?,
|
||||
IpProto::Udp => parse_udp_port(start, end, l4_start)?,
|
||||
_ => return Err(()),
|
||||
};
|
||||
|
||||
*(target as *mut u32) = 0;
|
||||
|
||||
let ipv4_data_ptr = (target as *mut u8).add(16);
|
||||
|
||||
core::ptr::write(ipv4_data_ptr as *mut IpProto, ipv4.proto);
|
||||
core::ptr::copy_nonoverlapping(
|
||||
ipv4.src_addr.as_ptr(),
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, src_ip)),
|
||||
4,
|
||||
);
|
||||
core::ptr::copy_nonoverlapping(
|
||||
ipv4.dst_addr.as_ptr(),
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, dst_ip)),
|
||||
4,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, src_port)) as *mut u16,
|
||||
source_port,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, dst_port)) as *mut u16,
|
||||
destination_port,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, packet_length)) as *mut u32,
|
||||
(end - start) as u32,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv4_data_ptr.add(core::mem::offset_of!(IPv4Event, timestamp_us)) as *mut u64,
|
||||
bpf_ktime_get_ns(),
|
||||
);
|
||||
|
||||
Ok(())
|
||||
if start + IPV4_HEADER_END > end {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
let ipv4 = &*((start + IPV4_HEADER_START) as *const Ipv4Hdr);
|
||||
|
||||
// ihl() returns actual IP header length in bytes (20–60); use it so IP Options
|
||||
// don't shift the L4 header offset and corrupt port extraction.
|
||||
let ip_hdr_len = ipv4.ihl() as usize;
|
||||
if ip_hdr_len < core::mem::size_of::<Ipv4Hdr>() {
|
||||
return Err(());
|
||||
}
|
||||
let l4_start = ETHER_HEADER_END + ip_hdr_len;
|
||||
|
||||
let (src_port, dst_port, tcp_flags, l4_hdr_len, window_size) = match ipv4.proto {
|
||||
IpProto::Tcp => parse_tcp(start, end, l4_start)?,
|
||||
IpProto::Udp => parse_udp(start, end, l4_start)?,
|
||||
_ => return Err(()),
|
||||
};
|
||||
|
||||
let payload_length = (ipv4.tot_len() as u32)
|
||||
.saturating_sub(ip_hdr_len as u32)
|
||||
.saturating_sub(l4_hdr_len as u32);
|
||||
|
||||
*(target as *mut u32) = 0;
|
||||
|
||||
let p = (target as *mut u8).add(16);
|
||||
|
||||
core::ptr::write(p as *mut IpProto, ipv4.proto);
|
||||
core::ptr::copy_nonoverlapping(ipv4.src_addr.as_ptr(), p.add(core::mem::offset_of!(IPv4Event, src_ip)), 4);
|
||||
core::ptr::copy_nonoverlapping(ipv4.dst_addr.as_ptr(), p.add(core::mem::offset_of!(IPv4Event, dst_ip)), 4);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, src_port)) as *mut u16, src_port);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, dst_port)) as *mut u16, dst_port);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, packet_length)) as *mut u32, (end - start) as u32);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, payload_length)) as *mut u32, payload_length);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, header_length)) as *mut u16, l4_hdr_len);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, timestamp_us)) as *mut u64, bpf_ktime_get_ns());
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, tcp_flags)) as *mut TcpFlags, tcp_flags);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, tcp_window_size)) as *mut u16, window_size);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv4Event, is_forward)) as *mut bool, false);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
unsafe fn parse_ipv6_packet(start: usize, end: usize, target: *mut Event) -> Result<(), ()> {
|
||||
unsafe {
|
||||
if start + IPV6_HEADER_END > end {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
let ipv6 = &*((start + IPV6_HEADER_START) as *const Ipv6Hdr);
|
||||
|
||||
let (source_port, destination_port) = match ipv6.next_hdr {
|
||||
IpProto::Tcp => parse_tcp_port(start, end, IPV6_TCP_HEADER_START)?,
|
||||
IpProto::Udp => parse_udp_port(start, end, IPV6_UDP_HEADER_START)?,
|
||||
_ => return Err(()),
|
||||
};
|
||||
|
||||
*(target as *mut u32) = 1;
|
||||
|
||||
let ipv6_data_ptr = (target as *mut u8).add(16);
|
||||
|
||||
core::ptr::write(ipv6_data_ptr as *mut IpProto, ipv6.next_hdr);
|
||||
core::ptr::copy_nonoverlapping(
|
||||
ipv6.src_addr.as_ptr(),
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, src_ip)),
|
||||
16,
|
||||
);
|
||||
core::ptr::copy_nonoverlapping(
|
||||
ipv6.dst_addr.as_ptr(),
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, dst_ip)),
|
||||
16,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, src_port)) as *mut u16,
|
||||
source_port,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, dst_port)) as *mut u16,
|
||||
destination_port,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, packet_length)) as *mut u32,
|
||||
(end - start) as u32,
|
||||
);
|
||||
core::ptr::write(
|
||||
ipv6_data_ptr.add(core::mem::offset_of!(IPv6Event, timestamp_us)) as *mut u64,
|
||||
bpf_ktime_get_ns(),
|
||||
);
|
||||
|
||||
Ok(())
|
||||
if start + IPV6_HEADER_END > end {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
let ipv6 = &*((start + IPV6_HEADER_START) as *const Ipv6Hdr);
|
||||
|
||||
let (src_port, dst_port, tcp_flags, l4_hdr_len, window_size) = match ipv6.next_hdr {
|
||||
IpProto::Tcp => parse_tcp(start, end, IPV6_TCP_HEADER_START)?,
|
||||
IpProto::Udp => parse_udp(start, end, IPV6_UDP_HEADER_START)?,
|
||||
_ => return Err(()),
|
||||
};
|
||||
|
||||
// payload_len() is everything after the fixed 40-byte IPv6 header (L4 header + data).
|
||||
let payload_length = (ipv6.payload_len() as u32).saturating_sub(l4_hdr_len as u32);
|
||||
|
||||
*(target as *mut u32) = 1;
|
||||
|
||||
let p = (target as *mut u8).add(16);
|
||||
|
||||
core::ptr::write(p as *mut IpProto, ipv6.next_hdr);
|
||||
core::ptr::copy_nonoverlapping(ipv6.src_addr.as_ptr(), p.add(core::mem::offset_of!(IPv6Event, src_ip)), 16);
|
||||
core::ptr::copy_nonoverlapping(ipv6.dst_addr.as_ptr(), p.add(core::mem::offset_of!(IPv6Event, dst_ip)), 16);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, src_port)) as *mut u16, src_port);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, dst_port)) as *mut u16, dst_port);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, packet_length)) as *mut u32, (end - start) as u32);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, payload_length)) as *mut u32, payload_length);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, header_length)) as *mut u16, l4_hdr_len);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, timestamp_us)) as *mut u64, bpf_ktime_get_ns());
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, tcp_flags)) as *mut TcpFlags, tcp_flags);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, tcp_window_size)) as *mut u16, window_size);
|
||||
core::ptr::write(p.add(core::mem::offset_of!(IPv6Event, is_forward)) as *mut bool, false);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Mirror the C XDP pattern for variable-offset packet access:
|
||||
@ -148,26 +121,43 @@ unsafe fn parse_ipv6_packet(start: usize, end: usize, target: *mut Event) -> Res
|
||||
// instructions in LLVM IR, preventing the merge and keeping each protocol's
|
||||
// bounds check and accesses in the same basic block.
|
||||
#[inline(always)]
|
||||
unsafe fn parse_tcp_port(start: usize, end: usize, tcp_start: usize) -> Result<(u16, u16), ()> {
|
||||
unsafe {
|
||||
let tcph = (start as *const u8).add(tcp_start) as *const TcpHdr;
|
||||
// tcph.add(1) advances by size_of::<TcpHdr>() = 20 bytes, matching
|
||||
// the C pattern "if (tcph + 1 > data_end)".
|
||||
if tcph.add(1) as usize > end {
|
||||
return Err(());
|
||||
}
|
||||
Ok((u16::from_be_bytes((*tcph).source), u16::from_be_bytes((*tcph).dest)))
|
||||
unsafe fn parse_tcp(
|
||||
start: usize,
|
||||
end: usize,
|
||||
tcp_start: usize,
|
||||
) -> Result<(u16, u16, TcpFlags, u16, u16), ()> {
|
||||
let tcph = (start as *const u8).add(tcp_start) as *const TcpHdr;
|
||||
// tcph.add(1) advances by size_of::<TcpHdr>() = 20 bytes, matching
|
||||
// the C pattern "if (tcph + 1 > data_end)".
|
||||
if tcph.add(1) as usize > end {
|
||||
return Err(());
|
||||
}
|
||||
let src_port = u16::from_be_bytes((*tcph).source);
|
||||
let dst_port = u16::from_be_bytes((*tcph).dest);
|
||||
// doff() is the 4-bit data offset field; multiply by 4 to get header length in bytes.
|
||||
let header_length = (*tcph).doff() * 4;
|
||||
let window_size = u16::from_be_bytes((*tcph).window);
|
||||
let flags_byte = ((*tcph).fin() as u8)
|
||||
| (((*tcph).syn() as u8) << 1)
|
||||
| (((*tcph).rst() as u8) << 2)
|
||||
| (((*tcph).psh() as u8) << 3)
|
||||
| (((*tcph).ack() as u8) << 4)
|
||||
| (((*tcph).urg() as u8) << 5)
|
||||
| (((*tcph).ece() as u8) << 6)
|
||||
| (((*tcph).cwr() as u8) << 7);
|
||||
Ok((src_port, dst_port, TcpFlags::from_byte(flags_byte), header_length, window_size))
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
unsafe fn parse_udp_port(start: usize, end: usize, udp_start: usize) -> Result<(u16, u16), ()> {
|
||||
unsafe {
|
||||
let udph = (start as *const u8).add(udp_start) as *const UdpHdr;
|
||||
// udph.add(1) advances by size_of::<UdpHdr>() = 8 bytes.
|
||||
if udph.add(1) as usize > end {
|
||||
return Err(());
|
||||
}
|
||||
Ok(((*udph).src_port(), (*udph).dst_port()))
|
||||
unsafe fn parse_udp(
|
||||
start: usize,
|
||||
end: usize,
|
||||
udp_start: usize,
|
||||
) -> Result<(u16, u16, TcpFlags, u16, u16), ()> {
|
||||
let udph = (start as *const u8).add(udp_start) as *const UdpHdr;
|
||||
// udph.add(1) advances by size_of::<UdpHdr>() = 8 bytes.
|
||||
if udph.add(1) as usize > end {
|
||||
return Err(());
|
||||
}
|
||||
Ok(((*udph).src_port(), (*udph).dst_port(), TcpFlags::default(), 8, 0))
|
||||
}
|
||||
|
||||
@ -1,34 +1,34 @@
|
||||
use aya_ebpf::macros::map;
|
||||
use aya_ebpf::maps::LruHashMap;
|
||||
use aya_ebpf::maps::LruPerCpuHashMap;
|
||||
use common::define::setting::MAX_STATS;
|
||||
use common::model::event::{IPv4Event, IPv6Event};
|
||||
use common::model::flow_stats::FlowStats;
|
||||
use common::model::ip_address::{AddrPortV4, AddrPortV6};
|
||||
|
||||
#[map]
|
||||
static IPV4_EGRESS_SRC_1MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_SRC_1MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_EGRESS_SRC_10MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_SRC_10MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_EGRESS_SRC_1HOUR: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_SRC_1HOUR: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_SRC_1MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_SRC_1MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_SRC_10MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_SRC_10MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_SRC_1HOUR: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_SRC_1HOUR: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_EGRESS_DST_1MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_DST_1MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_EGRESS_DST_10MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_DST_10MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_EGRESS_DST_1HOUR: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_EGRESS_DST_1HOUR: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_DST_1MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_DST_1MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_DST_10MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_DST_10MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_EGRESS_DST_1HOUR: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_EGRESS_DST_1HOUR: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
|
||||
pub fn ipv4_update_stats(event: &IPv4Event) {
|
||||
let source = event.source_addr();
|
||||
@ -62,7 +62,7 @@ pub fn ipv6_update_stats(event: &IPv6Event) {
|
||||
// high load is tolerable. A future migration to PerCpuLruHashMap would eliminate
|
||||
// the race at the cost of per-CPU aggregation on the userspace read path.
|
||||
#[inline(always)]
|
||||
unsafe fn ipv4_update_flow_stats(map: &LruHashMap<AddrPortV4, FlowStats>, key: &AddrPortV4, event: &IPv4Event) {
|
||||
unsafe fn ipv4_update_flow_stats(map: &LruPerCpuHashMap<AddrPortV4, FlowStats>, key: &AddrPortV4, event: &IPv4Event) {
|
||||
if let Some(status) = map.get_ptr_mut(key) {
|
||||
(*status).bytes += event.packet_length as u64;
|
||||
(*status).packets += 1;
|
||||
@ -77,7 +77,7 @@ unsafe fn ipv4_update_flow_stats(map: &LruHashMap<AddrPortV4, FlowStats>, key: &
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
unsafe fn ipv6_update_flow_status(map: &LruHashMap<AddrPortV6, FlowStats>, key: &AddrPortV6, event: &IPv6Event) {
|
||||
unsafe fn ipv6_update_flow_status(map: &LruPerCpuHashMap<AddrPortV6, FlowStats>, key: &AddrPortV6, event: &IPv6Event) {
|
||||
if let Some(status) = map.get_ptr_mut(key) {
|
||||
(*status).bytes += event.packet_length as u64;
|
||||
(*status).packets += 1;
|
||||
|
||||
@ -1,34 +1,34 @@
|
||||
use aya_ebpf::macros::map;
|
||||
use aya_ebpf::maps::LruHashMap;
|
||||
use aya_ebpf::maps::LruPerCpuHashMap;
|
||||
use common::define::setting::MAX_STATS;
|
||||
use common::model::event::{IPv4Event, IPv6Event};
|
||||
use common::model::flow_stats::FlowStats;
|
||||
use common::model::ip_address::{AddrPortV4, AddrPortV6};
|
||||
|
||||
#[map]
|
||||
static IPV4_INGRESS_SRC_1MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_SRC_1MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_INGRESS_SRC_10MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_SRC_10MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_INGRESS_SRC_1HOUR: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_SRC_1HOUR: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_SRC_1MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_SRC_1MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_SRC_10MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_SRC_10MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_SRC_1HOUR: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_SRC_1HOUR: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_INGRESS_DST_1MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_DST_1MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_INGRESS_DST_10MIN: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_DST_10MIN: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV4_INGRESS_DST_1HOUR: LruHashMap<AddrPortV4, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV4_INGRESS_DST_1HOUR: LruPerCpuHashMap<AddrPortV4, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_DST_1MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_DST_1MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_DST_10MIN: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_DST_10MIN: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
#[map]
|
||||
static IPV6_INGRESS_DST_1HOUR: LruHashMap<AddrPortV6, FlowStats> = LruHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
static IPV6_INGRESS_DST_1HOUR: LruPerCpuHashMap<AddrPortV6, FlowStats> = LruPerCpuHashMap::with_max_entries(MAX_STATS as u32, 0);
|
||||
|
||||
pub fn ipv4_update_stats(event: &IPv4Event) {
|
||||
let source = event.source_addr();
|
||||
@ -62,7 +62,7 @@ pub fn ipv6_update_stats(event: &IPv6Event) {
|
||||
// high load is tolerable. A future migration to PerCpuLruHashMap would eliminate
|
||||
// the race at the cost of per-CPU aggregation on the userspace read path.
|
||||
#[inline(always)]
|
||||
unsafe fn ipv4_update_flow_stats(map: &LruHashMap<AddrPortV4, FlowStats>, key: &AddrPortV4, event: &IPv4Event) {
|
||||
unsafe fn ipv4_update_flow_stats(map: &LruPerCpuHashMap<AddrPortV4, FlowStats>, key: &AddrPortV4, event: &IPv4Event) {
|
||||
if let Some(status) = map.get_ptr_mut(key) {
|
||||
(*status).bytes += event.packet_length as u64;
|
||||
(*status).packets += 1;
|
||||
@ -77,7 +77,7 @@ unsafe fn ipv4_update_flow_stats(map: &LruHashMap<AddrPortV4, FlowStats>, key: &
|
||||
}
|
||||
|
||||
#[inline(always)]
|
||||
unsafe fn ipv6_update_flow_status(map: &LruHashMap<AddrPortV6, FlowStats>, key: &AddrPortV6, event: &IPv6Event) {
|
||||
unsafe fn ipv6_update_flow_status(map: &LruPerCpuHashMap<AddrPortV6, FlowStats>, key: &AddrPortV6, event: &IPv6Event) {
|
||||
if let Some(status) = map.get_ptr_mut(key) {
|
||||
(*status).bytes += event.packet_length as u64;
|
||||
(*status).packets += 1;
|
||||
|
||||
@ -2,7 +2,7 @@ use std::collections::HashMap;
|
||||
use std::net::{IpAddr, SocketAddrV4, SocketAddrV6};
|
||||
use std::sync::Arc;
|
||||
|
||||
use aya::maps::{HashMap as AyaHashMap, MapData};
|
||||
use aya::maps::{MapData, PerCpuHashMap};
|
||||
use aya::{Ebpf, Pod};
|
||||
use common::model::flow_stats::FlowStats;
|
||||
use common::model::ip_address::{AddrPortV4, AddrPortV6};
|
||||
@ -230,14 +230,14 @@ impl Statistics {
|
||||
}
|
||||
}
|
||||
|
||||
struct FlowMap<T> {
|
||||
map: AyaHashMap<MapData, T, FlowStats>,
|
||||
struct FlowMap<T: Pod> {
|
||||
map: PerCpuHashMap<MapData, T, FlowStats>,
|
||||
}
|
||||
|
||||
impl<T: NativeConvert + Pod> FlowMap<T> {
|
||||
fn new(ebpf: &mut Ebpf, map_name: &str) -> Result<Self, Error> {
|
||||
let map = ebpf.take_map(map_name).ok_or(EbpfError::MapNotFound)?;
|
||||
let map = AyaHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
|
||||
let map = PerCpuHashMap::try_from(map).map_err(EbpfError::MapOperationError)?;
|
||||
Ok(Self { map })
|
||||
}
|
||||
|
||||
@ -245,7 +245,18 @@ impl<T: NativeConvert + Pod> FlowMap<T> {
|
||||
self.map
|
||||
.iter()
|
||||
.filter_map(Result::ok)
|
||||
.map(|(key, value)| (key.into_native(), FlowStats::from(value)))
|
||||
.map(|(key, per_cpu)| {
|
||||
let agg = per_cpu.iter().fold(
|
||||
FlowStats { bytes: 0, packets: 0, last_seen: 0 },
|
||||
|mut acc, v| {
|
||||
acc.bytes += v.bytes;
|
||||
acc.packets += v.packets;
|
||||
acc.last_seen = acc.last_seen.max(v.last_seen);
|
||||
acc
|
||||
},
|
||||
);
|
||||
(key.into_native(), agg)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
@ -254,9 +265,13 @@ impl<T: NativeConvert + Pod> FlowMap<T> {
|
||||
.map
|
||||
.iter()
|
||||
.filter_map(|result| {
|
||||
result
|
||||
.ok()
|
||||
.and_then(|(key, stats)| (now - stats.last_seen - boot_time > window).then_some(key))
|
||||
result.ok().and_then(|(key, per_cpu)| {
|
||||
let max_last_seen = per_cpu.iter().map(|s| s.last_seen).max().unwrap_or(0);
|
||||
now.saturating_sub(max_last_seen)
|
||||
.saturating_sub(boot_time)
|
||||
.gt(&window)
|
||||
.then_some(key)
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
expired_keys.iter().for_each(|key| {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user