diff --git a/.gitignore b/.gitignore index f9ca7d05..ad054ff6 100644 --- a/.gitignore +++ b/.gitignore @@ -166,4 +166,5 @@ config.json CLAUDE.md bindings.rs -site/ \ No newline at end of file +site/ +*.jsonc \ No newline at end of file diff --git a/core/Cargo.lock b/core/Cargo.lock index b5bb8f71..4a48b5b1 100644 --- a/core/Cargo.lock +++ b/core/Cargo.lock @@ -458,7 +458,7 @@ dependencies = [ "aya-log", "bytemuck", "bytes", - "cortexbrain-common 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", + "cortexbrain-common 0.1.2", "libc", "nix", "opentelemetry", diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml index 38b556ea..4f702ed8 100644 --- a/core/common/Cargo.toml +++ b/core/common/Cargo.toml @@ -24,7 +24,7 @@ opentelemetry-otlp = { version = "0.32.0", features = ["logs", "grpc-tonic"] } bytemuck = "1.25.0" bytes = "1.11.0" bytemuck_derive = "1.10.2" -tokio = "1.49.0" +tokio = {version="1.49.0",features=["sync"]} serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/core/common/src/buffer_type.rs b/core/common/src/buffer_type.rs index 2a0f1f06..a9cbba13 100644 --- a/core/common/src/buffer_type.rs +++ b/core/common/src/buffer_type.rs @@ -110,6 +110,7 @@ pub struct PacketLossMetrics { pub sk_receive_buffer_size: i32, // Offset 244 pub sk_ack_backlog: u32, // Offset 604 pub sk_drops: i32, // Offset 136 + pub cgroup_id: u64, } #[cfg(feature = "monitoring-structs")] unsafe impl aya::Pod for PacketLossMetrics {} @@ -129,6 +130,7 @@ pub struct TimeStampMetrics { pub daddr_v4: u32, pub saddr_v6: [u32; 4], pub daddr_v6: [u32; 4], + pub cgroup_id: u64, } #[cfg(feature = "monitoring-structs")] unsafe impl aya::Pod for TimeStampMetrics {} @@ -151,6 +153,7 @@ pub struct MemAlloc { pub length: u64, pub addr: u64, pub command: [u8; TASK_COMM_LEN], + pub cgroup_id: u64, } #[cfg(feature = "monitoring-structs")] unsafe impl aya::Pod for MemAlloc {} @@ -162,6 +165,7 @@ pub struct SchedStatWait { pub tgid: u32, pub delay: u64, pub command: [u8; TASK_COMM_LEN], + pub cgroup_id: u64, } #[cfg(feature = "monitoring-structs")] unsafe impl aya::Pod for SchedStatWait {} @@ -173,6 +177,7 @@ pub struct SchedStatRuntime { pub tgid: u32, pub runtime: u64, pub command: [u8; TASK_COMM_LEN], + pub cgroup_id: u64, } #[cfg(feature = "monitoring-structs")] unsafe impl aya::Pod for SchedStatRuntime {} @@ -197,6 +202,7 @@ pub struct SslEvent { pub direction: u8, // 0 = read, 1 = write pub size: i32, // return value (bytes transferred or <0 on error) pub requested: i32, // num argument passed to SSL_read/SSL_write + pub cgroup_id: u64, } unsafe impl aya::Pod for SslEvent {} diff --git a/core/common/src/consumer.rs b/core/common/src/consumer.rs index ef082bc6..46f3f40e 100644 --- a/core/common/src/consumer.rs +++ b/core/common/src/consumer.rs @@ -22,9 +22,12 @@ use crate::buffer_type::{PacketLog, TcpPacketRegistry, VethLog}; use crate::metadata::Metadata; #[cfg(feature = "monitoring-structs")] use crate::otel_metrics::Metrics; +use crate::service_cache::ServiceCache; use bytes::BytesMut; #[cfg(feature = "monitoring-structs")] use std::sync::Arc; +#[cfg(feature = "buffer-reader")] +use tokio::sync::RwLock; use tracing::{error, info, warn}; /// Discriminator for perf-buffer event types consumed by the collector. @@ -271,6 +274,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -293,9 +297,12 @@ impl Consumer { match exporter { "otlp" => { - let mut metadata = - Metadata::from_ebpf(Some(packet_loss.tgid), &packet_loss.comm); - metadata.enrich(); + let mut metadata = Metadata::from_ebpf( + Some(packet_loss.tgid), + Some(packet_loss.cgroup_id), + &packet_loss.comm, + ); + metadata.enrich(&cache).await; metrics.record_packet_loss_metrics(&packet_loss, &metadata); } _ => continue, @@ -313,7 +320,8 @@ impl Consumer { 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: {}", + "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: {}", tgid, comm, ts_us, @@ -323,7 +331,7 @@ impl Consumer { sk_backlog_len, sk_write_memory_queued, sk_ack_backlog, - sk_receive_buffer_size + sk_receive_buffer_size, ); } } @@ -339,6 +347,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -363,9 +372,10 @@ impl Consumer { "otlp" => { let mut metadata = Metadata::from_ebpf( Some(time_stamp_event.tgid), + Some(time_stamp_event.cgroup_id), &time_stamp_event.comm, ); - metadata.enrich(); + metadata.enrich(&cache).await; metrics.record_timestamp_metrics(&time_stamp_event, &metadata); } _ => continue, @@ -394,6 +404,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -418,9 +429,10 @@ impl Consumer { "otlp" => { let mut metadata = Metadata::from_ebpf( Some(cpu_freq_metrics.pid), + None, &cpu_freq_metrics.command, ); - metadata.enrich(); + metadata.enrich(&cache).await; metrics.record_cpu_bytes_alloc(&cpu_freq_metrics, &metadata); } _ => continue, @@ -445,6 +457,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -467,9 +480,12 @@ impl Consumer { match exporter { "otlp" => { - let mut metadata = - Metadata::from_ebpf(Some(mem_alloc.tgid), &mem_alloc.command); - metadata.enrich(); + let mut metadata = Metadata::from_ebpf( + Some(mem_alloc.tgid), + Some(mem_alloc.cgroup_id), + &mem_alloc.command, + ); + metadata.enrich(&cache).await; metrics.record_enter_mem_alloc(&mem_alloc, &metadata); } _ => continue, @@ -496,6 +512,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -520,9 +537,10 @@ impl Consumer { "otlp" => { let mut metadata = Metadata::from_ebpf( Some(sched_stat_wait.tgid), + Some(sched_stat_wait.cgroup_id), &sched_stat_wait.command, ); - metadata.enrich(); + metadata.enrich(&cache).await; metrics.record_sched_stat_wait(&sched_stat_wait, &metadata); } _ => continue, @@ -548,6 +566,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -572,9 +591,10 @@ impl Consumer { "otlp" => { let mut metadata = Metadata::from_ebpf( Some(sched_stat_runtime.tgid), + Some(sched_stat_runtime.cgroup_id), &sched_stat_runtime.command, ); - metadata.enrich(); + metadata.enrich(&cache).await; metrics.record_sched_stat_runtime(&sched_stat_runtime, &metadata); } _ => continue, @@ -600,6 +620,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { let vec_bytes = &buffers[i as usize]; @@ -622,7 +643,7 @@ impl Consumer { match exporter { "otlp" => { - let metadata = Metadata::from_ebpf(None, &[]); + let metadata = Metadata::from_ebpf(None, None, &[]); metrics.record_cpu_idle(&cpu_idle, &metadata); } _ => continue, @@ -645,6 +666,7 @@ impl Consumer { offset: i32, exporter: &str, metrics: Arc, + cache: Arc>, ) { for i in offset..tot_events { use crate::buffer_type::SslEvent; @@ -673,8 +695,12 @@ impl Consumer { // 0 = read // 1= write 0 => match exporter { "otlp" => { - let mut metadata = Metadata::from_ebpf(None, &[]); - metadata.enrich(); + let mut metadata = Metadata::from_ebpf( + Some(ssl_event.tgid), + Some(ssl_event.cgroup_id), + &ssl_event.comm, + ); + metadata.enrich(&cache).await; metrics.record_ssl_read_bytes(&ssl_event, &metadata); let tgid = ssl_event.tgid; let command = String::from_utf8_lossy(&ssl_event.comm); @@ -690,8 +716,12 @@ impl Consumer { }, 1 => match exporter { "otlp" => { - let mut metadata = Metadata::from_ebpf(None, &[]); - metadata.enrich(); + let mut metadata = Metadata::from_ebpf( + Some(ssl_event.tgid), + Some(ssl_event.cgroup_id), + &ssl_event.comm, + ); + metadata.enrich(&cache).await; metrics.record_ssl_write_bytes(&ssl_event, &metadata); let tgid = ssl_event.tgid; let command = String::from_utf8_lossy(&ssl_event.comm); @@ -729,6 +759,7 @@ pub async fn read_perf_buffer>( mut buffers: Vec, consumer: Consumer, #[cfg(feature = "monitoring-structs")] metrics: Option>, + cache: Option>>, ) { loop { for buf in array_buffers.iter_mut() { @@ -767,6 +798,7 @@ pub async fn read_perf_buffer>( metrics .clone() .expect("Metrics required for PacketLossMetrics"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -780,6 +812,7 @@ pub async fn read_perf_buffer>( metrics .clone() .expect("Metric required for TimeStampMetrics"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -791,6 +824,7 @@ pub async fn read_perf_buffer>( offset, "otlp", metrics.clone().expect("Metric required for CpuFrequency"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -802,6 +836,7 @@ pub async fn read_perf_buffer>( offset, "otlp", metrics.clone().expect("Metric required for MemAlloc"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -813,6 +848,7 @@ pub async fn read_perf_buffer>( offset, "otlp", metrics.clone().expect("Metric required for SchedStatWait"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -826,6 +862,7 @@ pub async fn read_perf_buffer>( metrics .clone() .expect("Metric required for SchedStatRuntime"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -837,6 +874,7 @@ pub async fn read_perf_buffer>( offset, "otlp", metrics.clone().expect("Metric required for CpuIdle"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } @@ -848,6 +886,7 @@ pub async fn read_perf_buffer>( offset, "otlp", metrics.clone().expect("Metric required for SslEvents"), + cache.clone().expect("cache required for PacketLossMetrics"), ) .await } diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 39231dd5..1ca47065 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -19,3 +19,4 @@ pub mod metadata; pub mod consumer; #[cfg(feature = "experimental")] pub mod service_discovery; +pub mod service_cache; diff --git a/core/common/src/metadata.rs b/core/common/src/metadata.rs index 9c97bbb1..ff76939a 100644 --- a/core/common/src/metadata.rs +++ b/core/common/src/metadata.rs @@ -1,4 +1,14 @@ +use anyhow::Error; use std::fs; +use std::os::unix::fs::MetadataExt; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +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)] @@ -9,27 +19,28 @@ pub enum ContainerRuntime { } /// Process / container metadata enriched by service discovery. -/// /// Built from raw eBPF data and enriched with a cgroup -> Docker -> K8s lookup. #[derive(Debug, Clone)] pub struct Metadata { pub tgid: Option, + pub cgroup_id: Option, pub command: String, pub runtime: ContainerRuntime, pub container_name: Option, pub container_id: Option, pub pod_name: Option, - pub namespace: Option, + pub namespace: Option, // TODO: implement this one } impl Metadata { /// Build base metadata from eBPF data. - pub fn from_ebpf(tgid: Option, command: &[u8]) -> Self { + pub fn from_ebpf(tgid: Option, cgroup_id: Option, command: &[u8]) -> Self { let command = String::from_utf8_lossy(command) .trim_end_matches('\0') .to_string(); Self { tgid, + cgroup_id, command, runtime: ContainerRuntime::Unknown, container_name: None, @@ -39,65 +50,151 @@ impl Metadata { } } - /// Lookup rules: first Docker (filesystem), then Kubernetes (API). + /// Lookup rules: prefer cgroup_id (v2), fall back to /proc//cgroup (v1). /// - /// 1. Reads `/proc//cgroup`. - /// 2. Extracts the container ID from the cgroup path. - /// 3. Tries to resolve the container name from `/var/lib/docker/containers//config.v2.json`. - /// 4. If Docker is not found, attempts K8s lookup. - pub fn enrich(&mut self) { - self.try_resolve_docker(); - // K8s lookup will be enabled later with an LRU cache. + /// 1. If cgroup_id is set and non-zero, resolve it via /sys/fs/cgroup (cgroup v2). + /// 2. Otherwise (cgroup v1, or eBPF didn't provide cgroup_id), fall back to reading + /// /host/proc//cgroup. + pub async fn enrich(&mut self, cache: &Arc>) { + if let Some(cgid) = self.cgroup_id + && cgid != 0 + && detect_cgroup_v2() + && self.try_resolve_from_cgroup_id(cgid, cache).await + { + return; + } else { + debug!("cgroup_v2 not detected. Using cgroup v1"); + self.try_resolve_docker(); + } + self.try_resolve_k8s(cache).await; + } + + /// Resolve pod/container from a kernel cgroup_id by walking /sys/fs/cgroup (v2). + /// Returns true if resolution succeeded (pod_uid extracted), false to allow caller fallback. + async fn try_resolve_from_cgroup_id( + &mut self, + cgroup_id: u64, + cache: &Arc>, + ) -> bool { + let Some(cgroup_path) = find_cgroup_path_by_id(cgroup_id) else { + debug!( + "cgroup_id {} not found under {}", + cgroup_id, "/host/sys/fs/cgroup" + ); + return false; + }; + + if let Some(id) = extract_pod_uid(cgroup_path.to_string_lossy().to_string()) { + self.container_id = Some(id.clone()); + self.runtime = ContainerRuntime::Kubernetes; + + // get pod from cache + // acquire cache lock + let cache_lock = cache.read().await; + match cache_lock.get_from_cache(&id).await { + Some(name) => self.pod_name = Some(name), + None => { + // fallback to the container_id if the k8s API cannot resolve the name + self.pod_name = Some(id.clone()); + } + } + return true; + } + + // not a k8s pod cgroup — try docker/containerd/crio container id + if let Some(id) = extract_container_id_from_path(&cgroup_path.to_string_lossy()) { + self.container_id = Some(id.clone()); + self.runtime = ContainerRuntime::Docker; + match resolve_docker_name(&id) { + Some(name) => self.container_name = Some(name), + None => self.container_name = self.container_id.clone(), + } + return true; + } + + false } /// Docker resolution via local filesystem. - /// + /// This part is triggered when the container is already detected // TODO: this is working for Linux, can anyone check if this works on macOs systems ? fn try_resolve_docker(&mut self) { let Some(tgid) = self.tgid else { return }; // Step 1: read the cgroup path from procfs - let cgroup_info = match fs::read_to_string(format!("/proc/{}/cgroup", tgid)) { - Ok(s) => s, - Err(e) => { - tracing::debug!("Cannot read /proc/{}/cgroup: {}", tgid, e); + if let Some(cgroup_info) = get_cgroup_info(tgid) { + // Extract the actual path from the cgroup file (format: hierarchy:id:path) + let cgroup_path = cgroup_info + .lines() + .filter_map(|line| line.split(':').nth(2)) + .next() + .unwrap_or(""); + + if cgroup_path.is_empty() { + info!("cgroup_path is empty"); return; } + + // Step 2: extract container ID from the path + if let Some(id) = extract_container_id_from_path(cgroup_path) { + self.container_id = Some(id.clone()); + self.runtime = ContainerRuntime::Docker; + + // Step 3: resolve container name from Docker metadata JSON + match resolve_docker_name(&id) { + Some(name) => self.container_name = Some(name), + None => { + self.container_name = self.container_id.clone(); // fallback to the container_id if the system cannot resolve the name after the 2 steps + } + } + } }; + } - // Extract the actual path from the cgroup file (format: hierarchy:id:path) - let cgroup_path = cgroup_info - .lines() - .filter_map(|line| line.split(':').nth(2)) - .next() - .unwrap_or(""); + async fn try_resolve_k8s(&mut self, cache: &Arc>) { + let Some(tgid) = self.tgid else { return }; - if cgroup_path.is_empty() { - return; - } + // Step 1: read the cgroup path from procfs + if let Some(cgroup_info) = get_cgroup_info(tgid) { + // Extract the actual path from the cgroup file (format: hierarchy:id:path) + let cgroup_path = cgroup_info + .lines() + .filter_map(|line| line.split(':').nth(2)) + .next() + .unwrap_or(""); - // Step 2: extract container ID from the path - if let Some(id) = extract_container_id_from_path(cgroup_path) { - self.container_id = Some(id.clone()); - self.runtime = ContainerRuntime::Docker; + if cgroup_path.is_empty() { + info!("cgroup_path is empty"); + return; + } - // Step 3: resolve container name from Docker metadata JSON - match resolve_docker_name(&id) { - Some(name) => self.container_name = Some(name), - None => self.container_name = Some("null".to_string()), + // Step 2: extract container ID from the path + if let Some(id) = extract_pod_uid(cgroup_path.to_string()) { + self.container_id = Some(id.clone()); + self.runtime = ContainerRuntime::Kubernetes; + + // Step 3: resolve container name from the k8s API + + //acquire cache lock + let cache_lock = cache.read().await; + + match cache_lock.get_from_cache(&id).await { + Some(name) => self.pod_name = Some(name), + None => { + // fallback to the container_id if the k8s API cannot resolve the name + self.pod_name = Some(id.clone()); + } + } } } } - - /// Manual enrichment from Kubernetes (for external use, e.g. identity service). - pub fn enrich_from_k8s(&mut self, pod_name: impl Into, namespace: impl Into) { - self.runtime = ContainerRuntime::Kubernetes; - self.pod_name = Some(pod_name.into()); - self.namespace = Some(namespace.into()); - } } -/// Extract the container ID from a cgroup path, supporting multiple prefixes. +/// Helpers + +/// Helper function to extract the container ID from a cgroup path, supporting multiple prefixes. +/// Supports Docker, Containerd, CRI-O, Docker cgroup V1. +/// Used in try_resolve_docker function fn extract_container_id_from_path(cgroup_path: &str) -> Option { let parts: Vec<&str> = cgroup_path.split('/').collect(); @@ -127,10 +224,9 @@ fn extract_container_id_from_path(cgroup_path: &str) -> Option { None } -/// Resolve a Docker container name by reading `config.v2.json`. -/// -// TODO: Does this work on macOs? +/// Resolve a Docker container name by reading config.v2.json. fn resolve_docker_name(container_id: &str) -> Option { + // TODO: Does this work on macOs? let path = format!("/var/lib/docker/containers/{}/config.v2.json", container_id); let json_str = fs::read_to_string(&path).ok()?; let parsed: serde_json::Value = serde_json::from_str(&json_str).ok()?; @@ -139,9 +235,108 @@ fn resolve_docker_name(container_id: &str) -> Option { .get("Image")? .as_str()? .trim_start_matches('/'); + if image_name.is_empty() { + return None; // if images is empty returns None + } + Some(image_name.to_string()) } +/// Helper function to extract the pod name, called 'target' from a vector of splits ['','',''] +fn extract_target_from_splits(splits: Vec<&str>, target: &str) -> Result { + for (index, split) in splits.iter().enumerate() { + // find the split that contains the word 'pod' + if split.contains(target) { + debug!("Target index; {}", index); + return Ok(index); + } + } + Err(Error::msg("'-pod' word not found in split")) +} + +fn extract_pod_uid(cgroup_path: String) -> Option { + // example of cgroup path: + // /sys/fs/cgroup/kubelet.slice/kubelet-kubepods.slice/kubelet-kubepods-besteffort.slice/kubelet-kubepods-besteffort-pod93580201_87d5_44e6_9779_f6153ca17637.slice + // or + // /sys/fs/cgroup/kubelet.slice/kubelet-kubepods.slice/kubelet-kubepods-burstable.slice/kubelet-kubepods-burstable-poddd3a1c6b_af40_41b1_8e1c_9e31fe8d96cb.slice + + // split the path by "/" + let splits: Vec<&str> = cgroup_path.split("/").collect(); + debug!("Debugging splits: {:?}", &splits); + + let index = extract_target_from_splits(splits.clone(), "-pod").ok(); + + match index { + Some(idx) => { + let pod_split = splits[idx] + .trim_start_matches("kubelet-kubepods-besteffort-") + .trim_start_matches("kubelet-kubepods-burstable-") + .trim_start_matches("kubepods-besteffort-") + .trim_start_matches("kubepods-burstable-"); + + let uid_ = pod_split + .trim_start_matches("pod") + .trim_end_matches(".slice"); //return uids with underscore (_) [ex.dd3a1c6b_af40_41b1_8e1c_9e31fe8d96cb] + + let uid = Some(uid_.replace("_", "-")); + uid + } + None => { + debug!("Index returned a value of None"); + None + } + } +} + +/// Detect if the host is running cgroup v2. +/// On cgroup v2 the file /sys/fs/cgroup/cgroup.controllers exists; on v1 it +/// does not (the controllers are split across /sys/fs/cgroup/). +/// this is a way to detect if we can attach the try_resolve_k8s functions +fn detect_cgroup_v2() -> bool { + Path::new("/host/sys/fs/cgroup") + .join("cgroup.controllers") + .is_file() +} + +/// Scans /sys/fs/cgroup recursively and return the path of the directory that +/// matches cgroup_id. This lookup is guaranteed because of the structure of the cgroup v2 in linux +fn find_cgroup_path_by_id(cgroup_id: u64) -> Option { + let root = Path::new("/host/sys/fs/cgroup"); + let mut stack = vec![root.to_path_buf()]; + + while let Some(dir) = stack.pop() { + let meta = match fs::metadata(&dir) { + Ok(m) => m, + Err(_) => continue, + }; + + if meta.ino() == cgroup_id { + return Some(dir); + } + + if let Ok(entries) = fs::read_dir(&dir) { + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + stack.push(path); + } + } + } + } + + None +} + +fn get_cgroup_info(tgid: u32) -> Option { + let cgroup_info = match fs::read_to_string(format!("/proc/{}/cgroup", tgid)) { + Ok(s) => return Some(s), + Err(e) => { + debug!("Cannot read /proc/{}/cgroup: {}", tgid, e); + return None; + } + }; +} + #[cfg(test)] mod tests { use super::*; @@ -194,9 +389,10 @@ mod tests { println!("{}", &container_id); let docker_container_name = match resolve_docker_name(&container_id) { Some(name) => Some(name), - None => Some("null".to_string()), + None => None, }; assert_eq!(docker_container_name, Some("busybox:latest".to_string())) } + // TODO: missing tests for extract_pod_uid, extract_target_from_splits, try_resolve_docker, try_resolve_k8s, enrich. } diff --git a/core/common/src/otel_metrics.rs b/core/common/src/otel_metrics.rs index 5e3c3fbc..dd664162 100644 --- a/core/common/src/otel_metrics.rs +++ b/core/common/src/otel_metrics.rs @@ -14,10 +14,11 @@ use crate::buffer_type::{ CpuFrequency, CpuIdle, MemAlloc, PacketLossMetrics, SchedStatRuntime, SchedStatWait, SslEvent, TimeStampMetrics, }; -use crate::metadata::{ContainerRuntime, Metadata}; +use crate::metadata::{Metadata}; use crate::semantic::Semantic; use opentelemetry::KeyValue; use opentelemetry::metrics::{Counter, Gauge, Histogram, Meter}; +use tracing::debug; pub struct Metrics { /// Total number of eBPF events processed across all perf buffers. @@ -212,13 +213,11 @@ impl Metrics { attrs.push(KeyValue::new("command", metadata.command.clone())); // container metadata - attrs.push(KeyValue::new( - "container.name", - match &metadata.container_name { - Some(name) => name.clone(), - None => "null".to_string(), - }, - )); + + match &metadata.container_name { + Some(name) => attrs.push(KeyValue::new("container.name", name.clone())), + None => (), // process is not associated with a container, the cgroup path does not contain an container id + } if let Some(ref id) = metadata.container_id { attrs.push(KeyValue::new("container.id", id.clone())); @@ -244,6 +243,7 @@ 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); self.events_total.add(1, attrs); self.socket_events_total.add(1, attrs); @@ -257,6 +257,7 @@ 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); self.events_total.add(1, attrs); self.tcp_latency_us.record(m.delta_us, attrs); @@ -266,6 +267,7 @@ 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); self.cpu_bytes_alloc_events_total.add(1, attrs); self.cpu_bytes_alloc.record(bytes_allocated as i64, attrs); @@ -279,6 +281,7 @@ 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); self.events_total.add(1, attrs); self.mem_alloc_events_total.add(1, attrs); @@ -292,7 +295,8 @@ 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); + 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); @@ -306,6 +310,9 @@ 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); + + self.events_total.add(1, attrs); self.sched_stat_runtime.record(m.runtime as i64, attrs); self.sched_stat_runtime_distribution @@ -318,6 +325,9 @@ impl Metrics { /// `cpu_id`. Events are only emitted by eBPF when the state changes. 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); + attrs.push(KeyValue::new("cpu_id", m.cpu_id as i64)); self.events_total.add(1, &attrs); @@ -326,12 +336,16 @@ 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); + 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); + 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 new file mode 100644 index 00000000..80523be2 --- /dev/null +++ b/core/common/src/service_cache.rs @@ -0,0 +1,71 @@ +use anyhow::Error; +use k8s_openapi::api::core::v1::Pod; +use kube::api::ObjectList; +use kube::{Api, Client}; +use std::collections::HashMap; +use tracing::debug; + +#[derive(Clone, 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>, +} +impl ServiceCache { + pub fn init(&mut self) { + self.service_map = Some(HashMap::::new()); + } + + /// Helper function to query all pods form the Kubernetes API + pub async fn query_all_pods_from_kubeapi(&mut self) -> Result, Error> { + debug!("Connecting to kubernetes client"); + let client = Client::try_default().await?; + debug!("Querying all the pods"); + let pods: Api = Api::all(client); + let lp = kube::api::ListParams::default(); // default list params + let pod_list = pods.list(&lp).await?; + + Ok(pod_list) + } + + /// Helper function to populate and mantain a cache service map + pub async fn populate_map_with_pod_info(&mut self) -> Result<(), Error> { + let all_pods = self.query_all_pods_from_kubeapi().await?; + + debug!("Querying and updating all the pods"); + for pod in all_pods { + if let (Some(name), Some(uid)) = (pod.metadata.name, pod.metadata.uid) { + if let Some(map) = self.service_map.as_mut() { + map.insert(uid, name); + } + } + } // insert the pod name and uid from the KubeAPI + Ok(()) + } + pub async fn get_from_cache(&self, container_id: &str) -> Option { + // pass a container_id and returns the object meta infos: + /* pub struct ObjectMeta { + pub annotations: Option>, + pub creation_timestamp: Option