[#175]: added modules to initialize the Metrics exporter using opentelemetry sdk. added srtruct Metrics to group all the metrics in one place. added auxiliary functions record_network_metrics and record_timestamp_metrics. added exporter setting in buffer_type/read_network_metrics and buffer_type/read_timestamp_metrics

Author: LorenzoTettamanti

core/common/src/buffer_type.rs

@@ -1,9 +1,14 @@
+#[cfg(feature = "monitoring-structs")]
+use crate::otel_metrics::Metrics;
 #[cfg(feature = "buffer-reader")]
 use aya::maps::{MapData, PerfEventArray};
 use aya::{maps::perf::PerfEventArrayBuffer, util::online_cpus};
 use bytemuck_derive::Zeroable;
 use bytes::BytesMut;
 use std::net::Ipv4Addr;
+#[cfg(feature = "buffer-reader")]
+#[cfg(feature = "monitoring-structs")]
+use std::sync::Arc;
 use tracing::{error, info, warn};
 
 //
@@ -342,7 +347,39 @@ impl BufferType {
         }
     }
     #[cfg(feature = "monitoring-structs")]
-    pub async fn read_network_metrics(buffers: &mut [BytesMut], tot_events: i32, offset: i32) {
+    /// Continuously read [`NetworkMetrics`] events and record OpenTelemetry
+    /// observations.
+    ///
+    /// This helper mirrors the core behaviour of
+    /// [`cortexbrain_common::buffer_type::read_perf_buffer`] but adds the OTel
+    /// instrumentation layer.
+    ///
+    /// # Loop
+    ///
+    /// 1. For every CPU buffer call `read_events`.
+    /// 2. Parse each raw [`BytesMut`] into [`NetworkMetrics`] using an
+    ///    unaligned read (the struct is `#[repr(C, packed)]` and `Pod`).
+    /// 3. Call [`Metrics::record_network_metrics`].
+    /// 4. Retain the legacy `tracing::info!` log for human-readable local output.
+    /// 5. Sleep 100 ms between polls.
+    ///
+    /// # Safety
+    ///
+    /// `std::ptr::read_unaligned` is safe here because the eBPF program writes
+    /// exactly the `NetworkMetrics` layout into the ring buffer and the struct
+    /// implements [`aya::Pod`].
+    /// Continuously read [`TimeStampMetrics`] events and record OpenTelemetry
+    /// observations.
+    ///
+    /// Counterpart to [`read_network_buffer`] for the `time_stamp_events` map.
+
+    pub async fn read_network_metrics(
+        buffers: &mut [BytesMut],
+        tot_events: i32,
+        offset: i32,
+        exporter: &str,
+        metrics: Arc<Metrics>,
+    ) {
         for i in offset..tot_events {
             let vec_bytes = &buffers[i as usize];
             if vec_bytes.len() < std::mem::size_of::<NetworkMetrics>() {
@@ -361,6 +398,11 @@ impl BufferType {
             if vec_bytes.len() >= std::mem::size_of::<NetworkMetrics>() {
                 let net_metrics: NetworkMetrics =
                     unsafe { std::ptr::read_unaligned(vec_bytes.as_ptr() as *const _) };
+
+                match exporter {
+                    "otlp" => metrics.record_network_metrics(&net_metrics),
+                    _ => continue, // skip
+                }
                 let tgid = net_metrics.tgid;
                 let comm = String::from_utf8_lossy(&net_metrics.comm);
                 let ts_us = net_metrics.ts_us;
@@ -389,7 +431,13 @@ impl BufferType {
         }
     }
     #[cfg(feature = "monitoring-structs")]
-    pub async fn read_timestamp_metrics(buffers: &mut [BytesMut], tot_events: i32, offset: i32) {
+    pub async fn read_timestamp_metrics(
+        buffers: &mut [BytesMut],
+        tot_events: i32,
+        offset: i32,
+        exporter: &str,
+        metrics: Arc<Metrics>,
+    ) {
         for i in offset..tot_events {
             let vec_bytes = &buffers[i as usize];
             if vec_bytes.len() < std::mem::size_of::<TimeStampMetrics>() {
@@ -408,6 +456,12 @@ impl BufferType {
             if vec_bytes.len() >= std::mem::size_of::<TimeStampMetrics>() {
                 let time_stamp_event: TimeStampMetrics =
                     unsafe { std::ptr::read_unaligned(vec_bytes.as_ptr() as *const _) };
+
+                match exporter {
+                    "otlp" => metrics.record_timestamp_metrics(&time_stamp_event),
+                    _ => continue,
+                }
+
                 let delta_us = time_stamp_event.delta_us;
                 let ts_us = time_stamp_event.ts_us;
                 let tgid = time_stamp_event.tgid;
@@ -431,6 +485,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
     mut array_buffers: Vec<PerfEventArrayBuffer<T>>,
     mut buffers: Vec<bytes::BytesMut>,
     buffer_type: BufferType,
+    #[cfg(feature = "monitoring-structs")] metrics: Option<Arc<Metrics>>,
 ) {
     // loop over the buffers
     loop {
@@ -469,13 +524,29 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
                             }
                             #[cfg(feature = "monitoring-structs")]
                             BufferType::NetworkMetrics => {
-                                BufferType::read_network_metrics(&mut buffers, tot_events, offset)
-                                    .await
+                                BufferType::read_network_metrics(
+                                    &mut buffers,
+                                    tot_events,
+                                    offset,
+                                    "otlp",
+                                    metrics
+                                        .clone()
+                                        .expect("Metrics required for NetworkMetrics"),
+                                )
+                                .await
                             }
                             #[cfg(feature = "monitoring-structs")]
                             BufferType::TimeStampMetrics => {
-                                BufferType::read_timestamp_metrics(&mut buffers, tot_events, offset)
-                                    .await
+                                BufferType::read_timestamp_metrics(
+                                    &mut buffers,
+                                    tot_events,
+                                    offset,
+                                    "otlp",
+                                    metrics
+                                        .clone()
+                                        .expect("Metric required for TimeStampMetrics"),
+                                )
+                                .await
                             }
                         }
                     }

core/common/src/lib.rs

@@ -1,13 +1,15 @@
 #[cfg(any(
     feature = "buffer-reader",
     feature = "network-structs",
-    feature = "monitoring-structs"
+    feature = "monitoring-structs",
 ))]
 pub mod buffer_type;
 pub mod constants;
 pub mod formatters;
 pub mod logger;
 #[cfg(feature = "map-handlers")]
 pub mod map_handlers;
+#[cfg(feature = "monitoring-structs")]
+pub mod otel_metrics;
 #[cfg(feature = "program-handlers")]
 pub mod program_handlers;

core/common/src/otel_metrics.rs

@@ -0,0 +1,133 @@
+//! OpenTelemetry metric instruments for eBPF perf-buffer events.
+//!
+//! This module centralises every [`Meter`]-backed instrument that the
+//! `metrics` crate uses to observe raw eBPF events.  It provides a single
+//! [`Metrics`] handle that is cheap to [`Arc`]-clone and safe to use from
+//! multiple asynchronous tasks concurrently.
+//!
+//!  - An [`Arc<Metrics>`] is moved into each Tokio
+//!   task that reads a perf buffer.  All instrument operations are lock-free.
+//!  - Every observation is tagged with `tgid` and `comm`
+//!   extracted from the eBPF struct, allowing downstream collectors to group
+//!   telemetry by process.
+
+use crate::buffer_type::{NetworkMetrics, TimeStampMetrics};
+use opentelemetry::KeyValue;
+use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter};
+pub struct Metrics {
+    /// Total number of eBPF events processed across all perf buffers.
+    pub events_total: Counter<u64>,
+
+    /// Total number of network-related events produced by the `net_metrics`
+    /// eBPF map.
+    pub packets_total: Counter<u64>,
+
+    /// Observed socket drop count (`sk_drops`) from the kernel sock struct.
+    pub sk_drops: Gauge<i64>,
+
+    /// Observed socket error count (`sk_err`) from the kernel sock struct.
+    pub sk_err: Gauge<i64>,
+
+    /// Histogram of `delta_us` values supplied by the `time_stamp_events`
+    /// perf buffer.
+    pub delta_us: Histogram<u64>,
+
+    /// Histogram of `ts_us` values seen in both `net_metrics` and
+    /// `time_stamp_events`.
+    pub ts_us: Histogram<u64>,
+}
+
+impl Metrics {
+    /// Initialise all instruments backed by the supplied [`Meter`].
+    pub fn new(meter: &Meter) -> Self {
+        // total events
+        let events_total = meter
+            .u64_counter("cortexbrain_events_total")
+            .with_description("Total number of eBPF events processed")
+            .build();
+
+        // total packets
+        let packets_total = meter
+            .u64_counter("cortexbrain_packets_total")
+            .with_description("Total number of network events processed")
+            .build();
+
+        // socket drops
+        let sk_drops = meter
+            .i64_gauge("cortexbrain_sk_drops")
+            .with_description("Socket drop count per event")
+            .build();
+
+        // socket errors
+        let sk_err = meter
+            .i64_gauge("cortexbrain_sk_err")
+            .with_description("Socket error count per event")
+            .build();
+
+        // delta microseconds
+        let delta_us = meter
+            .u64_histogram("cortexbrain_delta_us")
+            .with_description("Distribution of delta_us values from timestamp events")
+            .build();
+
+        // timestamp microseconds grouped
+        let ts_us = meter
+            .u64_histogram("cortexbrain_ts_us")
+            .with_description("Distribution of timestamp values from eBPF events")
+            .build();
+
+        Self {
+            events_total,
+            packets_total,
+            sk_drops,
+            sk_err,
+            delta_us,
+            ts_us,
+        }
+    }
+
+    /// Record a single [`NetworkMetrics`] event.
+    ///
+    /// Increments `events_total` and `packets_total`, records `sk_drops` and
+    /// `sk_err` as gauges, and observes `ts_us` in the timestamp histogram.
+    ///
+    /// Every observation carries:
+    ///
+    /// -`tgid` – task group ID.
+    /// - `comm` – command name (null-terminated bytes converted to a UTF-8
+    ///   string and trimmed).
+    pub fn record_network_metrics(&self, m: &NetworkMetrics) {
+        let comm = String::from_utf8_lossy(&m.comm);
+        let comm_trimmed = comm.trim_end_matches('\0').to_string();
+        let attrs = &[
+            KeyValue::new("tgid", m.tgid as i64),
+            KeyValue::new("comm", comm_trimmed),
+        ];
+
+        self.events_total.add(1, attrs);
+        self.packets_total.add(1, attrs);
+        self.sk_drops.record(m.sk_drops as i64, attrs);
+        self.sk_err.record(m.sk_err as i64, attrs);
+        self.ts_us.record(m.ts_us, attrs);
+    }
+
+    /// Record a single [`TimeStampMetrics`] event.
+    ///
+    /// Increments `events_total`, and records `delta_us` and `ts_us` in their
+    /// respective histograms.
+    ///
+    /// Every observation carries `tgid` and `comm` (see
+    /// [`record_network_metrics`]).
+    pub fn record_timestamp_metrics(&self, m: &TimeStampMetrics) {
+        let comm = String::from_utf8_lossy(&m.comm);
+        let comm_trimmed = comm.trim_end_matches('\0').to_string();
+        let attrs = &[
+            KeyValue::new("tgid", m.tgid as i64),
+            KeyValue::new("comm", comm_trimmed),
+        ];
+
+        self.events_total.add(1, attrs);
+        self.delta_us.record(m.delta_us, attrs);
+        self.ts_us.record(m.ts_us, attrs);
+    }
+}