[#105]: added netns and pid parameter in identity service (user space). Better concurrency management. added bpf maps preliminary checks in main.rs

LorenzoTettamanti · LorenzoTettamanti · commit 9a115b928882 · 2025-07-24T15:19:57.000+02:00
diff --git a/core/src/components/identity/src/helpers.rs b/core/src/components/identity/src/helpers.rs
@@ -33,8 +33,6 @@ use tokio::{fs, signal};
  * decleare bpf path env variable
  */
 const BPF_PATH: &str = "BPF_PATH";
-const IFACE: &str = "IFACE";
-use std::result::Result::Ok as Okk;
 
 /*
  * TryFrom Trait implementation for IpProtocols enum
@@ -62,7 +60,7 @@ pub async fn display_events<T: BorrowMut<MapData>>(
     while running.load(Ordering::SeqCst) {
         for buf in perf_buffers.iter_mut() {
             match buf.read_events(&mut buffers) {
-                Ok(events) => {
+                std::result::Result::Ok(events) => {
                     for i in 0..events.read {
                         let data = &buffers[i];
                         if data.len() >= std::mem::size_of::<PacketLog>() {
@@ -72,10 +70,10 @@ pub async fn display_events<T: BorrowMut<MapData>>(
                             let dst = Ipv4Addr::from(u32::from_be(pl.dst_ip));
                             let src_port = u16::from_be(pl.src_port as u16);
                             let dst_port = u16::from_be(pl.dst_port as u16);
-                            let event_id = pl.event_id;
+                            let event_id = pl.pid;
 
                             match IpProtocols::try_from(pl.proto) {
-                                Ok(proto) => {
+                                std::result::Result::Ok(proto) => {
                                     info!(
                                         "Event Id: {} Protocol: {:?} SRC: {}:{} -> DST: {}:{}",
                                         event_id, proto, src, src_port, dst, dst_port
@@ -109,7 +107,7 @@ pub async fn display_veth_events<T: BorrowMut<MapData>>(
     while running.load(Ordering::SeqCst) {
         for buf in perf_buffers.iter_mut() {
             match buf.read_events(&mut buffers) {
-                Ok(events) => {
+                std::result::Result::Ok(events) => {
                     for i in 0..events.read {
                         let data = &buffers[i];
                         if data.len() >= std::mem::size_of::<VethLog>() {
@@ -123,6 +121,7 @@ pub async fn display_veth_events<T: BorrowMut<MapData>>(
                             let state = vethlog.state;
 
                             let dev_addr = dev_addr_bytes;
+                            let netns = vethlog.netns;
                             let mut event_type = String::new();
                             match vethlog.event_type {
                                 1 => {
@@ -134,15 +133,22 @@ pub async fn display_veth_events<T: BorrowMut<MapData>>(
                                 _ => warn!("unknown event_type"),
                             }
                             match name {
-                                Ok(veth_name) => {
+                                std::result::Result::Ok(veth_name) => {
                                     info!(
-                                        "Triggered action: register_netdevice event_type:{:?} Manipulated veth: {:?} state:{:?} dev_addr:{:?}",
+                                        "[{}] Triggered action: register_netdevice event_type:{:?} Manipulated veth: {:?} state:{:?} dev_addr:{:?}",
+                                        netns,
                                         event_type,
                                         veth_name.trim_end_matches("\0").to_string(),
                                         state,
                                         dev_addr
                                     );
-                                    attach_detach_veth(bpf.clone(), vethlog.event_type, veth_name, link_ids.clone()).await;
+                                    attach_detach_veth(
+                                        bpf.clone(),
+                                        vethlog.event_type,
+                                        veth_name,
+                                        link_ids.clone(),
+                                    )
+                                    .await;
                                 }
                                 Err(_) => info!("Unknown name or corrupted field"),
                             }
@@ -173,8 +179,7 @@ pub fn get_veth_channels() -> Vec<String> {
     if let Ok(ifaces) = if_nameindex() {
         for iface in &ifaces {
             let iface_name = iface.name().to_str().unwrap().to_owned();
-            if !ignore_iface(&iface_name)
-            {
+            if !ignore_iface(&iface_name) {
                 interfaces.push(iface_name);
             } else {
                 info!("skipping interface {:?}", iface_name);
@@ -185,8 +190,16 @@ pub fn get_veth_channels() -> Vec<String> {
     interfaces
 }
 
-async fn attach_detach_veth(bpf: Arc<Mutex<Bpf>>, event_type: u8, iface: &str, link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>) -> Result<(), anyhow::Error> {
-    info!("attach_detach_veth called: event_type={}, iface={}", event_type, iface);
+async fn attach_detach_veth(
+    bpf: Arc<Mutex<Bpf>>,
+    event_type: u8,
+    iface: &str,
+    link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>,
+) -> Result<(), anyhow::Error> {
+    info!(
+        "attach_detach_veth called: event_type={}, iface={}",
+        event_type, iface
+    );
     match event_type {
         1 => {
             let mut bpf = bpf.lock().unwrap();
@@ -204,11 +217,13 @@ async fn attach_detach_veth(bpf: Arc<Mutex<Bpf>>, event_type: u8, iface: &str, l
 
             let mut link_ids = link_ids.lock().unwrap();
             match program.attach(iface, TcAttachType::Ingress) {
-                Ok(link_id) => {
-                    info!("Program 'identity_classifier' attached to interface {}", iface);
+                std::result::Result::Ok(link_id) => {
+                    info!(
+                        "Program 'identity_classifier' attached to interface {}",
+                        iface
+                    );
                     link_ids.insert(iface.to_string(), link_id);
-
-                },
+                }
                 Err(e) => error!("Error attaching program to interface {}: {:?}", iface, e),
             }
         }
@@ -230,4 +245,4 @@ async fn attach_detach_veth(bpf: Arc<Mutex<Bpf>>, event_type: u8, iface: &str, l
         }
     }
     Ok(())
-}
+}
diff --git a/core/src/components/identity/src/main.rs b/core/src/components/identity/src/main.rs
@@ -7,37 +7,40 @@
  *   3. Track veth creation and deletion events
  *
  */
-#![allow(warnings)]
 #![allow(unused_mut)]
+#![allow(warnings)]
 
 mod enums;
 mod helpers;
 mod structs;
 use aya::{
+    Bpf,
     maps::{
-        perf::{PerfEventArray, PerfEventArrayBuffer}, Map, MapData
-    }, programs::{tc::SchedClassifierLinkId, KProbe, SchedClassifier, TcAttachType}, util::online_cpus, Bpf, Ebpf
+        Map, MapData,
+        perf::{PerfEventArray, PerfEventArrayBuffer},
+    },
+    programs::{KProbe, SchedClassifier, TcAttachType, tc::SchedClassifierLinkId},
+    util::online_cpus,
 };
 
