refactor(optimization):Removed unnecessary mut in metrics/helpers.rs

Examples/run-with-docker/docker-compose.yaml

@@ -12,6 +12,7 @@ services:
       - OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
       - OTEL_EXPORTER_OTLP_PROTOCOL=grpc
       - LIBSSL_PATH=/host/usr/lib/x86_64-linux-gnu/libssl.so.3
+      - DEFAULT_MAP_READING_TIME=100
     volumes:
       - type: bind
         source: /sys/fs/bpf

core/common/src/constants.rs

@@ -10,3 +10,5 @@ pub const PIN_MAP_PATH: &str = "PIN_MAP_PATH";
 /// When set, this path is used directly. When unset, the library is located
 /// by searching the default system library directories.
 pub const LIBSSL_PATH: &str = "LIBSSL_PATH";
+
+pub const DEFAULT_MAP_READING_TIME : &str = "DEFAULT_MAP_READING_TIME";

core/common/src/consumer.rs

@@ -16,6 +16,7 @@ use crate::buffer_type::{
     CpuFrequency, CpuIdle, MemAlloc, PacketLossMetrics, SchedStatRuntime, SchedStatWait,
     TimeStampMetrics,
 };
+use crate::constants::DEFAULT_MAP_READING_TIME;
 #[cfg(feature = "network-structs")]
 use crate::buffer_type::{PacketLog, TcpPacketRegistry, VethLog};
 #[cfg(feature = "monitoring-structs")]
@@ -899,6 +900,6 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
                 }
             }
         }
-        tokio::time::sleep(std::time::Duration::from_millis(5000)).await;
+        tokio::time::sleep(std::time::Duration::from_millis(std::env::var(DEFAULT_MAP_READING_TIME).ok().and_then(|v| v.parse::<u64>().ok()).unwrap_or(5000))).await;
     }
 }

core/common/src/map_handlers.rs

@@ -8,8 +8,7 @@ use kube::{Api, Client};
 use std::net::Ipv4Addr;
 use std::path::PathBuf;
 use std::str::FromStr;
-use std::sync::Arc;
-use std::sync::Mutex;
+use std::sync::{Arc, Mutex};
 use tracing::warn;
 use tracing::{error, info};
 
