[#158]: moved map manager from identity to common crate to reuse the function in metrics service

Author: LorenzoTettamanti

core/common/Cargo.toml

@@ -29,5 +29,6 @@ bytemuck_derive = "1.10.2"
 map-handlers = []
 program-handlers = []
 network-structs = []
+monitoring-structs = []
 buffer-reader = []
 experimental = []

core/common/src/map_handlers.rs

@@ -154,3 +154,57 @@ pub fn load_perf_event_array_from_mapdata(
     })?;
     Ok(perf_event_array)
 }
+
+#[cfg(feature = "map-handlers")]
+pub fn map_manager(
+    maps: BpfMapsData,
+) -> Result<
+    std::collections::HashMap<
+        String,
+        (
+            aya::maps::PerfEventArray<aya::maps::MapData>,
+            Vec<aya::maps::perf::PerfEventArrayBuffer<aya::maps::MapData>>,
+        ),
+    >,
+    Error,
+> {
+    use aya::maps::PerfEventArray;
+    use aya::maps::{MapData, perf::PerfEventArrayBuffer};
+    use tracing::debug;
+
+    let mut map_manager = std::collections::HashMap::<
+        String, // this will store the bpf map name
+        (PerfEventArray<MapData>, Vec<PerfEventArrayBuffer<MapData>>), // this will manage the BPF_MAP_TYPE_PERF_EVENT_ARRAY and its buffer
+    >::new();
+
+    // map_manager creates an hashmap that contains:
+    // MAP NAME as String (KEY)
+    //
+    // VALUES (tuple)
+    // a PERF_EVENT_ARRAY
+    // a vector of PERF_EVENT_ARRAY_BUFFER
+    //
+    // the map manager helps the event listener to specifically call a map by its pinned name
+    // e.g. veth_identity_map and returns the associated PERF_EVENT_ARRAY and PERF_EVENT_ARRAY_BUFFERS (1 per CPU)
+    // also the map manager helps to write a more complete debug context by linking map names with arrays and buffers.
+    // actually i cannot return the extact information using only the Aya library
+
+    // create the PerfEventArrays and the buffers from the BpfMapsData Objects
+    for (map, name) in maps
+        .bpf_obj_map
+        .into_iter()
+        .zip(maps.bpf_obj_names.into_iter())
+    // zip two iterators at the same time for map object and map names
+    {
+        debug!("Debugging map type:{:?} for map name {:?}", map, &name);
+        info!("Creating PerfEventArray for map name {:?}", &name);
+
+        // 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) {
+            map_manager.insert(name.clone(), (perf_event_array, Vec::new()));
+        } else {
+            warn!("Map {:?} is not a PerfEventArray, skipping load", &name);
+        }
+    }
+    Ok(map_manager)
+}

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

@@ -14,19 +14,16 @@ mod service_discovery;
 use crate::helpers::{get_veth_channels, read_perf_buffer};
 use aya::{
     Ebpf,
-    maps::{
-        MapData,
-        perf::{PerfEventArray, PerfEventArrayBuffer},
-    },
     programs::{SchedClassifier, TcAttachType, tc::SchedClassifierLinkId},
     util::online_cpus,
 };
 
 #[cfg(feature = "experimental")]
 use crate::helpers::scan_cgroup_cronjob;
 