-use crate::enums::IpProtocols;
 use crate::helpers::{display_events, display_veth_events, get_veth_channels};
 use bytes::BytesMut;
 use std::{
     convert::TryInto,
     path::Path,
     sync::{
-        atomic::{AtomicBool, Ordering}, Arc, Mutex
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
     },
 };
 
 use anyhow::{Context, Ok};
-use tokio::{fs, signal, sync::broadcast::error};
-use tracing::{error, info, warn};
+use tokio::{fs, signal};
+use tracing::{error, info};
 use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan};
 
 const BPF_PATH: &str = "BPF_PATH"; //BPF env path
 use std::collections::HashMap;
-use std::result::Result::Ok as Okk;
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
@@ -68,25 +71,35 @@ async fn main() -> Result<(), anyhow::Error> {
     //init bpf data
     let bpf = Arc::new(Mutex::new(Bpf::load(&data)?));
 
-    //load veth_trace program ref veth_trace.rs
-    init_veth_tracer(bpf.clone());
-    let bpf_maps = init_bpf_maps(bpf.clone()).unwrap();
+    match init_bpf_maps(bpf.clone()) {
+        std::result::Result::Ok(bpf_maps) => {
+            info!("Successfully loaded bpf maps");
 
-    let interfaces = get_veth_channels();
+            //load veth_trace program ref veth_trace.rs
+            init_veth_tracer(bpf.clone()).await?;
 
-    info!("Found interfaces: {:?}", interfaces);
-    init_tc_classifier(bpf.clone(), interfaces, link_ids.clone())
-        .context("An error occured during the execution of attach_bpf_program function")?;
+            let interfaces = get_veth_channels();
 
-    event_listener(bpf_maps, link_ids.clone(), bpf.clone())
-        .await
-        .context("Error initializing event_listener")?;
+            info!("Found interfaces: {:?}", interfaces);
+            init_tc_classifier(bpf.clone(), interfaces, link_ids.clone())
+                .await
+                .context("An error occured during the execution of attach_bpf_program function")?;
+
+            event_listener(bpf_maps, link_ids.clone(), bpf.clone())
+                .await
+                .context("Error initializing event_listener")?;
+        }
+        Err(e) => {
+            error!("Error while loading bpf maps {}", e);
+            signal::ctrl_c();
+        }
+    }
 
     Ok(())
 }
 
 //attach the tc classifier program to a vector of interfaces