@@ -23,28 +22,33 @@ use tracing::{error, info};
 pub struct BpfMapsData {
     pub bpf_obj_names: Vec<String>,
     pub bpf_obj_map: Vec<Map>,
+    pub bpf: Arc<Mutex<Ebpf>>,
 }
 
 #[cfg(feature = "map-handlers")]
 pub fn init_bpf_maps(
     bpf: Arc<Mutex<Ebpf>>,
     map_names: Vec<String>,
 ) -> Result<BpfMapsData, anyhow::Error> {
-    let mut bpf_new = bpf
-        .lock()
-        .map_err(|e| anyhow::anyhow!("Cannot get value from lock. Reason: {}", e))?;
-
     let mut maps = Vec::new(); // stores bpf_maps_objects
 
-    for name in &map_names {
-        let bpf_map_init = bpf_new
-            .take_map(&name)
-            .ok_or_else(|| anyhow::anyhow!("{} map not found", &name))?;
-        maps.push(bpf_map_init);
+    {
+        let mut bpf_new = bpf
+            .lock()
+            .map_err(|e| anyhow::anyhow!("Cannot get value from lock. Reason: {}", e))?;
+
+        for name in &map_names {
+            let bpf_map_init = bpf_new
+                .take_map(&name)
+                .ok_or_else(|| anyhow::anyhow!("{} map not found", &name))?;
+            maps.push(bpf_map_init);
+        }
     }
+
     Ok(BpfMapsData {
         bpf_obj_names: map_names.clone(),
         bpf_obj_map: maps,
+        bpf,
     })
 }
 
@@ -68,6 +72,7 @@ pub fn map_pinner(maps: BpfMapsData, path: &PathBuf) -> Result<BpfMapsData, Erro
     let mut owned_bpf_maps_data = BpfMapsData {
         bpf_obj_names: Vec::new(),
         bpf_obj_map: Vec::new(),
+        bpf: maps.bpf,
     };
     // an iterator that iterates two iterators simultaneously
     for (map_obj, name) in maps

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

@@ -1,19 +1,28 @@
 use anyhow::anyhow;
-use aya::util::online_cpus;
-use cortexbrain_common::map_handlers::map_manager;
-use cortexbrain_common::{buffer_type::BufferSize, map_handlers::BpfMapsData};
+use aya::{
+    Ebpf,
+    programs::{ProgramError, UProbe},
+    util::online_cpus,
+};
+use cortexbrain_common::{
+    buffer_type::BufferSize,
+    constants,
+    consumer::{Consumer, read_perf_buffer},
+    map_handlers::{BpfMapsData, map_manager},
+    otel_metrics::Metrics,
+    service_cache::ServiceCache,
+};
 use opentelemetry::metrics::Meter;
-use std::env;
-use std::path::Path;
-use std::sync::Arc;
-use std::time::Duration;
+use std::{
+    collections::HashSet,
+    env, fs,
+    os::unix::fs::MetadataExt,
+    path::Path,
+    sync::{Arc, Mutex, Once},
+    time::Duration,
+};
 use tokio::{signal, time};
-use tracing::{error, info};
-
-use cortexbrain_common::constants;
-use cortexbrain_common::consumer::{Consumer, read_perf_buffer};
-use cortexbrain_common::otel_metrics::Metrics;
-use cortexbrain_common::service_cache::ServiceCache;
+use tracing::{debug, error, info, warn};
 
 /// Locate the OpenSSL shared library used for the SSL uprobes.
 ///
@@ -76,6 +85,131 @@ pub async fn init_metrics_and_cache(
     Ok((metrics, cache))
 }
 
+const LIBSSL_NAMES: [&str; 4] = [
+    "libssl.so.3",
+    "libssl.so.1.1",
+    "libssl.so.1.0.0",
+    "libssl.so",
+];
+
+const LIBSSL_DIRS: [&str; 8] = [
+    "/usr/lib/x86_64-linux-gnu",
+    "/usr/lib64",
+    "/usr/lib/aarch64-linux-gnu",
+    "/lib/x86_64-linux-gnu",
+    "/lib64",
+    "/usr/lib",
+    "/lib",
+    "/usr/local/lib",
+];
+
+const SSL_UPROBES: [(&str, &str); 4] = [
+    ("ssl_read", "SSL_read"),
+    ("ssl_read_ret", "SSL_read"),
+    ("ssl_write", "SSL_write"),
+    ("ssl_write_ret", "SSL_write"),
+];
+
+static INVALID_LIBSSL_PATH: Once = Once::new();
+
+fn resolve_libssl_paths() -> Vec<String> {
+    let mut paths = Vec::new();
+
+    if let Ok(path) = env::var(constants::LIBSSL_PATH) {
+        if Path::new(&path).is_file() {
+            paths.push(path);
+        } else if !path.is_empty() {
+            INVALID_LIBSSL_PATH.call_once(|| warn!("Invalid LIBSSL_PATH: {}", path));
+        }
+    }
+
+    let Some(proc_dir) = ["/host/proc", "/proc"]
+        .into_iter()
+        .find(|dir| Path::new(dir).is_dir())
+    else {
+        return paths;
+    };
+    let Ok(entries) = fs::read_dir(proc_dir) else {
+        return paths;
+    };
+
+    let mut roots = HashSet::new();
+    for entry in entries.flatten() {
+        let Ok(pid) = entry.file_name().to_string_lossy().parse::<u32>() else {
+            continue;
+        };
+        let pid = pid.to_string();
+        let root = Path::new(proc_dir).join(&pid).join("root");
+        let Ok(metadata) = fs::metadata(&root) else {
+            continue;
+        };
+        if !roots.insert((metadata.dev(), metadata.ino())) {
+            continue;
+        }
+
+        for dir in LIBSSL_DIRS {
+            for name in LIBSSL_NAMES {
+                let path = root.join(dir.trim_start_matches('/')).join(name);
+                if path.is_file() {
+                    paths.push(path.to_string_lossy().into_owned());
+                }
+            }
+        }
+    }
+
+    paths
+}
+
+fn attach_uprobe(
+    bpf: &Arc<Mutex<Ebpf>>,
+    program_name: &str,
+    symbol: &str,
+    target: &str,
+) -> Result<(), anyhow::Error> {
+    let mut bpf = bpf
+        .lock()
+        .map_err(|e| anyhow::anyhow!("Cannot get value from lock. Reason: {}", e))?;
+    let program: &mut UProbe = bpf
+        .program_mut(program_name)
+        .ok_or_else(|| anyhow::anyhow!("Program {} not found", program_name))?
+        .try_into()
+        .map_err(|e| anyhow::anyhow!("Failed to convert program: {:?}", e))?;
+    match program.load() {
+        Ok(()) | Err(ProgramError::AlreadyLoaded) => {}
+        Err(e) => {
+            return Err(anyhow::anyhow!(
+                "Cannot load program: {}. Error: {}",
+                program_name,
+                e
+            ));
+        }
+    }
+    program
+        .attach(Some(symbol), 0, target, None)
+        .map_err(|e| anyhow::anyhow!("Failed to attach {} to {}: {:?}", program_name, target, e))?;
+    info!(
+        "{} program attached successfully to user space function {}",
+        program_name, target
+    );
+    Ok(())
+}
+
+fn attach_ssl_uprobes(bpf: &Arc<Mutex<Ebpf>>, attached: &mut HashSet<(u64, u64)>) {
+    for path in resolve_libssl_paths() {
+        let Ok(metadata) = fs::metadata(&path) else {
+            continue;
+        };
+        if !attached.insert((metadata.dev(), metadata.ino())) {
+            continue;
+        }
+        for (program, symbol) in SSL_UPROBES {
+            if let Err(e) = attach_uprobe(bpf, program, symbol, &path) {
+                debug!("SSL uprobe {} -> {}: {}", program, path, e);
+            }
+        }
+    }
+}
+
 /// Listen for eBPF perf-buffer events and record OpenTelemetry metrics.
 ///
 /// This function bridges the eBPF perf-buffer layer with the OpenTelemetry
@@ -97,6 +231,7 @@ pub async fn init_metrics_and_cache(
 pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), anyhow::Error> {
     info!("Getting CPU count...");
 
+    let bpf = Arc::clone(&bpf_maps.bpf);
     let mut maps = map_manager(bpf_maps)?;
 
     let cpu_count = online_cpus().map_err(|e| anyhow::anyhow!("Error {:?}", e))?;
@@ -183,11 +318,27 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
         })
     };
 
