Feature/ebpf core

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

@@ -0,0 +1,54 @@
+use aya::{
+    maps::{
+        MapData,
+        perf::{PerfEventArrayBuffer},
+    }
+};
+
+use bytes::BytesMut;
+use std::{
+    sync::{
+        atomic::{AtomicBool, Ordering},
+    },
+};
+
+use tracing::{error, info};
+
+use crate::structs::NetworkMetrics;
+
+pub async fn display_metrics_map(
+    mut perf_buffers: Vec<PerfEventArrayBuffer<MapData>>,
+    running: AtomicBool,
+    mut buffers: Vec<BytesMut>,
+) {
+    while running.load(Ordering::SeqCst) {
+        for buf in perf_buffers.iter_mut() {
+            match buf.read_events(&mut buffers) {
+                std::result::Result::Ok(events) => {
+                    for i in 0..events.read {
+                        let data = &buffers[i];
+                        if data.len() >= std::mem::size_of::<NetworkMetrics>() {
+                            let net_metrics: NetworkMetrics =
+                                unsafe { std::ptr::read_unaligned(data.as_ptr() as *const _) };
+                            let sk_drop_count = net_metrics.sk_drops;
+                            let sk_err = net_metrics.sk_err;
+                            let sk_err_soft = net_metrics.sk_err_soft;
+                            let sk_backlog_len = net_metrics.sk_backlog_len;
+                            let sk_write_memory_queued = net_metrics.sk_write_memory_queued;
+                            let sk_ack_backlog = net_metrics.sk_ack_backlog;
+                            let sk_receive_buffer_size = net_metrics.sk_receive_buffer_size;
+                            info!(
+                                "sk_drops: {}, sk_err: {}, sk_err_soft: {}, sk_backlog_len: {}, sk_write_memory_queued: {}, sk_ack_backlog: {}, sk_receive_buffer_size: {}",
+                                sk_drop_count, sk_err, sk_err_soft, sk_backlog_len, sk_write_memory_queued, sk_ack_backlog, sk_receive_buffer_size
+                            );
+                        }
+                    }
+                }
+                Err(e) => {
+                    error!("Error reading events: {:?}", e);
+                }
+            }
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    }
+}

core/src/components/metrics/src/main.rs

@@ -1,39 +1,34 @@
 use aya::{
     Ebpf,
     maps::{
-        Map, MapData,
+         MapData,
         perf::{PerfEventArray, PerfEventArrayBuffer},
     },
-    programs::{KProbe, SchedClassifier, TcAttachType, tc::SchedClassifierLinkId},
+    programs::{KProbe},
     util::online_cpus,
 };
 
-use aya_log::EbpfLogger;
 use bytes::BytesMut;
 use std::{
     convert::TryInto,
     env, fs,
-    net::Ipv4Addr,
     path::Path,
     sync::{
-        Arc, Mutex,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicBool},
     },
 };
 
 use anyhow::{Context, Ok};
-use tokio::{signal, sync::broadcast::error};
-use tracing::{error, info, warn};
+use tokio::{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;
 
-#[repr(C)]
-#[derive(Clone, Copy)]
-struct NetworkMetrics {
-    src_addr: u32,
-}
+mod helpers;
+use crate::helpers::display_metrics_map;
+
+mod structs;
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
@@ -74,7 +69,7 @@ async fn main() -> Result<(), anyhow::Error> {
 
     match program.attach("tcp_identify_packet_loss", 0) {
         std::result::Result::Ok(_) => {
-            info!("program attacched successfully to the tcp_identify_packet_loss kprobe ")
+            info!("program attached successfully to the tcp_identify_packet_loss kprobe ")
         }
         Err(e) => error!(
             "An error occured while attaching the program to the tcp_identify_packet_loss kprobe. {:?} ",
@@ -97,36 +92,6 @@ async fn main() -> Result<(), anyhow::Error> {
         display_metrics_map(net_perf_buffer, running, buffers).await;
     });
 
-
     signal::ctrl_c().await?;
     Ok(())
 }
-
-pub async fn display_metrics_map(
-    mut perf_buffers: Vec<PerfEventArrayBuffer<MapData>>,
-    running: AtomicBool,
-    mut buffers: Vec<BytesMut>,
-) {
-    while running.load(Ordering::SeqCst) {
-        for buf in perf_buffers.iter_mut() {
-            match buf.read_events(&mut buffers) {
-                std::result::Result::Ok(events) => {
-                    for i in 0..events.read {
-                        let data = &buffers[i];
-                        if data.len() >= std::mem::size_of::<NetworkMetrics>() {
-                            let net_metrics: NetworkMetrics =
-                                unsafe { std::ptr::read(data.as_ptr() as *const _) };
-                            let src = Ipv4Addr::from(u32::from_be(net_metrics.src_addr));
-
-                            info!("Detected packet loss SRC: {}", src);
-                        }
-                    }
-                }
-                Err(e) => {
-                    error!("Error reading events: {:?}", e);
-                }
-            }
-        }
-        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-    }
-}

core/src/components/metrics/src/mod.rs

@@ -0,0 +1,3 @@
+mod structs;
+mod enums;
+mod helpers;

core/src/components/metrics/src/structs.rs

@@ -0,0 +1,13 @@
+
+
+#[repr(C)]
+#[derive(Clone, Copy)]
+pub struct NetworkMetrics {
+    pub sk_err: i32,          // Offset 284
+    pub sk_err_soft: i32,     // Offset 600
+    pub sk_backlog_len: i32,  // Offset 196
+    pub sk_write_memory_queued: i32,  // Offset 376
+    pub sk_receive_buffer_size: i32,       // Offset 244
+    pub sk_ack_backlog: u32,  // Offset 604
+    pub sk_drops: i32,        // Offset 136
+}