[Feature] Implement eBPF kernel programs and userspace integration (#682, #683, #684, #685)

Phase 2 of the Rust dataplane migration: implement the three eBPF
kernel programs and wire up their userspace integration.

eBPF programs (novaedge-ebpf):
- XDP L4 Load Balancer: VIP lookup, conntrack, backend selection via
  5-tuple hash, DNAT rewrite with IP/L4 checksum updates, XDP_TX
- XDP ARP Responder: responds to ARP requests for VIP addresses using
  VIP_ADDRS map, generates ARP replies in-place
- TC Rate Limiter: per-source-IP token bucket algorithm with
  configurable rate/burst, emits flow events on drops

Shared types (novaedge-common):
- Add RateLimitCfg struct for per-IP rate limit configuration
- Add Pod impl for RateLimitCfg (Linux userspace)

Userspace integration (novaedge-dataplane):
- maps.rs: implement RealMaps with actual aya map types (vips,
  backends, conntrack, rate_limits, rate_limit_cfg, vip_addrs)
- loader.rs: extract real map handles after loading, add attach_xdp
  and attach_tc helpers, return flow ring buffer handle
- flows.rs: add run_flow_reader_real() using AsyncFd for ring buffer
  polling on Linux
- server.rs: formatting improvements from cargo fmt

Author: piwi3910

dataplane/novaedge-common/src/lib.rs

@@ -105,6 +105,16 @@ pub struct RateLimitKey {
     pub src_ip: u32,
 }
 
+/// Rate limiting configuration (per source IP).
+#[repr(C)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct RateLimitCfg {
+    /// Tokens per second (refill rate).
+    pub rate: u64,
+    /// Maximum tokens (bucket capacity).
+    pub burst: u64,
+}
+
 /// Rate limiting value (token bucket state).
 #[repr(C)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -159,6 +169,7 @@ impl_pod!(
     ConnTrackKey,
     ConnTrackValue,
     RateLimitKey,
+    RateLimitCfg,
     RateLimitValue,
     FlowEvent,
 );
@@ -203,6 +214,11 @@ mod tests {
         assert_eq!(mem::size_of::<RateLimitKey>(), 4);
     }
 
