Improved metrics semantics. Bug fix

core/common/src/buffer_type.rs

@@ -95,7 +95,7 @@ pub const TASK_COMM_LEN: usize = 16; // linux/sched.h
 #[cfg(feature = "monitoring-structs")]
 #[repr(C, packed)]
 #[derive(Clone, Copy, Zeroable)]
-pub struct NetworkMetrics {
+pub struct PacketLossMetrics {
     pub tgid: u32,
     pub comm: [u8; TASK_COMM_LEN],
     pub ts_us: u64,
@@ -108,7 +108,7 @@ pub struct NetworkMetrics {
     pub sk_drops: i32,               // Offset 136
 }
 #[cfg(feature = "monitoring-structs")]
-unsafe impl aya::Pod for NetworkMetrics {}
+unsafe impl aya::Pod for PacketLossMetrics {}
 
 #[cfg(feature = "monitoring-structs")]
 #[repr(C, packed)]
@@ -208,7 +208,7 @@ pub enum BufferType {
     #[cfg(feature = "network-structs")]
     VethLog,
     #[cfg(feature = "monitoring-structs")]
-    NetworkMetrics,
+    PacketLossMetrics,
     #[cfg(feature = "monitoring-structs")]
     TimeStampMetrics,
     #[cfg(feature = "monitoring-structs")]
@@ -436,7 +436,7 @@ impl BufferType {
     ///
     /// Counterpart to [`read_network_buffer`] for the `time_stamp_events` map.
 
-    pub async fn read_network_metrics(
+    pub async fn read_packet_loss_metrics(
         buffers: &mut [BytesMut],
         tot_events: i32,
         offset: i32,
@@ -445,7 +445,7 @@ impl BufferType {
     ) {
         for i in offset..tot_events {
             let vec_bytes = &buffers[i as usize];
-            if vec_bytes.len() < std::mem::size_of::<NetworkMetrics>() {
+            if vec_bytes.len() < std::mem::size_of::<PacketLossMetrics>() {
                 error!(
                     "Corrupted Network Metrics data. Raw data: {}. Readed {} bytes expected {} bytes",
                     vec_bytes
@@ -454,28 +454,28 @@ impl BufferType {
                         .collect::<Vec<_>>()
                         .join(" "),
                     vec_bytes.len(),
-                    std::mem::size_of::<NetworkMetrics>()
+                    std::mem::size_of::<PacketLossMetrics>()
                 );
                 continue;
             }
-            if vec_bytes.len() >= std::mem::size_of::<NetworkMetrics>() {
-                let net_metrics: NetworkMetrics =
+            if vec_bytes.len() >= std::mem::size_of::<PacketLossMetrics>() {
+                let packet_loss: PacketLossMetrics =
                     unsafe { std::ptr::read_unaligned(vec_bytes.as_ptr() as *const _) };
 
                 match exporter {
-                    "otlp" => metrics.record_network_metrics(&net_metrics),
+                    "otlp" => metrics.record_packet_loss_metrics(&packet_loss),
                     _ => continue, // skip
                 }
-                let tgid = net_metrics.tgid;
-                let comm = String::from_utf8_lossy(&net_metrics.comm);
-                let ts_us = net_metrics.ts_us;
-                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;
+                let tgid = packet_loss.tgid;
+                let comm = String::from_utf8_lossy(&packet_loss.comm);
+                let ts_us = packet_loss.ts_us;
+                let sk_drop_count = packet_loss.sk_drops;
+                let sk_err = packet_loss.sk_err;
+                let sk_err_soft = packet_loss.sk_err_soft;
+                let sk_backlog_len = packet_loss.sk_backlog_len;
+                let sk_write_memory_queued = packet_loss.sk_write_memory_queued;
+                let sk_ack_backlog = packet_loss.sk_ack_backlog;
+                let sk_receive_buffer_size = packet_loss.sk_receive_buffer_size;
 
                 info!(
                     "tgid: {}, comm: {}, ts_us: {}, sk_drops: {}, sk_err: {}, sk_err_soft: {}, sk_backlog_len: {}, sk_write_memory_queued: {}, sk_ack_backlog: {}, sk_receive_buffer_size: {}",
@@ -811,15 +811,15 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
                                 .await
                             }
                             #[cfg(feature = "monitoring-structs")]
-                            BufferType::NetworkMetrics => {
-                                BufferType::read_network_metrics(
+                            BufferType::PacketLossMetrics => {
+                                BufferType::read_packet_loss_metrics(
                                     &mut buffers,
                                     tot_events,
                                     offset,
                                     "otlp",
                                     metrics
                                         .clone()
-                                        .expect("Metrics required for NetworkMetrics"),
+                                        .expect("Metrics required for PacketLossMetrics"),
                                 )
                                 .await
                             }
@@ -939,7 +939,7 @@ impl BufferSize {
             #[cfg(feature = "network-structs")]
             BufferSize::TcpEvents => std::mem::size_of::<TcpPacketRegistry>(),
             #[cfg(feature = "monitoring-structs")]
-            BufferSize::NetworkMetricsEvents => std::mem::size_of::<NetworkMetrics>(),
+            BufferSize::NetworkMetricsEvents => std::mem::size_of::<PacketLossMetrics>(),
             #[cfg(feature = "monitoring-structs")]
             BufferSize::TimeMetricsEvents => std::mem::size_of::<TimeStampMetrics>(),
             #[cfg(feature = "monitoring-structs")]

core/common/src/lib.rs

@@ -13,3 +13,4 @@ pub mod map_handlers;
 pub mod otel_metrics;
 #[cfg(feature = "program-handlers")]
 pub mod program_handlers;
+mod semantic;

core/common/src/otel_metrics.rs

@@ -12,9 +12,10 @@
 //!   telemetry by process.
 
 use crate::buffer_type::{
-    CpuFrequency, CpuIdle, MemAlloc, NetworkMetrics, SchedStatRuntime, SchedStatWait,
+    CpuFrequency, CpuIdle, MemAlloc, PacketLossMetrics, SchedStatRuntime, SchedStatWait,
     TimeStampMetrics,
 };
+use crate::semantic::Semantic;
 use opentelemetry::KeyValue;
 use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter};
 pub struct Metrics {
@@ -23,7 +24,7 @@ pub struct Metrics {
 
     /// Total number of network-related events produced by the `net_metrics`
     /// eBPF map.
-    pub packets_total: Counter<u64>,
+    pub socket_events_total: Counter<u64>,
 
     /// Observed socket drop count (`sk_drops`) from the kernel sock struct.
     pub sk_drops: Gauge<i64>,
@@ -33,11 +34,7 @@ pub struct Metrics {
 
     /// 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>,
+    pub tcp_latency_us: Histogram<u64>,
 
     /// Cpu bytes alloc total events
     pub cpu_bytes_alloc_events_total: Counter<u64>,
@@ -61,94 +58,99 @@ pub struct Metrics {
     pub cpu_idle_state: Gauge<i64>,
 }
 
+// TODO: add identity metrics with TC classifier packet counts
+// TODO: introduce a metric called total_tcp_packets total_udp_packets
 impl Metrics {
     /// Initialise all instruments backed by the supplied [`Meter`].
     pub fn new(meter: &Meter) -> Self {
         // total events
         let events_total = meter
-            .u64_counter("events_total")
-            .with_description("Total number of eBPF events processed")
+            .u64_counter(Semantic::TotalEvents.title())
+            .with_description(Semantic::TotalEvents.description())
+            .with_unit("1")
             .build();
 
-        // total packets
-        let packets_total = meter
-            .u64_counter("packets_total")
-            .with_description("Total number of network events processed")
+        // total socket events
+        let socket_events_total = meter
+            .u64_counter(Semantic::SocketTotalEvents.title())
+            .with_description(Semantic::SocketTotalEvents.description())
+            .with_unit("1")
             .build();
 
         // socket drops
         let sk_drops = meter
-            .i64_gauge("sk_drops")
-            .with_description("Socket drop count per event")
+            .i64_gauge(Semantic::SocketDrops.title())
+            .with_description(Semantic::SocketDrops.description())
+            .with_unit("1")
             .build();
 
         // socket errors
         let sk_err = meter
-            .i64_gauge("sk_err")
-            .with_description("Socket error count per event")
-            .build();
-
-        // delta microseconds
-        let delta_us = meter
-            .u64_histogram("delta_us")
-            .with_description("Distribution of delta_us values from timestamp events")
+            .i64_gauge(Semantic::SocketErrorsCount.title())
+            .with_description(Semantic::SocketErrorsCount.description())
+            .with_unit("1")
             .build();
 
-        // timestamp microseconds grouped
-        let ts_us = meter
-            .u64_histogram("ts_us")
-            .with_description("Distribution of timestamp values from eBPF events")
+        // tcp latency microseconds
+        let tcp_latency_us = meter
+            .u64_histogram(Semantic::Latency.title())
+            .with_description(Semantic::Latency.description())
+            .with_unit("us")
             .build();
 
         // cpu bytes alloc total events
         let cpu_bytes_alloc_events_total = meter
-            .u64_counter("bytes_alloc_events_total")
-            .with_description("Total bytes_alloc events occuring in the CPU")
+            .u64_counter(Semantic::PerCpuTotalEvents.title())
+            .with_description(Semantic::PerCpuTotalEvents.description())
+            .with_unit("1")
             .build();
 
         // cpu bytes allocation
         let cpu_bytes_alloc = meter
-            .i64_gauge("cpu_bytes_alloc")
-            .with_description("Cpu bytes allocation per event")
+            .i64_gauge(Semantic::PerCpuBytesAllocated.title())
+            .with_description(Semantic::PerCpuBytesAllocated.description())
+            .with_unit("bytes")
             .build();
 
         // memory allocation (mmap) events total
         let mem_alloc_events_total = meter
-            .u64_counter("mem_alloc_events_total")
-            .with_description("Total number of memory allocation (mmap) events processed")
+            .u64_counter(Semantic::TotalMemoryAllocationEvents.title())
+            .with_description(Semantic::TotalMemoryAllocationEvents.description())
+            .with_unit("1")
             .build();
 
         // bytes requested via mmap syscalls
         let enter_mem_alloc = meter
-            .i64_gauge("enter_mem_alloc")
-            .with_description("Bytes requested via mmap syscalls")
+            .i64_gauge(Semantic::RequestedMemoryBytes.title())
+            .with_description(Semantic::RequestedMemoryBytes.description())
+            .with_unit("bytes")
             .build();
 
         // scheduler wait time in nanoseconds
         let sched_stat_wait = meter
-            .i64_gauge("sched_stat_wait")
-            .with_description("Scheduler wait time in nanoseconds from sched_stat_wait")
+            .i64_gauge(Semantic::SchedulerWaitTime.title())
+            .with_description(Semantic::SchedulerWaitTime.description())
+            .with_unit("ns")
             .build();
 
         // scheduler runtime in nanoseconds
         let sched_stat_runtime = meter
-            .i64_gauge("sched_stat_runtime")
-            .with_description("Scheduler runtime in nanoseconds from sched_stat_runtime")
+            .i64_gauge(Semantic::SchedulerRuntime.title())
+            .with_description(Semantic::SchedulerRuntime.description())
+            .with_unit("ns")
             .build();
 
         // current CPU idle C-state per cpu_id
         let cpu_idle_state = meter
-            .i64_gauge("cpu_idle_state")
-            .with_description("Current CPU idle C-state per cpu_id, updated only on state change")
+            .i64_gauge(Semantic::CpuIdleState.title())
+            .with_description(Semantic::CpuIdleState.description())
             .build();
-
         Self {
             events_total,
-            packets_total,
+            socket_events_total,
             sk_drops,
             sk_err,
-            delta_us,
-            ts_us,
+            tcp_latency_us,
             cpu_bytes_alloc,
             cpu_bytes_alloc_events_total,
             mem_alloc_events_total,
@@ -159,17 +161,17 @@ impl Metrics {
         }
     }
 
-    /// Record a single [`NetworkMetrics`] event.
+    /// Record a single [`PacketLossMetrics`] event.
     ///
-    /// Increments `events_total` and `packets_total`, records `sk_drops` and
-    /// `sk_err` as gauges, and observes `ts_us` in the timestamp histogram.
+    /// Increments `events_total` and `socket_events_total`, records `sk_drops`
+    /// and `sk_err` as gauges.
     ///
     /// Every observation carries:
     ///
-    /// -`tgid` – task group ID.
+    /// - `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) {
+    pub fn record_packet_loss_metrics(&self, m: &PacketLossMetrics) {
         let comm = String::from_utf8_lossy(&m.comm);
         let comm_trimmed = comm.trim_end_matches('\0').to_string();
         let attrs = &[
@@ -178,19 +180,18 @@ impl Metrics {
         ];
 
         self.events_total.add(1, attrs);
-        self.packets_total.add(1, attrs);
+        self.socket_events_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.
+    /// Increments `events_total`, and records `delta_us` in the latency
+    /// histogram.
     ///
     /// Every observation carries `tgid` and `comm` (see
-    /// [`record_network_metrics`]).
+    /// [`record_packet_loss_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();
@@ -200,8 +201,7 @@ impl Metrics {
         ];
 
         self.events_total.add(1, attrs);
-        self.delta_us.record(m.delta_us, attrs);
-        self.ts_us.record(m.ts_us, attrs);
+        self.tcp_latency_us.record(m.delta_us, attrs);
     }
 
     pub fn record_cpu_bytes_alloc(&self, m: &CpuFrequency) {
@@ -231,6 +231,7 @@ impl Metrics {
             KeyValue::new("command", command),
         ];
 
+        self.events_total.add(1, attrs);
         self.mem_alloc_events_total.add(1, attrs);
         self.enter_mem_alloc.record(m.length as i64, attrs);
     }
@@ -247,6 +248,7 @@ impl Metrics {
             KeyValue::new("command", command),
         ];
 
+        self.events_total.add(1, attrs);
         self.sched_stat_wait.record(m.delay as i64, attrs);
     }
 
@@ -262,6 +264,7 @@ impl Metrics {
             KeyValue::new("command", command),
         ];
 
+        self.events_total.add(1, attrs);
         self.sched_stat_runtime.record(m.runtime as i64, attrs);
     }
 
@@ -272,6 +275,7 @@ impl Metrics {
     pub fn record_cpu_idle(&self, m: &CpuIdle) {
         let attrs = &[KeyValue::new("cpu_id", m.cpu_id as i64)];
 
+        self.events_total.add(1, attrs);
         self.cpu_idle_state.record(m.state as i64, attrs);
     }
 }