[#175]: implemented the opentelemetry metrics export using the function in the common crate

Author: LorenzoTettamanti

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

@@ -1,14 +1,34 @@
 use anyhow::anyhow;
 use aya::util::online_cpus;
 use cortexbrain_common::map_handlers::map_manager;
-use cortexbrain_common::{
-    buffer_type::{BufferSize, BufferType, read_perf_buffer},
-    map_handlers::BpfMapsData,
-};
+use cortexbrain_common::{buffer_type::BufferSize, map_handlers::BpfMapsData};
+use opentelemetry::metrics::Meter;
+use std::sync::Arc;
 use tokio::signal;
 use tracing::{error, info};
 
-pub async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error> {
+use cortexbrain_common::buffer_type::{BufferType, read_perf_buffer};
+use cortexbrain_common::otel_metrics::Metrics;
+
+/// Listen for eBPF perf-buffer events and record OpenTelemetry metrics.
+///
+/// This function bridges the eBPF perf-buffer layer with the OpenTelemetry
+/// metrics pipeline.  It opens per-CPU buffers for the two maps of interest
+/// (`net_metrics` and `time_stamp_events`), spawns asynchronous consumers,
+/// and parks until a `Ctrl-C` signal is received or one of the consumers
+/// terminates.
+///
+/// # Arguments
+///
+/// -`bpf_maps` – handles for the pinned BPF maps produced by
+///   [`cortexbrain_common::map_handlers::map_pinner`].
+/// - `meter`    – an initialised OpenTelemetry [`Meter`].
+///
+/// # Errors
+///
+/// Returns `Err` if the map manager or CPU enumeration fails.
+///
+pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), anyhow::Error> {
     info!("Getting CPU count...");
 
     let mut maps = map_manager(bpf_maps)?;
@@ -35,48 +55,63 @@ pub async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error>
 
     info!("Perf buffers created successfully");
 
-    let (time_stamp_events_array, time_stamp_events_perf_buffer) = maps
+    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
+    let (_net_perf_array, net_perf_buffer) = maps
         .remove("net_metrics")
         .expect("Cannot create net_perf_buffer");
 
-    // Create proper sized buffers
+    // Allocate byte-buffers sized for each structure type
     let net_metrics_buffers = BufferSize::NetworkMetricsEvents.set_buffer();
     let time_stamp_events_buffers = BufferSize::TimeMetricsEvents.set_buffer();
 
+    let metrics = Arc::new(Metrics::new(&meter));
+
     info!("Starting event listener tasks...");
-    let metrics_map_displayer = tokio::spawn(async move {
-        read_perf_buffer(
-            net_perf_buffer,
-            net_metrics_buffers,
-            BufferType::NetworkMetrics,
-        )
-        .await;
-    });
-
-    let time_stamp_events_displayer = tokio::spawn(async move {
-        read_perf_buffer(
-            time_stamp_events_perf_buffer,
-            time_stamp_events_buffers,
-            BufferType::TimeStampMetrics,
-        )
-        .await;
-    });
+
+    let net_metrics_handle = {
+        let metrics = Arc::clone(&metrics);
+        let mut array_buffers = net_perf_buffer;
+        let mut buffers = net_metrics_buffers;
+        tokio::spawn(async move {
+            read_perf_buffer(
+                array_buffers,
+                buffers,
+                BufferType::NetworkMetrics,
+                Some(metrics),
+            )
+            .await;
+        })
+    };
+
+    let time_stamp_handle = {
+        let metrics = Arc::clone(&metrics);
+        let mut array_buffers = time_stamp_events_perf_buffer;
+        let mut buffers = time_stamp_events_buffers;
+        tokio::spawn(async move {
+            read_perf_buffer(
+                array_buffers,
+                buffers,
+                BufferType::TimeStampMetrics,
+                Some(metrics),
+            )
+            .await;
+        })
+    };
 
     info!("Event listeners started, entering main loop...");
 
     tokio::select! {
-        result = metrics_map_displayer => {
+        result = net_metrics_handle => {
             if let Err(e) = result {
-                error!("Metrics map displayer task failed: {:?}", e);
+                error!("Network metrics task failed: {:?}", e);
             }
         }
 
-        result = time_stamp_events_displayer => {
+        result = time_stamp_handle => {
             if let Err(e) = result {
-                error!("Time stamp events displayer task failed: {:?}", e);
+                error!("Timestamp events task failed: {:?}", e);
             }
         }
 
@@ -85,6 +120,5 @@ pub async fn event_listener(bpf_maps: BpfMapsData) -> Result<(), anyhow::Error>
         }
     }
 
-    // return success
     Ok(())
 }

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

@@ -1,14 +1,26 @@
-use anyhow::{Context, Ok};
+//! CortexBrain metrics service – eBPF-based telemetry with OpenTelemetry export.
+//!
+//! This binary is the node-level metrics agent for CortexBrain.  It:
+//!
+//! 1. Initialises an OpenTelemetry metrics pipeline (OTLP / gRPC).
+//! 2. Loads a compiled eBPF object and pins its maps to the BPF filesystem.
+//! 3. Attaches a set of kernel kprobe programs.
+//! 4. Starts asynchronous consumers that read per-CPU perf buffers and
+//!    emit OpenTelemetry instruments for every event.
+//! 5. Blocks until `Ctrl-C` is received, then shuts down cleanly.
+
+use anyhow::Context;
 use aya::Ebpf;
 use std::{
     env, fs,
     path::Path,
     sync::{Arc, Mutex},
 };
 use tracing::{error, info};
-
 mod helpers;
+mod otel_init;
 use crate::helpers::event_listener;
+use crate::otel_init::{init_opentelemetry, shutdown_opentelemetry};
 
 use cortexbrain_common::{
     constants,
@@ -19,12 +31,14 @@ use cortexbrain_common::{
 
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
-    //init tracing subscriber
-    let otlp_provider = otlp_logger_init("metrics-service".to_string());
+    let _otlp_log_provider = otlp_logger_init("metrics-service".to_string());
 
     info!("Starting metrics service...");
     info!("fetching data");
 
+    let meter =
+        init_opentelemetry().context("Failed to initialise OpenTelemetry metrics pipeline")?;
+
     let bpf_path =
         env::var(constants::BPF_PATH).context("BPF_PATH environment variable required")?;
     let data = fs::read(Path::new(&bpf_path)).context("Failed to load file from path")?;
@@ -35,54 +49,58 @@ async fn main() -> Result<(), anyhow::Error> {
 
     info!("Running Ebpf logger");
     info!("loading programs");
-    let bpf_map_save_path = std::env::var(constants::PIN_MAP_PATH)
-        .context("PIN_MAP_PATH environment variable required")?;
+
+    let bpf_map_save_path =
+        env::var(constants::PIN_MAP_PATH).context("PIN_MAP_PATH environment variable required")?;
 
     let map_data = vec!["time_stamp_events".to_string(), "net_metrics".to_string()];
 
     match init_bpf_maps(bpf.clone(), map_data) {
-        std::result::Result::Ok(bpf_maps) => {
+        Ok(bpf_maps) => {
             info!("BPF maps loaded successfully");
             let pin_path = std::path::PathBuf::from(&bpf_map_save_path);
             info!("About to call map_pinner with path: {:?}", pin_path);
+
             match map_pinner(bpf_maps, &pin_path) {
-                std::result::Result::Ok(maps) => {
+                Ok(maps) => {
                     info!("BPF maps pinned successfully to {}", bpf_map_save_path);
 
                     {
                         load_program(bpf.clone(), "metrics_tracer", "tcp_identify_packet_loss")
                             .context(
-                                "An error occured during the execution of load_program function",
+                                "An error occurred during the execution of load_program function",
                             )?;
 
-                        load_program(tcp_bpf,"tcp_v4_connect","tcp_v4_connect")
-                        .context("An error occured during the execution of load_and_attach_tcp_programs function")?;
-                        load_program(tcp_v6_bpf,"tcp_v6_connect","tcp_v6_connect")
-                        .context("An error occured during the execution of load_and_attach_tcp_programs function")?;
+                        load_program(tcp_bpf, "tcp_v4_connect", "tcp_v4_connect")
+                            .context("An error occurred during the execution of load_and_attach_tcp_programs function")?;
+
+                        load_program(tcp_v6_bpf, "tcp_v6_connect", "tcp_v6_connect")
+                            .context("An error occurred during the execution of load_and_attach_tcp_programs function")?;
 
                         load_program(
                             tcp_rev_bpf,
                             "tcp_rcv_state_process",
                             "tcp_rcv_state_process",
                         )
                         .context(
-                            "An error occured during the execution of load_program function",
+                            "An error occurred during the execution of load_program function",
                         )?;
                     }
-                    event_listener(maps).await?;
+
+                    // Hand off to the async event consumer
+                    event_listener(maps, meter).await
                 }
                 Err(e) => {
                     error!("Error pinning BPF maps: {:?}", e);
-                    return Err(e);
+                    shutdown_opentelemetry();
+                    Err(e)
                 }
             }
         }
         Err(e) => {
             error!("Error initializing BPF maps: {:?}", e);
-            let _ = otlp_provider.shutdown();
-            return Err(e);
+            shutdown_opentelemetry();
+            Err(e)
         }
     }
-
-    Ok(())
 }

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

@@ -1,3 +1,2 @@
-mod structs;
-mod enums;
-mod helpers;
+mod helpers;
+mod otel_init;