[#158]: added a map_manager to call BpfMaps using the map name instead of relying on the declaration order during the startup

Author: LorenzoTettamanti

core/common/src/map_handlers.rs

@@ -53,7 +53,7 @@ pub fn init_bpf_maps(
 //takes an array of bpf maps and pin them to persist session data
 
 #[cfg(feature = "map-handlers")]
-pub fn map_pinner(maps: BpfMapsData, path: &PathBuf) -> Result<Vec<Map>, Error> {
+pub fn map_pinner(maps: BpfMapsData, path: &PathBuf) -> Result<BpfMapsData, Error> {
     if !path.exists() {
         info!("Pin path {:?} does not exist. Creating it...", path);
         std::fs::create_dir_all(&path)?;
@@ -64,7 +64,11 @@ pub fn map_pinner(maps: BpfMapsData, path: &PathBuf) -> Result<Vec<Map>, Error>
         }
     }
 
-    let mut owned_maps = Vec::new(); // aya::Maps does not implement the clone trait i need to create a raw copy of the vec map
+    //let mut owned_maps = Vec::new(); // aya::Maps does not implement the clone trait i need to create a raw copy of the vec map
+    let mut owned_bpf_maps_data = BpfMapsData {
+        bpf_obj_names: Vec::new(),
+        bpf_obj_map: Vec::new(),
+    };
     // an iterator that iterates two iterators simultaneously
     for (map_obj, name) in maps
         .bpf_obj_map
@@ -79,10 +83,12 @@ pub fn map_pinner(maps: BpfMapsData, path: &PathBuf) -> Result<Vec<Map>, Error>
         }
         info!("Trying to pin map {:?} in map path: {:?}", name, &map_path);
         map_obj.pin(&map_path)?;
-        owned_maps.push(map_obj);
+        //owned_maps.push(map_obj);
+        owned_bpf_maps_data.bpf_obj_names.push(name);
+        owned_bpf_maps_data.bpf_obj_map.push(map_obj);
     }
 
-    Ok(owned_maps)
+    Ok(owned_bpf_maps_data) // return a BpfMapsData type 
 }
 
 #[cfg(feature = "map-handlers")]

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

@@ -14,7 +14,10 @@ mod service_discovery;
 use crate::helpers::{get_veth_channels, read_perf_buffer};
 use aya::{
     Ebpf,
-    maps::{Map, perf::PerfEventArray},
+    maps::{
+        MapData,
+        perf::{PerfEventArray, PerfEventArrayBuffer},
+    },
     programs::{SchedClassifier, TcAttachType, tc::SchedClassifierLinkId},
     util::online_cpus,
 };
@@ -25,6 +28,7 @@ use crate::helpers::scan_cgroup_cronjob;
 use bytes::BytesMut;
 use cortexbrain_common::map_handlers::{init_bpf_maps, map_pinner, populate_blocklist};
 use cortexbrain_common::program_handlers::load_program;
+use cortexbrain_common::{buffer_type::BufferType, map_handlers::BpfMapsData};
 use std::{
     convert::TryInto,
     path::Path,
@@ -40,8 +44,7 @@ use std::collections::HashMap;
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
-    //init tracing subscriber
-    //logger::init_default_logger();
+    //init otlè tracing subscriber
     let otlp_provider = logger::otlp_logger_init("identity_service-OTLP".to_string());
 
     info!("Starting identity service...");
@@ -195,57 +198,65 @@ async fn init_tcp_registry(bpf: Arc<Mutex<Ebpf>>) -> Result<(), anyhow::Error> {
 //   perf_veth_array: contains is associated with the network events stored in the veth_map (veth_identity_map)
 //
 //
-async fn event_listener(
-    bpf_maps: Vec<Map>,
-    //link_ids: Arc<Mutex<HashMap<String, SchedClassifierLinkId>>>,
-    //bpf: Arc<Mutex<Ebpf>>,
-) -> Result<(), anyhow::Error> {
+async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error> {
     info!("Preparing perf_buffers and perf_arrays");
 
     //TODO: try to change from PerfEventArray to a RingBuffer data structure
 
-    let mut perf_event_arrays = Vec::new(); // contains a vector of PerfEventArrays
-    let mut event_buffers = Vec::new(); // contains a vector of buffers
+    let mut map_manager =
+        HashMap::<String, (PerfEventArray<MapData>, Vec<PerfEventArrayBuffer<MapData>>)>::new();
+
+    // create the PerfEventArrays and the buffers from the BpfMapsData Objects
+    for (map, name) in bpf_maps
+        .bpf_obj_map
+        .into_iter()
+        .zip(bpf_maps.bpf_obj_names.into_iter())
+    // zip two iterators at the same time for map and mapnames
+    {
+        debug!("Debugging map type:{:?} for map name {:?}", map, &name);
+        info!("Creating PerfEventArray for map name {:?}", &name);
 
-    // create the PerfEventArrays and the buffers
-    for map in bpf_maps {
-        debug!("Debugging map type:{:?}", map);
+        // save the map in a registry if is a PerfEventArray to access them by name
         if let std::result::Result::Ok(perf_event_array) = PerfEventArray::try_from(map) {
-            perf_event_arrays.push(perf_event_array); // this is step 1
-            let perf_event_array_buffer = Vec::new();
-            event_buffers.push(perf_event_array_buffer); //this is step 2 
+            map_manager.insert(name.clone(), (perf_event_array, Vec::new()));
+
+        //    perf_event_arrays.push(perf_event_array); // this is step 1
+        //    let perf_event_array_buffer = Vec::new();
+        //    event_buffers.push(perf_event_array_buffer); //this is step 2
         } else {
-            warn!("Map is not a PerfEventArray, skipping load");
+            warn!("Map {:?} is not a PerfEventArray, skipping load", &name);
         }
     }
 
     // fill the input buffers with data from the PerfEventArrays
-    let cpus = online_cpus().map_err(|e| anyhow::anyhow!("Error {:?}", e))?;
-
-    for (perf_evt_array, perf_evt_array_buffer) in
-        perf_event_arrays.iter_mut().zip(event_buffers.iter_mut())
-    {
-        for cpu_id in &cpus {
-            let single_buffer = perf_evt_array.open(*cpu_id, None)?;
-            perf_evt_array_buffer.push(single_buffer);
+    for cpu_id in online_cpus().map_err(|e| anyhow::anyhow!("Error {:?}", e))? {
+        for (name, (perf_evt_array, perf_evt_array_buffer)) in map_manager.iter_mut() {
+            let buf = perf_evt_array.open(cpu_id, None)?;
+            info!(
+                "Buffer created for map {:?} on cpu_id {:?}. Buffer size: {}",
+                name,
+                cpu_id,
+                std::mem::size_of_val(&buf)
+            );
+            perf_evt_array_buffer.push(buf);
         }
     }
 
     info!("Listening for events...");
 
-    let mut event_buffers = event_buffers.into_iter();
-    let perf_veth_buffer = event_buffers
-        .next()
+    // i need to use remove to move the values from the Map Manager to the the async tasks
+    let (perf_veth_array, perf_veth_buffers) = map_manager
+        .remove("veth_identity_map")
         .expect("Cannot create perf_veth buffer");
-    let perf_net_events_buffer = event_buffers
-        .next()
+    let (perf_net_events_array, perf_net_events_buffers) = map_manager
+        .remove("events_map")
         .expect("Cannot create perf_net_events buffer");
-    let tcp_registry_buffer = event_buffers
-        .next()
+    let (tcp_registry_array, tcp_registry_buffers) = map_manager
+        .remove("TcpPacketRegistry")
         .expect("Cannot create tcp_registry buffer");
 
     // init output buffers
-    let veth_buffers = vec![BytesMut::with_capacity(1024); online_cpus().iter().len()];
+    let veth_buffers = vec![BytesMut::with_capacity(10 * 1024); online_cpus().iter().len()];
     let events_buffers = vec![BytesMut::with_capacity(1024); online_cpus().iter().len()];
     let tcp_buffers = vec![BytesMut::with_capacity(1024); online_cpus().iter().len()];
 
@@ -254,26 +265,23 @@ async fn event_listener(
 
     // spawn async tasks
     let veth_events_displayer = tokio::spawn(async move {
-        //display_veth_events(bpf.clone(), perf_veth_buffer, veth_buffers, veth_link_ids).await;
-        read_perf_buffer(perf_veth_buffer, veth_buffers, helpers::BufferType::VethLog).await;
+        read_perf_buffer(perf_veth_buffers, veth_buffers, BufferType::VethLog).await;
     });
 
     let net_events_displayer = tokio::spawn(async move {
-        //display_events(perf_net_events_buffer, events_buffers).await;
         read_perf_buffer(
-            perf_net_events_buffer,
+            perf_net_events_buffers,
             events_buffers,
-            helpers::BufferType::PacketLog,
+            BufferType::PacketLog,
         )
         .await;
     });
 
     let tcp_registry_events_displayer: tokio::task::JoinHandle<()> = tokio::spawn(async move {
-        //display_tcp_registry_events(tcp_registry_buffer, tcp_buffers).await;
         read_perf_buffer(
-            tcp_registry_buffer,
+            tcp_registry_buffers,
             tcp_buffers,
-            helpers::BufferType::TcpPacketRegistry,
+            BufferType::TcpPacketRegistry,
         )
         .await;
     });