+    let mut attached = HashSet::new();
+    if let Ok(Some(path)) = resolve_libssl_path() {
+        if let Ok(metadata) = fs::metadata(&path) {
+            attached.insert((metadata.dev(), metadata.ino()));
+        }
+    }
+    let _ssl_handle = {
+        tokio::spawn(async move {
+            let mut interval = time::interval(Duration::from_secs(60));
+            loop {
+                interval.tick().await;
+                attach_ssl_uprobes(&bpf, &mut attached);
+            }
+        })
+    };
+
     let net_metrics_handle = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = net_perf_buffer;
-        let mut buffers = net_metrics_buffers;
+        let array_buffers = net_perf_buffer;
+        let buffers = net_metrics_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -203,8 +354,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let time_stamp_handle = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = time_stamp_events_perf_buffer;
-        let mut buffers = time_stamp_events_buffers;
+        let array_buffers = time_stamp_events_perf_buffer;
+        let buffers = time_stamp_events_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -220,8 +371,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let cpu_frequency_metrics = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = cpu_frequency_perf_buffer;
-        let mut buffers = cpu_frequency_events_buffers;
+        let array_buffers = cpu_frequency_perf_buffer;
+        let buffers = cpu_frequency_events_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -237,8 +388,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let cpu_idle_metrics = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = cpu_idle_perf_buffer;
-        let mut buffers = cpu_idle_buffers;
+        let array_buffers = cpu_idle_perf_buffer;
+        let buffers = cpu_idle_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -254,8 +405,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let mem_alloc_metrics = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = mem_alloc_perf_buffer;
-        let mut buffers = mem_alloc_buffers;
+        let array_buffers = mem_alloc_perf_buffer;
+        let buffers = mem_alloc_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -272,8 +423,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
 
-        let mut array_buffers = sched_stat_wait_perf_buffer;
-        let mut buffers = sched_stat_wait_buffers;
+        let array_buffers = sched_stat_wait_perf_buffer;
+        let buffers = sched_stat_wait_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -289,8 +440,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let sched_stat_runtime_metrics = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = sched_stat_runtime_perf_buffer;
-        let mut buffers = sched_stat_runtime_buffers;
+        let array_buffers = sched_stat_runtime_perf_buffer;
+        let buffers = sched_stat_runtime_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,
@@ -306,8 +457,8 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a
     let ssl_events_metrics = {
         let metrics = Arc::clone(&metrics);
         let service_cache = Arc::clone(&cache);
-        let mut array_buffers = ssl_events_perf_buffer;
-        let mut buffers = ssl_events_buffers;
+        let array_buffers = ssl_events_perf_buffer;
+        let buffers = ssl_events_buffers;
         tokio::spawn(async move {
             read_perf_buffer(
                 array_buffers,

core/src/components/metrics_tracer/Cargo.toml

@@ -8,9 +8,6 @@ aya-ebpf = { git = "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/aya-rs/aya" }
 aya-log-ebpf = { git = "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/aya-rs/aya" }
 bytemuck = {version ="1.23.0",features = ["derive"]}
 
-[build-dependencies]
-which = { version = "7.0.0", default-features = false }
-
 [[bin]]
 name = "metrics_tracer"
 path = "src/main.rs"