+    #[test]
+    fn test_rate_limit_cfg_size() {
+        assert_eq!(mem::size_of::<RateLimitCfg>(), 16);
+    }
+
     #[test]
     fn test_rate_limit_value_size() {
         assert_eq!(mem::size_of::<RateLimitValue>(), 16);
@@ -220,15 +236,24 @@ mod tests {
 
     #[test]
     fn test_vip_key_fields() {
-        let key = VipKey { vip: 0x0A000001, port: 80, protocol: 6, _pad: 0 };
+        let key = VipKey {
+            vip: 0x0A000001,
+            port: 80,
+            protocol: 6,
+            _pad: 0,
+        };
         assert_eq!(key.vip, 0x0A000001);
         assert_eq!(key.port, 80);
         assert_eq!(key.protocol, 6);
     }
 
     #[test]
     fn test_backend_value_fields() {
-        let val = BackendValue { addr: 0x0A000002, port: 8080, weight: 100 };
+        let val = BackendValue {
+            addr: 0x0A000002,
+            port: 8080,
+            weight: 100,
+        };
         assert_eq!(val.addr, 0x0A000002);
         assert_eq!(val.port, 8080);
         assert_eq!(val.weight, 100);
@@ -260,7 +285,12 @@ mod tests {
 
     #[test]
     fn test_default_values() {
-        let key = VipKey { vip: 0, port: 0, protocol: 0, _pad: 0 };
+        let key = VipKey {
+            vip: 0,
+            port: 0,
+            protocol: 0,
+            _pad: 0,
+        };
         assert_eq!(key.vip, 0);
     }
 }

dataplane/novaedge-dataplane/src/flows.rs

@@ -8,6 +8,9 @@ use std::net::Ipv4Addr;
 use tokio::sync::broadcast;
 use tracing::{debug, info};
 
+#[cfg(target_os = "linux")]
+use tracing::warn;
+
 use crate::proto;
 
 /// Capacity of the broadcast channel for flow events.
@@ -18,7 +21,10 @@ const FLOW_CHANNEL_CAPACITY: usize = 4096;
 /// Returns `(sender, receiver)`. The sender is cloneable and should be passed
 /// to the flow reader task and the gRPC service. Additional receivers can be
 /// created via `sender.subscribe()`.
-pub fn flow_channel() -> (broadcast::Sender<proto::FlowEvent>, broadcast::Receiver<proto::FlowEvent>) {
+pub fn flow_channel() -> (
+    broadcast::Sender<proto::FlowEvent>,
+    broadcast::Receiver<proto::FlowEvent>,
+) {
     broadcast::channel(FLOW_CHANNEL_CAPACITY)
 }
 
@@ -80,6 +86,54 @@ impl MockFlowInjector {
     }
 }
 
+/// Start the real flow reader on Linux, reading from the eBPF ring buffer.
+///
+/// This uses `AsyncFd` to efficiently wait for ring buffer data availability,
+/// then reads and broadcasts each event.
+#[cfg(target_os = "linux")]
+pub async fn run_flow_reader_real(
+    mut ring_buf: aya::maps::RingBuf<aya::maps::MapData>,
+    tx: broadcast::Sender<proto::FlowEvent>,
+) {
+    use std::os::fd::AsRawFd;
+    use tokio::io::unix::AsyncFd;
+
+    info!("Flow reader: starting eBPF ring buffer reader");
+
+    let fd = ring_buf.as_raw_fd();
+    let async_fd = match AsyncFd::new(fd) {
+        Ok(afd) => afd,
+        Err(e) => {
+            warn!("Flow reader: failed to create AsyncFd: {e}");
+            return;
+        }
+    };
+
+    loop {
+        // Wait for the ring buffer to become readable.
+        let mut guard = match async_fd.readable().await {
+            Ok(g) => g,
+            Err(e) => {
+                warn!("Flow reader: readable() error: {e}");
+                break;
+            }
+        };
+
+        // Drain all available events.
+        while let Some(event_data) = ring_buf.next() {
+            if event_data.len() < core::mem::size_of::<novaedge_common::FlowEvent>() {
+                continue;
+            }
+            let raw: &novaedge_common::FlowEvent =
+                unsafe { &*(event_data.as_ptr() as *const novaedge_common::FlowEvent) };
+            let proto_event = raw_to_proto(raw);
+            let _ = tx.send(proto_event);
+        }
+
+        guard.clear_ready();
+    }
+}
+
 /// Start the flow reader task.
 ///
 /// On Linux with a real ring buffer, this would read from the eBPF ring buffer
@@ -88,10 +142,7 @@ impl MockFlowInjector {
 ///
 /// Flow events are broadcast via the sender. The gRPC `StreamFlows` RPC
 /// subscribes to this channel to deliver events to clients.
-pub async fn run_flow_reader(
-    tx: broadcast::Sender<proto::FlowEvent>,
-    mock_mode: bool,
-) {
+pub async fn run_flow_reader(tx: broadcast::Sender<proto::FlowEvent>, mock_mode: bool) {
     if mock_mode {
         info!("Flow reader running in mock mode (no eBPF ring buffer)");
         // In mock mode, we just keep the task alive. Events can be injected
@@ -101,22 +152,12 @@ pub async fn run_flow_reader(
         return;
     }
 
-    // On Linux with real eBPF, we would read from the ring buffer here.
-    // This is stubbed for Phase 2 when eBPF programs are implemented.
+    // On Linux with real eBPF, the ring buffer reader is started via
+    // `run_flow_reader_real()` with the actual RingBuf handle.
+    // If we reach here in non-mock mode without a ring buffer, just wait.
     #[cfg(target_os = "linux")]
     {
-        info!("Flow reader: eBPF ring buffer reader not yet implemented");
-        // TODO(Phase 2): Use aya::maps::RingBuf with AsyncFd to poll events.
-        //   let async_fd = AsyncFd::new(ring_buf_fd)?;
-        //   loop {
-        //       let guard = async_fd.readable().await?;
-        //       while let Some(event) = ring_buf.next() {
-        //           let raw = parse_flow_event(event);
-        //           let proto_event = raw_to_proto(&raw);
-        //           let _ = tx.send(proto_event);
-        //       }
-        //       guard.clear_ready();
-        //   }
+        info!("Flow reader: waiting for ring buffer (use run_flow_reader_real for eBPF)");
         let _ = std::future::pending::<()>().await;
     }
 

dataplane/novaedge-dataplane/src/loader.rs

@@ -12,25 +12,121 @@ pub struct LoadResult {
     pub map_manager: MapManager,
     /// Whether real eBPF programs were loaded (vs mock fallback).
     pub is_real: bool,
+    /// Ring buffer for flow events (Linux only, None in mock mode).
+    #[cfg(target_os = "linux")]
+    pub flow_ring_buf: Option<aya::maps::RingBuf<aya::maps::MapData>>,
 }
 
 /// Load eBPF programs from the compiled object file and return a `LoadResult`.
+///
+/// Extracts all map handles (VIPs, backends, conntrack, rate limits, VIP
+/// addresses), attaches the XDP and TC programs, and returns a
+/// `MapManager` wrapping the real maps plus the flow-event ring buffer.
 #[cfg(target_os = "linux")]
 pub fn load_ebpf(path: &str) -> anyhow::Result<LoadResult> {
+    use aya::programs::{tc, SchedClassifier, TcAttachType, Xdp, XdpFlags};
     use aya::Ebpf;
     use tracing::info;
 
+    use crate::maps::RealMaps;
+
     info!(path = %path, "Loading eBPF programs");
-    let _bpf = Ebpf::load_file(path)?;
+    let mut bpf = Ebpf::load_file(path)?;
+
+    // ── Extract map handles ───────────────────────────────────────────
+    let vips = aya::maps::HashMap::try_from(
+        bpf.take_map("VIPS")
+            .ok_or_else(|| anyhow::anyhow!("map VIPS not found"))?,
+    )?;
+    let backends = aya::maps::HashMap::try_from(
+        bpf.take_map("BACKENDS")
+            .ok_or_else(|| anyhow::anyhow!("map BACKENDS not found"))?,
+    )?;
+    let conntrack = aya::maps::HashMap::try_from(
+        bpf.take_map("CONNTRACK")
+            .ok_or_else(|| anyhow::anyhow!("map CONNTRACK not found"))?,
+    )?;
+    let rate_limits = aya::maps::HashMap::try_from(
+        bpf.take_map("RATE_LIMIT_STATE")
+            .ok_or_else(|| anyhow::anyhow!("map RATE_LIMIT_STATE not found"))?,
+    )?;
+    let rate_limit_cfg = aya::maps::HashMap::try_from(
+        bpf.take_map("RATE_LIMIT_CFG")
+            .ok_or_else(|| anyhow::anyhow!("map RATE_LIMIT_CFG not found"))?,
+    )?;
+    let vip_addrs = aya::maps::HashMap::try_from(
+        bpf.take_map("VIP_ADDRS")
+            .ok_or_else(|| anyhow::anyhow!("map VIP_ADDRS not found"))?,
+    )?;
+    let flow_ring_buf = aya::maps::RingBuf::try_from(
+        bpf.take_map("FLOW_EVENTS")
+            .ok_or_else(|| anyhow::anyhow!("map FLOW_EVENTS not found"))?,
+    )?;
+
+    info!("eBPF maps extracted successfully");
+
+    let maps = RealMaps {
+        vips,
+        backends,
+        conntrack,
+        rate_limits,
+        rate_limit_cfg,
+        vip_addrs,
+    };
 
-    // TODO(Phase 2): Extract map handles and attach XDP/TC programs.
-    info!("eBPF programs loaded (maps not yet extracted, using mock)");
     Ok(LoadResult {
-        map_manager: MapManager::new_mock(),
-        is_real: false,
+        map_manager: MapManager::new_real(maps),
+        is_real: true,
+        flow_ring_buf: Some(flow_ring_buf),
     })
 }
 
+/// Attach the XDP programs to the specified network interface.
+///
+/// This is called separately from `load_ebpf` so the caller can choose
+/// when and to which interface to attach.
+#[cfg(target_os = "linux")]
+pub fn attach_xdp(bpf: &mut aya::Ebpf, program_name: &str, interface: &str) -> anyhow::Result<()> {
+    use aya::programs::{Xdp, XdpFlags};
+    use tracing::info;
+
+    let prog: &mut Xdp = bpf
+        .program_mut(program_name)
+        .ok_or_else(|| anyhow::anyhow!("XDP program '{program_name}' not found"))?
+        .try_into()?;
+    prog.load()?;
+    prog.attach(interface, XdpFlags::default())?;
+    info!(
+        program = program_name,
+        interface = interface,
+        "XDP program attached"
+    );
+    Ok(())
+}
+
+/// Attach the TC classifier program to the specified network interface.
+#[cfg(target_os = "linux")]
+pub fn attach_tc(bpf: &mut aya::Ebpf, program_name: &str, interface: &str) -> anyhow::Result<()> {
+    use aya::programs::{tc, SchedClassifier, TcAttachType};
+    use tracing::info;
+
+    // Add clsact qdisc (ignore error if already exists)
+    let _ = tc::qdisc_add_clsact(interface);
+
+    let prog: &mut SchedClassifier = bpf
+        .program_mut(program_name)
+        .ok_or_else(|| anyhow::anyhow!("TC program '{program_name}' not found"))?
+        .try_into()?;
+    prog.load()?;
+    prog.attach(interface, TcAttachType::Ingress)?;
+    info!(
+        program = program_name,
+        interface = interface,
+        "TC program attached"
+    );
+    Ok(())
+}
+
 /// Stub for non-Linux platforms.
 #[cfg(not(target_os = "linux"))]
 pub fn load_ebpf(_path: &str) -> anyhow::Result<LoadResult> {
@@ -46,6 +142,8 @@ mod tests {
         let result = LoadResult {
             map_manager: MapManager::new_mock(),
             is_real: false,
+            #[cfg(target_os = "linux")]
+            flow_ring_buf: None,
         };
         assert!(!result.is_real);
         assert_eq!(result.map_manager.mode(), "mock");