-pub fn init_tc_classifier(
+async fn init_tc_classifier(
     bpf: Arc<Mutex<Bpf>>,
     ifaces: Vec<String>,
     link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>,
@@ -108,7 +121,7 @@ pub fn init_tc_classifier(
 
     for interface in ifaces {
         match program.attach(&interface, TcAttachType::Ingress) {
-            Okk(link_id) => {
+            std::result::Result::Ok(link_id) => {
                 info!(
                     "Program 'identity_classifier' attached to interface {}",
                     interface
@@ -126,7 +139,7 @@ pub fn init_tc_classifier(
     Ok(())
 }
 
-fn init_veth_tracer(bpf: Arc<Mutex<Bpf>>) -> Result<(), anyhow::Error> {
+async fn init_veth_tracer(bpf: Arc<Mutex<Bpf>>) -> Result<(), anyhow::Error> {
     //this functions init the veth_tracer used to make the InterfacesRegistry
 
     let mut bpf_new = bpf.lock().unwrap();
@@ -138,7 +151,10 @@ fn init_veth_tracer(bpf: Arc<Mutex<Bpf>>) -> Result<(), anyhow::Error> {
         .try_into()?;
     veth_creation_tracer.load()?;
 
-    veth_creation_tracer.attach("register_netdevice", 0)?;
+    match veth_creation_tracer.attach("register_netdevice", 0) {
+        std::result::Result::Ok(_) => info!("veth_creation_tracer program attached successfully"),
+        Err(e) => error!("Error attaching veth_creation_tracer program {:?}", e),
+    }
 
     //deletion tracer
     let veth_deletion_tracer: &mut KProbe = bpf_new
@@ -188,7 +204,11 @@ fn init_bpf_maps(bpf: Arc<Mutex<Bpf>>) -> Result<(Map, Map), anyhow::Error> {
     Ok((events_map, veth_map))
 }
 
-async fn event_listener(bpf_maps: (Map, Map), link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>, bpf: Arc<Mutex<Bpf>>) -> Result<(), anyhow::Error> {
+async fn event_listener(
+    bpf_maps: (Map, Map),
+    link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>,
+    bpf: Arc<Mutex<Bpf>>,
+) -> Result<(), anyhow::Error> {
     // this function init the event listener. Listens for veth events (creation/deletion) and network events (pod to pod communications)
     /* Doc:
 
@@ -230,7 +250,14 @@ async fn event_listener(bpf_maps: (Map, Map), link_ids: Arc<Mutex<HashMap<String
 
     //display_events(perf_buffers, running, buffers).await;
     let veth_events_displayer = tokio::spawn(async move {
-        display_veth_events(bpf.clone(), perf_veth_buffer, veth_running, veth_buffers, veth_link_ids, ).await;
+        display_veth_events(
+            bpf.clone(),
+            perf_veth_buffer,
+            veth_running,
+            veth_buffers,
+            veth_link_ids,
+        )
+        .await;
     });
     let net_events_displayer = tokio::spawn(async move {
         display_events(perf_net_events_buffer, net_events_running, events_buffers).await;
diff --git a/core/src/components/identity/src/structs.rs b/core/src/components/identity/src/structs.rs
@@ -6,11 +6,12 @@
 #[derive(Clone, Copy)]
 pub struct PacketLog {
     pub proto: u8,
-    pub  src_ip: u32,
-    pub  src_port: u16,
+    pub src_ip: u32,
+    pub src_port: u16,
     pub dst_ip: u32,
     pub dst_port: u16,
     pub event_id: u16,
+    pub pid : u32
 }
 /*
  * Connection Array that contains the hash_id associated with an active connection
@@ -32,6 +33,8 @@ unsafe impl aya::Pod for ConnArray {}
 pub struct VethLog {
     pub name: [u8; 16],
     pub state: u64,
-    pub dev_addr: [u32;8],
+    pub dev_addr: [u32; 8],
     pub event_type: u8,
+    pub netns: u32,
+    pub pid: u32,
 }
