Skip to content

Fixed container.name null bug when using Docker. Updated metadata enrichment for k8s environments. Added service cache #216

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
LorenzoTettamanti wants to merge 9 commits into
base: 0.1.5
Choose a base branch
from
Draft

Fixed container.name null bug when using Docker. Updated metadata enrichment for k8s environments. Added service cache #216

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
9fb1d10
(fix): fix dockerfile for metrics build after the latest bpf-linker u…
LorenzoTettamanti Aug 24, 2026
e2d9124
(fix): fixed "null" in container.name when using Docker.
LorenzoTettamanti Aug 24, 2026
c289826
(update): updated gitignore
LorenzoTettamanti Aug 24, 2026
e67759a
(update): added cgroup_id extraction for every metrics traced
LorenzoTettamanti Aug 25, 2026
709787d
(debug): added debug points for debugging purpose
LorenzoTettamanti Aug 25, 2026
7016a62
(update): added simple service_cache
LorenzoTettamanti Aug 26, 2026
3cb611d
(update): added sync feature to tokio in cargo.toml
LorenzoTettamanti Aug 26, 2026
2bc6f7a
(feature): implemented service cache in metrics userspace
LorenzoTettamanti Aug 26, 2026
c3a1c04
(feature): updated metadata enrichment. Added try_resolve_from_cgroup…
LorenzoTettamanti Aug 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,4 +166,5 @@ config.json
CLAUDE.md

bindings.rs
site/
site/
*.jsonc
2 changes: 1 addition & 1 deletion core/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion core/common/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down
6 changes: 6 additions & 0 deletions core/common/src/buffer_type.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {}
Expand All @@ -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 {}
Expand All @@ -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 {}
Expand All @@ -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 {}
Expand All @@ -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 {}
Expand All @@ -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 {}

Expand Down
73 changes: 56 additions & 17 deletions core/common/src/consumer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -271,6 +274,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -323,7 +331,7 @@ impl Consumer {
sk_backlog_len,
sk_write_memory_queued,
sk_ack_backlog,
sk_receive_buffer_size
sk_receive_buffer_size,
);
}
}
Expand All @@ -339,6 +347,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand Down Expand Up @@ -394,6 +404,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -445,6 +457,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -496,6 +512,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -548,6 +566,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -600,6 +620,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
let vec_bytes = &buffers[i as usize];
Expand All @@ -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,
Expand All @@ -645,6 +666,7 @@ impl Consumer {
offset: i32,
exporter: &str,
metrics: Arc<Metrics>,
cache: Arc<RwLock<ServiceCache>>,
) {
for i in offset..tot_events {
use crate::buffer_type::SslEvent;
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -729,6 +759,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
mut buffers: Vec<bytes::BytesMut>,
consumer: Consumer,
#[cfg(feature = "monitoring-structs")] metrics: Option<Arc<Metrics>>,
cache: Option<Arc<RwLock<ServiceCache>>>,
) {
loop {
for buf in array_buffers.iter_mut() {
Expand Down Expand Up @@ -767,6 +798,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
metrics
.clone()
.expect("Metrics required for PacketLossMetrics"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -780,6 +812,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
metrics
.clone()
.expect("Metric required for TimeStampMetrics"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -791,6 +824,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
offset,
"otlp",
metrics.clone().expect("Metric required for CpuFrequency"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -802,6 +836,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
offset,
"otlp",
metrics.clone().expect("Metric required for MemAlloc"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -813,6 +848,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
offset,
"otlp",
metrics.clone().expect("Metric required for SchedStatWait"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -826,6 +862,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
metrics
.clone()
.expect("Metric required for SchedStatRuntime"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -837,6 +874,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
offset,
"otlp",
metrics.clone().expect("Metric required for CpuIdle"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand All @@ -848,6 +886,7 @@ pub async fn read_perf_buffer<T: std::borrow::BorrowMut<aya::maps::MapData>>(
offset,
"otlp",
metrics.clone().expect("Metric required for SslEvents"),
cache.clone().expect("cache required for PacketLossMetrics"),
)
.await
}
Expand Down
1 change: 1 addition & 0 deletions core/common/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,3 +19,4 @@ pub mod metadata;
pub mod consumer;
#[cfg(feature = "experimental")]
pub mod service_discovery;
pub mod service_cache;
Loading