-use bytes::BytesMut;
-use cortexbrain_common::map_handlers::{init_bpf_maps, map_pinner, populate_blocklist};
+use cortexbrain_common::map_handlers::{
+    init_bpf_maps, map_manager, map_pinner, populate_blocklist,
+};
 use cortexbrain_common::program_handlers::load_program;
 use cortexbrain_common::{buffer_type::BufferType, map_handlers::BpfMapsData};
 use std::{
@@ -36,11 +33,11 @@ use std::{
 };
 
 use anyhow::{Context, Ok};
+use cortexbrain_common::buffer_type::BufferSize;
 use cortexbrain_common::{constants, logger};
-use tokio::{fs, signal};
-use tracing::{debug, error, info, warn};
-
 use std::collections::HashMap;
+use tokio::{fs, signal};
+use tracing::{error, info};
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
@@ -203,34 +200,11 @@ async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error> {
 
     //TODO: try to change from PerfEventArray to a RingBuffer data structure
 
-    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);
-
-        // 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) {
-            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", &name);
-        }
-    }
+    let mut maps = map_manager(bpf_maps)?;
 
     // fill the input buffers with data from the PerfEventArrays
     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() {
+        for (name, (perf_evt_array, perf_evt_array_buffer)) in maps.iter_mut() {
             let buf = perf_evt_array.open(cpu_id, None)?;
             info!(
                 "Buffer created for map {:?} on cpu_id {:?}. Buffer size: {}",
@@ -245,23 +219,20 @@ async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error> {
     info!("Listening for events...");
 
     // 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
+    let (perf_veth_array, perf_veth_buffers) = maps
         .remove("veth_identity_map")
         .expect("Cannot create perf_veth buffer");
-    let (perf_net_events_array, perf_net_events_buffers) = map_manager
+    let (perf_net_events_array, perf_net_events_buffers) = maps
         .remove("events_map")
         .expect("Cannot create perf_net_events buffer");
-    let (tcp_registry_array, tcp_registry_buffers) = map_manager
+    let (tcp_registry_array, tcp_registry_buffers) = maps
         .remove("TcpPacketRegistry")
         .expect("Cannot create tcp_registry buffer");
 
     // init output buffers
-    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()];
-
-    // init veth link ids
-    //let veth_link_ids = link_ids;
+    let veth_buffers = BufferSize::VethEvents.set_buffer();
+    let events_buffers = BufferSize::ClassifierNetEvents.set_buffer();
+    let tcp_buffers = BufferSize::TcpEvents.set_buffer();
 
     // spawn async tasks
     let veth_events_displayer = tokio::spawn(async move {

core/src/components/metrics/src/helpers.rs

@@ -1,5 +1,5 @@
 use aya::{
-    maps::{Map, MapData, PerfEventArray, perf::PerfEventArrayBuffer},
+    maps::{MapData, perf::PerfEventArrayBuffer},
     util::online_cpus,
 };
 
@@ -10,10 +10,14 @@ use std::sync::{
 };
 use tokio::signal;
 
-use tracing::{debug, error, info, warn};
+use tracing::{error, info};
 
-use crate::structs::NetworkMetrics;
-use crate::structs::TimeStampMetrics;
+use cortexbrain_common::map_handlers::map_manager;
+use cortexbrain_common::{
+    buffer_type::{BufferSize, BufferType},
+    buffer_type::{NetworkMetrics, TimeStampMetrics},
+    map_handlers::BpfMapsData,
+};
 
 pub async fn display_metrics_map(
     mut perf_buffers: Vec<PerfEventArrayBuffer<MapData>>,
@@ -119,50 +123,36 @@ pub async fn display_time_stamp_events_map(
     info!("Timestamp event listener stopped");
 }
 
-pub async fn event_listener(bpf_maps: Vec<Map>) -> Result<(), anyhow::Error> {
+pub async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error> {
     info!("Getting CPU count...");
 
-    let mut perf_event_arrays = Vec::new(); // contains a vector of PerfEventArrays
-    let mut event_buffers = Vec::new(); // contains a vector of buffers
-
-    info!("Creating perf buffers...");
-    for map in bpf_maps {
-        debug!("Debugging map type:{:?}", map);
-        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 
-        } else {
-            warn!("Map is not a PerfEventArray, skipping load");
-        }
-    }
+    let mut maps = map_manager(bpf_maps)?;
 
     let cpu_count = online_cpus().map_err(|e| anyhow::anyhow!("Error {:?}", e))?;
 
-    //info!("CPU count: {}", cpu_count);
-    for (perf_evt_array, perf_evt_array_buffer) in
-        perf_event_arrays.iter_mut().zip(event_buffers.iter_mut())
-    {
-        for cpu_id in &cpu_count {
-            let single_buffer = perf_evt_array.open(*cpu_id, None)?;
-            perf_evt_array_buffer.push(single_buffer);
+    for cpu_id in cpu_count {
+        for (name, (perf_event_array, perf_event_buffer)) in maps.iter_mut() {
+            let buf = perf_event_array.open(cpu_id, None)?;
+            perf_event_buffer.push(buf);
         }
     }
 
-    //info!("Opening perf buffers for {} CPUs...", cpu_count);
     info!("Perf buffers created successfully");
-    let mut event_buffers = event_buffers.into_iter();
 
-    let time_stamp_events_perf_buffer = event_buffers.next().expect("");
-    let net_perf_buffer = event_buffers.next().expect("");
+    let (time_stamp_events_array, time_stamp_events_perf_buffer) = maps
+        .remove("time_stamp_events")
+        .expect("Cannot create time_stamp_events_buffer");
+    let (net_perf_array, net_perf_buffer) = maps
+        .remove("net_metrics")
+        .expect("Cannot create net_perf_buffer");
 
     // Create shared running flags
     let net_metrics_running = Arc::new(AtomicBool::new(true));
     let time_stamp_events_running = Arc::new(AtomicBool::new(true));
 
     // Create proper sized buffers
-    let net_metrics_buffers = vec![BytesMut::with_capacity(1024); cpu_count.len()];
-    let time_stamp_events_buffers = vec![BytesMut::with_capacity(1024); cpu_count.len()];
+    let net_metrics_buffers = BufferSize::NetworkMetricsEvents.set_buffer();
+    let time_stamp_events_buffers = BufferSize::TimeMetricsEvents.set_buffer();
 
     // Clone for the signal handler
     let net_metrics_running_signal = net_metrics_running.clone();