#107]: fixed problem with veth_tracer execution and tc_classifier execution. Code refactoring. Logging system partial improvements

Author: LorenzoTettamanti

core/src/components/identity/src/helpers.rs

@@ -1,25 +1,31 @@
+use crate::enums::IpProtocols;
+use crate::structs::{PacketLog, VethLog};
 use aya::{
-    maps::{ perf::{ PerfEventArray, PerfEventArrayBuffer }, MapData },
-    programs::{ SchedClassifier, TcAttachType },
-    util::online_cpus,
     Bpf,
+    maps::{
+        MapData,
+        perf::{PerfEventArray, PerfEventArrayBuffer},
+    },
+    programs::{SchedClassifier, TcAttachType},
+    util::online_cpus,
 };
-use crate::structs::{ PacketLog, VethLog };
 use bytes::BytesMut;
+use nix::net::if_::if_nameindex;
 use std::{
     ascii,
     borrow::BorrowMut,
     net::Ipv4Addr,
     string,
-    sync::{ atomic::{ AtomicBool, Ordering }, Arc },
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
 };
-use crate::enums::IpProtocols;
-use tracing::{ error, event, info, warn };
-use nix::net::if_::if_nameindex;
+use tracing::{error, event, info, warn};
 
-use tokio::{ fs, signal };
-use std::path::Path;
 use anyhow::Context;
+use std::path::Path;
+use tokio::{fs, signal};
 /*
  * decleare bpf path env variable
  */
@@ -47,19 +53,17 @@ impl TryFrom<u8> for IpProtocols {
 pub async fn display_events<T: BorrowMut<MapData>>(
     mut perf_buffers: Vec<PerfEventArrayBuffer<T>>,
     running: Arc<AtomicBool>,
-    mut buffers: Vec<BytesMut>
+    mut buffers: Vec<BytesMut>,
 ) {
-    info!("Triggering network events:");
     while running.load(Ordering::SeqCst) {
         for buf in perf_buffers.iter_mut() {
             match buf.read_events(&mut buffers) {
                 Ok(events) => {
                     for i in 0..events.read {
                         let data = &buffers[i];
                         if data.len() >= std::mem::size_of::<PacketLog>() {
-                            let pl: PacketLog = unsafe {
-                                std::ptr::read(data.as_ptr() as *const _)
-                            };
+                            let pl: PacketLog =
+                                unsafe { std::ptr::read(data.as_ptr() as *const _) };
                             let src = Ipv4Addr::from(u32::from_be(pl.src_ip));
                             let dst = Ipv4Addr::from(u32::from_be(pl.dst_ip));
                             let src_port = u16::from_be(pl.src_port as u16);
@@ -70,20 +74,12 @@ pub async fn display_events<T: BorrowMut<MapData>>(
                                 Ok(proto) => {
                                     info!(
                                         "Event Id: {} Protocol: {:?} SRC: {}:{} -> DST: {}:{}",
-                                        event_id,
-                                        proto,
-                                        src,
-                                        src_port,
-                                        dst,
-                                        dst_port
+                                        event_id, proto, src, src_port, dst, dst_port
                                     );
                                 }
-                                Err(_) =>
-                                    info!(
-                                        "Event Id: {} Protocol: Unknown ({})",
-                                        event_id,
-                                        pl.proto
-                                    ),
+                                Err(_) => {
+                                    info!("Event Id: {} Protocol: Unknown ({})", event_id, pl.proto)
+                                }
                             };
                         } else {
                             warn!("Received packet data too small: {} bytes", data.len());
@@ -104,17 +100,15 @@ pub async fn display_veth_events<T: BorrowMut<MapData>>(
     running: Arc<AtomicBool>,
     mut buffers: Vec<BytesMut>
 ) {
-    info!("Triggering veth events:");
     while running.load(Ordering::SeqCst) {
         for buf in perf_buffers.iter_mut() {
             match buf.read_events(&mut buffers) {
                 Ok(events) => {
                     for i in 0..events.read {
                         let data = &buffers[i];
                         if data.len() >= std::mem::size_of::<VethLog>() {
-                            let vethlog: VethLog = unsafe {
-                                std::ptr::read(data.as_ptr() as *const _)
-                            };
+                            let vethlog: VethLog =
+                                unsafe { std::ptr::read(data.as_ptr() as *const _) };
 
                             let name_bytes = vethlog.name;
 
@@ -167,11 +161,10 @@ 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
-                iface_name != "eth0" &&
-                iface_name != "docker0" &&
-                iface_name != "tunl0" &&
-                iface_name != "lo"
+            if iface_name != "eth0"
+                && iface_name != "docker0"
+                && iface_name != "tunl0"
+                && iface_name != "lo"
             {
                 interfaces.push(iface_name);
             } else {

core/src/components/identity/src/main.rs

@@ -3,39 +3,48 @@
  * Features:
  *   1. TCP, UDP , ICMP events tracker
  *   2. Track Connections using a PerfEventArray named ConnArray
+ *   3. Track veth creation and deletion events
  *
  */
 #![allow(warnings)]
 #![allow(unused_mut)]
 
+mod enums;
 mod helpers;
 mod structs;
-mod enums;
 use aya::{
-    maps::{ perf::{ PerfEventArray, PerfEventArrayBuffer }, Map, MapData },
-    programs::{ KProbe, SchedClassifier, TcAttachType },
+    Bpf, Ebpf,
+    maps::{
+        Map, MapData,
+        perf::{PerfEventArray, PerfEventArrayBuffer},
+    },
+    programs::{KProbe, SchedClassifier, TcAttachType},
     util::online_cpus,
-    Bpf,
-    Ebpf,
 };
 
-use bytes::BytesMut;
-use std::{ convert::TryInto, sync::{ atomic::{ AtomicBool, Ordering }, Arc }, path::Path };
-use crate::helpers::{ display_events, display_veth_events, get_veth_channels };
 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::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+};
 
-use tokio::{ fs, signal, sync::broadcast::error };
-use anyhow::{ Context, Ok };
-use tracing_subscriber::{ fmt::format::FmtSpan, EnvFilter };
-use tracing::{ info, error, warn };
+use anyhow::{Context, Ok};
+use tokio::{fs, signal, sync::broadcast::error};
+use tracing::{error, info, warn};
+use tracing_subscriber::{EnvFilter, fmt::format::FmtSpan};
 
 const BPF_PATH: &str = "BPF_PATH"; //BPF env path
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
     //init tracing subscriber
-    tracing_subscriber
-        ::fmt()
+    tracing_subscriber::fmt()
         .with_max_level(tracing::Level::INFO)
         .with_target(false)
         .with_level(true)
@@ -51,7 +60,9 @@ async fn main() -> Result<(), anyhow::Error> {
 
     //init conntracker data path
     let bpf_path = std::env::var(BPF_PATH).context("BPF_PATH environment variable required")?;
-    let data = fs::read(Path::new(&bpf_path)).await.context("failed to load file from path")?;
+    let data = fs::read(Path::new(&bpf_path))
+        .await
+        .context("failed to load file from path")?;
 
     //init bpf data
     let mut bpf = Bpf::load(&data)?;
@@ -66,35 +77,46 @@ async fn main() -> Result<(), anyhow::Error> {
     // everytime a new interface enters the InterfacesRegistry attach a bpf program with the attach_bpf_program function below
 
     info!("Found interfaces: {:?}", interfaces);
-    attach_bpf_program(&data, interfaces).await?;
+    init_tc_classifier(&mut bpf, interfaces)
+        .context("An error occured during the execution of attach_bpf_program function")?;
 
-    event_listener(bpf_maps).await?;
+    event_listener(bpf_maps)
+        .await
+        .context("Error initializing event_listener")?;
 
     Ok(())
 }
 
-//attach a program to a vector of interfaces
-async fn attach_bpf_program(data: &[u8], ifaces: Vec<String>) -> Result<(), anyhow::Error> {
-    // this function attach a bpf program to a vector of network interfaces
-
+//attach the tc classifier program to a vector of interfaces
+pub fn init_tc_classifier(bpf: &mut Ebpf, ifaces: Vec<String>) -> Result<(), anyhow::Error> {
+    //this funtion initialize the tc classifier program
     info!("Loading programs");
 
-    for interface in ifaces.iter() {
-        let mut bpf = Bpf::load(&data)?;
-        let program: &mut SchedClassifier = bpf
-            .program_mut("identity_classifier")
-            .ok_or_else(|| anyhow::anyhow!("program 'identity_classifier' not found"))?
-            .try_into()?;
-        program.load()?;
-
-        program.attach(&interface, TcAttachType::Ingress)?;
+    let program: &mut SchedClassifier = bpf
+        .program_mut("identity_classifier")
+        .ok_or_else(|| anyhow::anyhow!("program 'identity_classifier' not found"))?
+        .try_into()
+        .context("Failed to init SchedClassifier program")?;
+
+    program
+        .load()
+        .context("Failed to load identity_classifier program")?;
+
+    for interface in ifaces {
+        match program.attach(&interface, TcAttachType::Ingress) {
+            std::result::Result::Ok(_) => info!(
+                "Program 'identity_classifier' attached to interface {}",
+                interface
+            ),
+            Err(e) => error!(
+                "Error attaching program to interface {}: {:?}",
+                interface, e
+            ),
+        }
     }
 
-    info!("Programs attached to interfaces successfully");
-
     Ok(())
 }
-
 fn init_veth_tracer(bpf: &mut Ebpf) -> Result<(), anyhow::Error> {
     //this functions init the veth_tracer used to make the InterfacesRegistry
 
@@ -112,20 +134,24 @@ fn init_veth_tracer(bpf: &mut Ebpf) -> Result<(), anyhow::Error> {
         .program_mut("veth_deletion_trace")
         .ok_or_else(|| anyhow::anyhow!("program 'veth_deletion_trace' not found"))?
         .try_into()?;
-    veth_deletion_tracer.load().context("Failed to load deletetion_tracer program")?;
-
     veth_deletion_tracer
-        .attach("unregister_netdevice_queue", 0)
-        .context("Failed to attach to unregister_netdevice_queue")?;
+        .load()
+        .context("Failed to load deletetion_tracer program")?;
+
+    match veth_deletion_tracer.attach("unregister_netdevice_queue", 0) {
+        std::result::Result::Ok(_) => info!("veth_deletion_trace program attached successfully"),
+        Err(e) => error!("Error attaching veth_deletetion_trace program {:?}", e),
+    }
+
     Ok(())
 }
 
 fn init_bpf_maps(bpf: &mut Ebpf) -> Result<(Map, Map), anyhow::Error> {
     // this function init the bpfs maps used in the main program
-    /* 
-        index 0: events_map
-        index 1: veth_map
-     */
+    /*
+       index 0: events_map
+       index 1: veth_map
+    */
     let events_map = bpf
         .take_map("EventsMap")
         .ok_or_else(|| anyhow::anyhow!("EventsMap map not found"))?;
@@ -135,28 +161,28 @@ fn init_bpf_maps(bpf: &mut Ebpf) -> Result<(Map, Map), anyhow::Error> {
         .ok_or_else(|| anyhow::anyhow!("veth_identity_map map not found"))?;
 
     /* EDIT: this part is paused right now
-    info!("loading bpf connections map");
+       info!("loading bpf connections map");
 
-    //init connection map
-    let connections_map_raw = bpf
-        .take_map("ConnectionMap")
-        .context("failed to take connections map")?;
+       //init connection map
+       let connections_map_raw = bpf
+           .take_map("ConnectionMap")
+           .context("failed to take connections map")?;
 
-    let connection_tracker_map = bpf
-        .take_map("ConnectionTrackerMap")
-        .context("failed to take ConnectionTrackerMap map")?;
- */
+       let connection_tracker_map = bpf
+           .take_map("ConnectionTrackerMap")
+           .context("failed to take ConnectionTrackerMap map")?;
+    */
     Ok((events_map, veth_map))
 }
 
 async fn event_listener(bpf_maps: (Map, Map)) -> Result<(), anyhow::Error> {
     // this function init the event listener. Listens for veth events (creation/deletion) and network events (pod to pod communications)
     /* Doc:
-    
-        perf_net_events_array: contains is associated with the network events stored in the events_map (EventsMap)
-        perf_veth_array: contains is associated with the network events stored in the veth_map (veth_identity_map)
-    
-     */
+
+       perf_net_events_array: contains is associated with the network events stored in the events_map (EventsMap)
+       perf_veth_array: contains is associated with the network events stored in the veth_map (veth_identity_map)
+
+    */
 
     info!("Preparing perf_buffers and perf_arrays");
     // init PerfEventArrays
@@ -183,7 +209,7 @@ async fn event_listener(bpf_maps: (Map, Map)) -> Result<(), anyhow::Error> {
     let net_events_running = Arc::new(AtomicBool::new(true));
 
     let mut veth_buffers = vec![BytesMut::with_capacity(1024); 10];
-    let mut events_buffers = vec![BytesMut::with_capacity(1024); 10];
+    let mut events_buffers = vec![BytesMut::with_capacity(1024); online_cpus().iter().len()];
     //   let mut connections_buffers = vec![BytesMut::with_capacity(1024); 10];
 
     let veth_running_signal = veth_running.clone();
@@ -201,14 +227,14 @@ async fn event_listener(bpf_maps: (Map, Map)) -> Result<(), anyhow::Error> {
         result = veth_events_displayer=>{
             match result{
                 Err(e)=>error!("veth_event_displayer panicked {:?}",e),
-                std::result::Result::Ok(_) => todo!(),
+                std::result::Result::Ok(_) => info!("Found new veth_event"),
                 }
         }
 
         result = net_events_displayer=>{
             match result{
                 Err(e)=>error!("net_event_displayer panicked {:?}",e),
-                std::result::Result::Ok(_)  => todo!(),
+                std::result::Result::Ok(_)  => info!("Found new net_event"),
             }
         }
         _= signal::ctrl_c()=>{