feat(metrics): added pod namespace in metadata builder

core/common/src/metadata.rs

@@ -8,9 +8,6 @@ use tokio::sync::RwLock;
 use tracing::{debug, info};
 
 use crate::service_cache::ServiceCache;
-
-// TODO: add containerd tracing and mutual exclusion for docker environments
-
 /// Detected container runtime.
 #[derive(Debug, Clone, PartialEq)]
 pub enum ContainerRuntime {
@@ -30,7 +27,7 @@ pub struct Metadata {
     pub container_name: Option<String>,
     pub container_id: Option<String>,
     pub pod_name: Option<String>,
-    pub namespace: Option<String>, // TODO: implement this one
+    pub namespace: Option<String>,
 }
 
 impl Metadata {
@@ -105,8 +102,9 @@ impl Metadata {
             // acquire cache lock
             let cache_lock = cache.read().await;
             match cache_lock.get_from_cache(&id).await {
-                Some(name) => {
-                    self.pod_name = Some(name);
+                Some(cached_data) => {
+                    self.pod_name = Some(cached_data[0].clone());
+                    self.namespace = Some(cached_data[1].clone());
                     return true;
                 }
                 None => {
@@ -150,10 +148,14 @@ impl Metadata {
 
                 let cache_lock = cache.read().await;
                 match cache_lock.get_from_cache(&id).await {
-                    Some(name) => self.container_name = Some(name),
+                    Some(cached_data) => {
+                        self.container_name = Some(cached_data[0].clone());
+                        self.namespace = Some(cached_data[1].clone());
+                    }
                     None => match resolve_docker_name(&id) {
                         Some(name) => {
                             self.container_name = Some(name);
+                            self.namespace = None;
                             return true;
                         }
                         None => return false,

core/common/src/otel_metrics.rs

@@ -14,7 +14,7 @@ use crate::buffer_type::{
     CpuFrequency, CpuIdle, MemAlloc, PacketLossMetrics, SchedStatRuntime, SchedStatWait, SslEvent,
     TimeStampMetrics,
 };
-use crate::metadata::{Metadata};
+use crate::metadata::Metadata;
 use crate::semantic::Semantic;
 use opentelemetry::KeyValue;
 use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter};
@@ -243,7 +243,10 @@ impl Metrics {
     /// and `sk_err` as gauges.
     pub fn record_packet_loss_metrics(&self, m: &PacketLossMetrics, metadata: &Metadata) {
         let attrs = &self.build_attrs(metadata);
-        debug!("Building attributes for packet loss metrics. Attributes: {:?}",attrs);
+        debug!(
+            "Building attributes for packet loss metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, attrs);
         self.socket_events_total.add(1, attrs);
@@ -257,7 +260,10 @@ impl Metrics {
     /// histogram.
     pub fn record_timestamp_metrics(&self, m: &TimeStampMetrics, metadata: &Metadata) {
         let attrs = &self.build_attrs(metadata);
-        debug!("Building attributes for latency metrics. Attributes: {:?}",attrs);
+        debug!(
+            "Building attributes for latency metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, attrs);
         self.tcp_latency_us.record(m.delta_us, attrs);
@@ -267,7 +273,10 @@ impl Metrics {
     pub fn record_cpu_bytes_alloc(&self, m: &CpuFrequency, metadata: &Metadata) {
         let bytes_allocated = m.bytes_alloc;
         let attrs = &self.build_attrs(metadata);
-        debug!("Building attributes for CPU metrics. Attributes: {:?}",attrs);
+        debug!(
+            "Building attributes for CPU metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.cpu_bytes_alloc_events_total.add(1, attrs);
         self.cpu_bytes_alloc.record(bytes_allocated as i64, attrs);
@@ -281,7 +290,10 @@ impl Metrics {
     /// events.
     pub fn record_enter_mem_alloc(&self, m: &MemAlloc, metadata: &Metadata) {
         let attrs = &self.build_attrs(metadata);
-        debug!("Building attributes for Mem alloc metrics. Attributes: {:?}",attrs);
+        debug!(
+            "Building attributes for Mem alloc metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, attrs);
         self.mem_alloc_events_total.add(1, attrs);
@@ -295,8 +307,11 @@ impl Metrics {
     /// histogram.
     pub fn record_sched_stat_wait(&self, m: &SchedStatWait, metadata: &Metadata) {
         let attrs = &self.build_attrs(metadata);
-        debug!("Building attributes for Sched stat wait metrics. Attributes: {:?}",attrs);
-
+        debug!(
+            "Building attributes for Sched stat wait metrics. Attributes: {:?}",
+            attrs
+        );
+
         self.events_total.add(1, attrs);
         self.sched_stat_wait.record(m.delay as i64, attrs);
         self.sched_stat_wait_distribution.record(m.delay, attrs);
@@ -310,8 +325,10 @@ impl Metrics {
     pub fn record_sched_stat_runtime(&self, m: &SchedStatRuntime, metadata: &Metadata) {
         let attrs = &self.build_attrs(metadata);
 
-        debug!("Building attributes for sched stat runtime metrics. Attributes: {:?}",attrs);
-
+        debug!(
+            "Building attributes for sched stat runtime metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, attrs);
         self.sched_stat_runtime.record(m.runtime as i64, attrs);
@@ -326,7 +343,10 @@ impl Metrics {
     pub fn record_cpu_idle(&self, m: &CpuIdle, metadata: &Metadata) {
         let mut attrs = self.build_attrs(metadata);
 
-        debug!("Building attributes for record cpu idle metrics. Attributes: {:?}",attrs);
+        debug!(
+            "Building attributes for record cpu idle metrics. Attributes: {:?}",
+            attrs
+        );
 
         attrs.push(KeyValue::new("cpu_id", m.cpu_id as i64));
 
@@ -336,16 +356,20 @@ impl Metrics {
 
     pub fn record_ssl_read_bytes(&self, m: &SslEvent, metadata: &Metadata) {
         let attrs = self.build_attrs(metadata);
-        debug!("Building attributes for SSL read metrics. Attributes: {:?}",attrs);
-
+        debug!(
+            "Building attributes for SSL read metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, &attrs);
         self.ssl_read_bytes.record(m.size as i64, &attrs);
     }
     pub fn record_ssl_write_bytes(&self, m: &SslEvent, metadata: &Metadata) {
         let attrs = self.build_attrs(metadata);
-        debug!("Building attributes for SSL write metrics. Attributes: {:?}",attrs);
-
+        debug!(
+            "Building attributes for SSL write metrics. Attributes: {:?}",
+            attrs
+        );
 
         self.events_total.add(1, &attrs);
         self.ssl_write_bytes.record(m.size as i64, &attrs);

core/common/src/service_cache.rs

@@ -9,11 +9,11 @@ use tracing::debug;
 pub struct ServiceCache {
     // service_map should maitain the Option Type. Since Option type can be also None
     // it's used to deal all the cases
-    pub service_map: Option<HashMap<String, String>>,
+    pub service_map: Option<HashMap<String, Vec<String>>>,
 }
 impl ServiceCache {
     pub fn init(&mut self) {
-        self.service_map = Some(HashMap::<String, String>::new());
+        self.service_map = Some(HashMap::<String, Vec<String>>::new());
     }
 
     /// Helper function to query all pods form the Kubernetes API
@@ -35,10 +35,11 @@ impl ServiceCache {
 
             debug!("Querying and updating all the pods");
             for pod in all_pods {
-                if let (Some(name), Some(uid), Some(annotations)) = (
+                if let (Some(name), Some(uid), Some(annotations), Some(namespace)) = (
                     pod.metadata.name,
                     pod.metadata.uid,
                     pod.metadata.annotations,
+                    pod.metadata.namespace,
                 ) {
                     // docs from kubernetes
                     // Annotations is an unstructured key value map stored with a resource that
@@ -47,20 +48,18 @@ impl ServiceCache {
                     //More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations
 
                     if let Some(map) = self.service_map.as_mut() {
-
-                        // static pods are managed by the kubelet itself instead all the other pods 
+                        // static pods are managed by the kubelet itself instead all the other pods
                         // are managed by controllers without the need of the API server
                         // the kubelet mirrors the static pods but the UID does not match the real signature of the pod (verified)
-                        // so the algorithm check if the pod is static by checking if the annotations contains the word 'file' or 'http' 
+                        // so the algorithm check if the pod is static by checking if the annotations contains the word 'file' or 'http'
                         // reference for the labels: https://kubernetes.io/docs/reference/labels-annotations-taints/
                         // and copies the config.hash (aka the signature of the static pod) into the service cache instead of the UID
 
                         // more about static pods: https://medium.com/@iamsteffinissac/mastering-static-pods-in-kubernetes-bootstrapping-troubleshooting-mirror-pods-advanced-update-a35a5f95e329
 
-
                         let is_static_pod = annotations
                             .get("kubernetes.io/config.source")
-                            .is_some_and(|a| a == "file" || a== "http");
+                            .is_some_and(|a| a == "file" || a == "http");
 
                         let key = if is_static_pod {
                             annotations
@@ -70,7 +69,8 @@ impl ServiceCache {
                         } else {
                             uid
                         };
-                        map.insert(key, name);
+                        let v = vec![name, namespace];
+                        map.insert(key, v);
                     }
                 }
             } // insert the pod name and uid from the KubeAPI
@@ -80,7 +80,7 @@ impl ServiceCache {
 
         Ok(())
     }
-    pub async fn get_from_cache(&self, container_id: &str) -> Option<String> {
+    pub async fn get_from_cache(&self, container_id: &str) -> Option<&Vec<String>> {
         // pass a container_id and returns the object meta infos:
         /* pub struct ObjectMeta {
             pub annotations: Option<BTreeMap<String, String>>,
@@ -101,7 +101,12 @@ impl ServiceCache {
         } */
 
         if let Some(map) = self.service_map.as_ref() {
-            map.get(container_id).cloned() // cache hit 
+            // cache hit that contains ["name","namespace"] or None
+            if let Some(cached_metadata) = map.get(container_id) {
+                Some(cached_metadata)
+            } else {
+                None
+            }
         } else {
             None
         }

core/src/components/metrics/Cargo.toml

@@ -27,7 +27,7 @@ tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }
 libc = "0.2.172"
 bytemuck = "1.23.0"
-cortexbrain-common = { version="0.1.4", features = [
+cortexbrain-common = { path="../../../common", features = [
     "map-handlers",
     "program-handlers",
     "buffer-reader",

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

@@ -65,13 +65,7 @@ pub fn resolve_libssl_path() -> anyhow::Result<Option<String>> {
 /// service cache used by every perf-buffer consumer.
 pub async fn init_metrics_and_cache(
     meter: &Meter,
-) -> Result<
-    (
-        Arc<Metrics>,
-        Arc<tokio::sync::RwLock<ServiceCache>>,
-    ),
-    anyhow::Error,
-> {
+) -> Result<(Arc<Metrics>, Arc<tokio::sync::RwLock<ServiceCache>>), anyhow::Error> {
     let metrics = Arc::new(Metrics::new(meter));
 
     let mut cache_obj = ServiceCache { service_map: None };

core/src/testing/metrics.yaml

@@ -19,8 +19,8 @@ spec:
       hostNetwork: true
       containers:
         - name: metrics
-          image: lorenzotettamanti/cortexflow-metrics:0.1.1-cache-5s-update-6
-          command: ["/bin/bash", "-c"]
+          image: lorenzotettamanti/alpine-cf-metrics:latest
+          command: ["/bin/sh", "-c"]
           args:
             - |
               echo "Running on kernel $(uname -r)"
@@ -74,33 +74,6 @@ spec:
                 - SYS_RESOURCE
                 - BPF
                 - SYS_PTRACE
-        - name: bpftool-control-manager
-          image: danielpacak/bpftool-runner:latest
-          command: ["/bin/bash", "-c", "sleep infinity"]
-          volumeMounts:
-            - name: bpf
-              mountPath: /sys/fs/bpf
-              mountPropagation: Bidirectional
-              readOnly: false
-            - name: proc
-              mountPath: /host/proc
-              readOnly: false
-            - name: kernel-dev
-              mountPath: /lib/modules
-              readOnly: false
-            - name: tracefs
-              mountPath: /sys/kernel/debug
-              readOnly: false
-          securityContext:
-            privileged: true
-            allowPrivilegeEscalation: true
-            capabilities:
-              add:
-                - SYS_ADMIN
-                - NET_ADMIN
-                - SYS_RESOURCE
-                - BPF
-                - SYS_PTRACE
       volumes:
         - name: bpf
           hostPath: