diff --git a/core/Cargo.lock b/core/Cargo.lock index 533c76e..df8fb62 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -556,7 +556,7 @@ dependencies = [ "aya-log", "bytemuck", "bytes", - "cortexbrain-common 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)", + "cortexbrain-common 0.1.4", "libc", "nix 0.31.3", "opentelemetry", diff --git a/core/common/src/metadata.rs b/core/common/src/metadata.rs index 448d7a4..dc22d40 100644 --- a/core/common/src/metadata.rs +++ b/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, pub container_id: Option, pub pod_name: Option, - pub namespace: Option, // TODO: implement this one + pub namespace: Option, } 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, diff --git a/core/common/src/otel_metrics.rs b/core/common/src/otel_metrics.rs index dd66416..1680ace 100644 --- a/core/common/src/otel_metrics.rs +++ b/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); diff --git a/core/common/src/service_cache.rs b/core/common/src/service_cache.rs index 9bd1b5c..36106d6 100644 --- a/core/common/src/service_cache.rs +++ b/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>, + pub service_map: Option>>, } impl ServiceCache { pub fn init(&mut self) { - self.service_map = Some(HashMap::::new()); + self.service_map = Some(HashMap::>::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 { + pub async fn get_from_cache(&self, container_id: &str) -> Option<&Vec> { // pass a container_id and returns the object meta infos: /* pub struct ObjectMeta { pub annotations: Option>, @@ -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 } diff --git a/core/src/components/metrics/Cargo.toml b/core/src/components/metrics/Cargo.toml index 1658418..823ff0a 100644 --- a/core/src/components/metrics/Cargo.toml +++ b/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", diff --git a/core/src/components/metrics/src/helpers.rs b/core/src/components/metrics/src/helpers.rs index 03b0a59..66473de 100644 --- a/core/src/components/metrics/src/helpers.rs +++ b/core/src/components/metrics/src/helpers.rs @@ -65,13 +65,7 @@ pub fn resolve_libssl_path() -> anyhow::Result> { /// service cache used by every perf-buffer consumer. pub async fn init_metrics_and_cache( meter: &Meter, -) -> Result< - ( - Arc, - Arc>, - ), - anyhow::Error, -> { +) -> Result<(Arc, Arc>), anyhow::Error> { let metrics = Arc::new(Metrics::new(meter)); let mut cache_obj = ServiceCache { service_map: None }; diff --git a/core/src/testing/metrics.yaml b/core/src/testing/metrics.yaml index ca2227d..80e16b3 100644 --- a/core/src/testing/metrics.yaml +++ b/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: