diff --git a/core/common/src/buffer_type.rs b/core/common/src/buffer_type.rs index 7124e55..8ec000c 100644 --- a/core/common/src/buffer_type.rs +++ b/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::() { + if vec_bytes.len() < std::mem::size_of::() { error!( "Corrupted Network Metrics data. Raw data: {}. Readed {} bytes expected {} bytes", vec_bytes @@ -454,28 +454,28 @@ impl BufferType { .collect::>() .join(" "), vec_bytes.len(), - std::mem::size_of::() + std::mem::size_of::() ); continue; } - if vec_bytes.len() >= std::mem::size_of::() { - let net_metrics: NetworkMetrics = + if vec_bytes.len() >= std::mem::size_of::() { + 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>( .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::(), #[cfg(feature = "monitoring-structs")] - BufferSize::NetworkMetricsEvents => std::mem::size_of::(), + BufferSize::NetworkMetricsEvents => std::mem::size_of::(), #[cfg(feature = "monitoring-structs")] BufferSize::TimeMetricsEvents => std::mem::size_of::(), #[cfg(feature = "monitoring-structs")] diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 15c4ad7..a0b2877 100644 --- a/core/common/src/lib.rs +++ b/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; diff --git a/core/common/src/otel_metrics.rs b/core/common/src/otel_metrics.rs index 79d08b8..a1b2c86 100644 --- a/core/common/src/otel_metrics.rs +++ b/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, + pub socket_events_total: Counter, /// Observed socket drop count (`sk_drops`) from the kernel sock struct. pub sk_drops: Gauge, @@ -33,11 +34,7 @@ pub struct Metrics { /// Histogram of `delta_us` values supplied by the `time_stamp_events` /// perf buffer. - pub delta_us: Histogram, - - /// Histogram of `ts_us` values seen in both `net_metrics` and - /// `time_stamp_events`. - pub ts_us: Histogram, + pub tcp_latency_us: Histogram, /// Cpu bytes alloc total events pub cpu_bytes_alloc_events_total: Counter, @@ -61,94 +58,99 @@ pub struct Metrics { pub cpu_idle_state: Gauge, } +// 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); } } diff --git a/core/common/src/semantic.rs b/core/common/src/semantic.rs new file mode 100644 index 0000000..b16bdbe --- /dev/null +++ b/core/common/src/semantic.rs @@ -0,0 +1,61 @@ +/// semantic conventions + +pub enum Semantic { + TotalEvents, + SocketTotalEvents, + SocketDrops, + SocketErrorsCount, + Latency, + PerCpuTotalEvents, + PerCpuBytesAllocated, + SchedulerRuntime, + SchedulerWaitTime, + TotalMemoryAllocationEvents, + RequestedMemoryBytes, + CpuIdleState, +} + +impl Semantic { + pub fn title(&self) -> &'static str { + match self { + Semantic::TotalEvents => "events_total", + Semantic::SocketTotalEvents => "socket_events_total", + Semantic::SocketDrops => "sk_drops", + Semantic::SocketErrorsCount => "sk_err", + Semantic::Latency => "latency_us", + Semantic::PerCpuTotalEvents => "bytes_alloc_events_total", + Semantic::PerCpuBytesAllocated => "cpu_bytes_alloc", + Semantic::SchedulerRuntime => "sched_stat_runtime", + Semantic::SchedulerWaitTime => "sched_stat_wait", + Semantic::TotalMemoryAllocationEvents => "mem_alloc_events_total", + Semantic::RequestedMemoryBytes => "enter_mem_alloc", + Semantic::CpuIdleState => "cpu_idle_state", + } + } + pub fn description(&self) -> &'static str { + match self { + Semantic::TotalEvents => { + "Total number of eBPF events processed across all perf buffers" + } + Semantic::SocketTotalEvents => "Total number of socket state events processed", + Semantic::SocketDrops => "Socket drop count per event", + Semantic::SocketErrorsCount => "Socket error count per event", + Semantic::Latency => "Distribution of latency values from timestamp events", + Semantic::PerCpuTotalEvents => "Total bytes_alloc events occurring in the CPU", + Semantic::PerCpuBytesAllocated => "CPU bytes allocation per event", + Semantic::SchedulerRuntime => { + "Scheduler runtime in nanoseconds from sched_stat_runtime" + } + Semantic::SchedulerWaitTime => { + "Scheduler wait time in nanoseconds from sched_stat_wait" + } + Semantic::TotalMemoryAllocationEvents => { + "Total number of memory allocation (mmap) events processed" + } + Semantic::RequestedMemoryBytes => "Bytes requested via mmap syscalls", + Semantic::CpuIdleState => { + "Current CPU idle C-state per cpu_id, updated only on state change" + } + } + } +} diff --git a/core/src/components/metrics/build-local-metrics.sh b/core/src/components/metrics/build-local-metrics.sh index b861e6e..d0ed752 100755 --- a/core/src/components/metrics/build-local-metrics.sh +++ b/core/src/components/metrics/build-local-metrics.sh @@ -3,7 +3,7 @@ # Building identity files echo "Building the metrics-tracer files" pushd ../metrics_tracer -./build-metrics-tracer.sh +./build-local-metrics-tracer.sh popd echo "Copying metrics_tracer binaries" diff --git a/core/src/components/metrics/src/helpers.rs b/core/src/components/metrics/src/helpers.rs index 141cad6..f162048 100644 --- a/core/src/components/metrics/src/helpers.rs +++ b/core/src/components/metrics/src/helpers.rs @@ -103,7 +103,7 @@ pub async fn event_listener(bpf_maps: BpfMapsData, meter: Meter) -> Result<(), a read_perf_buffer( array_buffers, buffers, - BufferType::NetworkMetrics, + BufferType::PacketLossMetrics, Some(metrics), ) .await; diff --git a/core/src/components/metrics/src/main.rs b/core/src/components/metrics/src/main.rs index bcf7de7..3a7bcbb 100644 --- a/core/src/components/metrics/src/main.rs +++ b/core/src/components/metrics/src/main.rs @@ -79,10 +79,14 @@ async fn main() -> Result<(), anyhow::Error> { info!("BPF maps pinned successfully to {}", bpf_map_save_path); { - load_program(bpf.clone(), "metrics_tracer", "tcp_identify_packet_loss") - .context( - "An error occurred during the execution of load_program function", - )?; + load_program( + bpf.clone(), + "packet_loss_tracer", + "tcp_identify_packet_loss", + ) + .context( + "An error occurred during the execution of load_program function", + )?; load_program(tcp_bpf, "tcp_v4_connect", "tcp_v4_connect") .context("An error occurred during the execution of load_and_attach_tcp_programs function")?; @@ -90,12 +94,8 @@ async fn main() -> Result<(), anyhow::Error> { load_program(tcp_v6_bpf, "tcp_v6_connect", "tcp_v6_connect") .context("An error occurred during the execution of load_and_attach_tcp_programs function")?; - load_program( - tcp_rev_bpf, - "tcp_rcv_state_process", - "tcp_rcv_state_process", - ) - .context( + load_program(tcp_rev_bpf, "tcp_latency_monitor", "tcp_rcv_state_process") + .context( "An error occurred during the execution of load_program function", )?; load_tracepoint_program( diff --git a/core/src/components/metrics_tracer/src/cpu.rs b/core/src/components/metrics_tracer/src/cpu.rs index 374ea36..dbab7d4 100644 --- a/core/src/components/metrics_tracer/src/cpu.rs +++ b/core/src/components/metrics_tracer/src/cpu.rs @@ -2,7 +2,7 @@ //tracepoint:power:cpu_frequency_limits //tracepoint:power:cpu_idle //tracepoint:power:cpu_idle_miss -use aya_ebpf::{EbpfContext, programs::TracePointContext}; +use aya_ebpf::{EbpfContext, helpers::bpf_get_current_pid_tgid, programs::TracePointContext}; use aya_log_ebpf::info; use crate::data_structures::{CPU_FREQUENCY, CPU_IDLE, CPU_IDLE_LAST_STATE, CpuFrequency, CpuIdle}; @@ -37,7 +37,9 @@ pub fn per_cpu_bytes_alloc(ctx: &TracePointContext) -> Result<((u32, u32, [u8; 1 let bytes_alloc_offset = 64; let pid_offset = 4; let bytes_alloc = unsafe { ctx.read_at(bytes_alloc_offset) }?; - let pid = unsafe { ctx.read_at(pid_offset) }?; + //let tgid: u32 = unsafe { ctx.read_at(tgid_offset) }?; + let pid_tgid: u64 = bpf_get_current_pid_tgid(); + let tgid: u32 = (pid_tgid >> 32) as u32; let command = ctx.command()?; //let cpu_freq_data = CpuFrequency { @@ -47,29 +49,33 @@ pub fn per_cpu_bytes_alloc(ctx: &TracePointContext) -> Result<((u32, u32, [u8; 1 //CPU_FREQUENCY.output(&ctx, &cpu_freq_data, 0); - Ok((bytes_alloc, pid, command)) + Ok((bytes_alloc, tgid, command)) } pub fn sched_stat_wait(ctx: &TracePointContext) -> Result<((u32, u64, [u8; 16])), i64> { let pid_offset = 4; let delay_offset = 16; - let pid = unsafe { ctx.read_at(pid_offset) }?; + //let tgid: u32 = unsafe { ctx.read_at(tgid_offset) }?; + let pid_tgid: u64 = bpf_get_current_pid_tgid(); + let tgid: u32 = (pid_tgid >> 32) as u32; let delay = unsafe { ctx.read_at(delay_offset) }?; let command = ctx.command()?; - Ok((pid, delay, command)) + Ok((tgid, delay, command)) } pub fn sched_stat_runtime(ctx: &TracePointContext) -> Result<((u32, u64, [u8; 16])), i64> { let pid_offset = 4; let runtime_offset = 16; - let pid = unsafe { ctx.read_at(pid_offset) }?; + //let tgid: u32 = unsafe { ctx.read_at(tgid_offset) }?; + let pid_tgid: u64 = bpf_get_current_pid_tgid(); + let tgid: u32 = (pid_tgid >> 32) as u32; let runtime = unsafe { ctx.read_at(runtime_offset) }?; let command = ctx.command()?; - Ok((pid, runtime, command)) + Ok((tgid, runtime, command)) } diff --git a/core/src/components/metrics_tracer/src/data_structures.rs b/core/src/components/metrics_tracer/src/data_structures.rs index d74ef35..cfc80b6 100644 --- a/core/src/components/metrics_tracer/src/data_structures.rs +++ b/core/src/components/metrics_tracer/src/data_structures.rs @@ -6,7 +6,7 @@ use aya_ebpf::{ pub const TASK_COMM_LEN: usize = 16; #[repr(C, packed)] -pub struct NetworkMetrics { +pub struct PacketLossMetrics { pub tgid: u32, pub comm: [u8; TASK_COMM_LEN], pub ts_us: u64, @@ -27,7 +27,8 @@ pub struct TimeStampStartInfo { pub tgid: u32, } -// Event we send to userspace when latency is computed +/// Event we send to userspace when latency is computed +/// used to compute tcp_delta_us, tcp_ts_us metrics #[repr(C, packed)] #[derive(Copy, Clone)] pub struct TimeStampEvent { @@ -50,7 +51,7 @@ pub struct CpuFrequency { //pub(crate) cpu_id: u32, //pub(crate) cpu_freq: u32, pub(crate) bytes_alloc: u32, - pub(crate) pid: u32, + pub(crate) tgid: u32, pub(crate) command: [u8; 16], } @@ -86,6 +87,25 @@ pub struct CpuIdle { pub(crate) state: u32, } +#[repr(C, packed)] +#[derive(Copy, Clone, Debug, bytemuck::Pod, bytemuck::Zeroable)] +pub struct SslEvent { + pub tgid: u32, + pub comm: [u8; TASK_COMM_LEN], + pub ts_us: u64, + 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 +} + +#[repr(C, packed)] +#[derive(Copy, Clone, Debug, bytemuck::Pod, bytemuck::Zeroable)] +pub struct SslCtxInfo { + pub ssl: u64, // SSL* pointer + pub requested: i32, + pub ts_ns: u64, +} + // Map: connect-start timestamp by socket pointer #[map(name = "time_stamp_start")] pub static mut TIME_STAMP_START: HashMap<*mut core::ffi::c_void, TimeStampStartInfo> = @@ -97,7 +117,7 @@ pub static mut TIME_STAMP_EVENTS: PerfEventArray = PerfEventArray::::new(0); #[map(name = "net_metrics")] -pub static NET_METRICS: PerfEventArray = PerfEventArray::new(0); +pub static NET_METRICS: PerfEventArray = PerfEventArray::new(0); #[map(name = "cpu_frequency")] pub static CPU_FREQUENCY: PerfEventArray = PerfEventArray::new(0); @@ -117,3 +137,10 @@ pub static CPU_IDLE: PerfEventArray = PerfEventArray::new(0); #[map(name = "cpu_idle_last_state")] pub static mut CPU_IDLE_LAST_STATE: HashMap = HashMap::::with_max_entries(256, 0); + +#[map(name = "ssl_ctx_map")] +pub static mut SSL_CTX_MAP: HashMap = + HashMap::::with_max_entries(4096, 0); + +#[map(name = "ssl_events")] +pub static SSL_EVENTS: PerfEventArray = PerfEventArray::new(0); diff --git a/core/src/components/metrics_tracer/src/main.rs b/core/src/components/metrics_tracer/src/main.rs index 55faa45..1da3451 100644 --- a/core/src/components/metrics_tracer/src/main.rs +++ b/core/src/components/metrics_tracer/src/main.rs @@ -6,6 +6,7 @@ mod bindings; mod cpu; mod data_structures; mod memory; +mod network; use crate::bindings::net_device; use crate::cpu::{cpu_idle, per_cpu_bytes_alloc, sched_stat_runtime, sched_stat_wait}; @@ -13,12 +14,13 @@ use crate::data_structures::CpuFrequency; use crate::data_structures::NET_METRICS; use crate::data_structures::{CPU_FREQUENCY, SchedStatWait}; use crate::data_structures::{ - CPU_IDLE, NetworkMetrics, TASK_COMM_LEN, TIME_STAMP_EVENTS, TIME_STAMP_START, TimeStampEvent, - TimeStampStartInfo, + CPU_IDLE, PacketLossMetrics, TASK_COMM_LEN, TIME_STAMP_EVENTS, TIME_STAMP_START, + TimeStampEvent, TimeStampStartInfo, }; use crate::data_structures::{MEM_ALLOC, SCHED_STAT_RUNTIME, SCHED_STAT_WAIT}; use crate::data_structures::{MemAlloc, SchedStatRuntime}; use crate::memory::enter_mmap; +use crate::network::{detect_packet_loss, on_connect, on_rcv_state_process}; use aya_ebpf::EbpfContext; use aya_ebpf::helpers::bpf_get_current_pid_tgid; use aya_ebpf::helpers::generated::{bpf_ktime_get_ns, bpf_perf_event_output}; @@ -35,7 +37,7 @@ const AF_INET6: u16 = 10; const TCP_SYN_SENT: u8 = 2; #[kprobe] -fn metrics_tracer(ctx: ProbeContext) -> u32 { +fn packet_loss_tracer(ctx: ProbeContext) -> u32 { match try_metrics_tracer(ctx) { Ok(ret) => ret, Err(ret) => ret.try_into().unwrap_or(1), @@ -43,64 +45,7 @@ fn metrics_tracer(ctx: ProbeContext) -> u32 { } fn try_metrics_tracer(ctx: ProbeContext) -> Result { - let sk_pointer = ctx.arg::<*const u8>(0).ok_or(1i64)?; - - if sk_pointer.is_null() { - return Err(1); - } - - let tgid = (unsafe { bpf_get_current_pid_tgid() } >> 32) as u32; - let comm = unsafe { bpf_get_current_comm() }.map_err(|_| 1i64)?; - let ts_us: u64 = unsafe { bpf_ktime_get_ns() } / 1_000; - let sk_err_offset = 284; - let sk_err_soft_offset = 600; - let sk_backlog_len_offset = 196; - let sk_write_memory_queued_offset = 376; - let sk_receive_buffer_size_offset = 244; - let sk_ack_backlog_offset = 604; - let sk_drops_offset = 136; - - let sk_err = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_err_offset) as *const i32).map_err(|_| 1)? - }; - let sk_err_soft = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_err_soft_offset) as *const i32) - .map_err(|_| 1)? - }; - let sk_backlog_len = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_backlog_len_offset) as *const i32) - .map_err(|_| 1)? - }; - let sk_write_memory_queued = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_write_memory_queued_offset) as *const i32) - .map_err(|_| 1)? - }; - let sk_receive_buffer_size = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_receive_buffer_size_offset) as *const i32) - .map_err(|_| 1)? - }; - let sk_ack_backlog = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_ack_backlog_offset) as *const u32) - .map_err(|_| 1)? - }; - let sk_drops = unsafe { - bpf_probe_read_kernel::(sk_pointer.add(sk_drops_offset) as *const i32) - .map_err(|_| 1)? - }; - - let net_metrics = NetworkMetrics { - tgid: tgid, - comm: comm, - ts_us: ts_us, - sk_err: sk_err, - sk_err_soft: sk_err_soft, - sk_backlog_len: sk_backlog_len, - sk_write_memory_queued: sk_write_memory_queued, - sk_receive_buffer_size: sk_receive_buffer_size, - sk_ack_backlog: sk_ack_backlog, - sk_drops: sk_drops, - }; - + let net_metrics = detect_packet_loss(&ctx)?; unsafe { NET_METRICS.output(&ctx, &net_metrics, 0); } @@ -108,7 +53,7 @@ fn try_metrics_tracer(ctx: ProbeContext) -> Result { Ok(0) } -// Monitor on tcp_sendmsg, tcp_v4_connect +/// Monitor on tcp_sendmsg, tcp_v6_connect #[kprobe] fn tcp_v6_connect(ctx: ProbeContext) -> u32 { match on_connect(ctx) { @@ -117,7 +62,7 @@ fn tcp_v6_connect(ctx: ProbeContext) -> u32 { } } -// Monitor on tcp_sendmsg, tcp_v4_connect +/// Monitor on tcp_sendmsg, tcp_v4_connect #[kprobe] fn tcp_v4_connect(ctx: ProbeContext) -> u32 { match on_connect(ctx) { @@ -126,149 +71,14 @@ fn tcp_v4_connect(ctx: ProbeContext) -> u32 { } } -fn on_connect(ctx: ProbeContext) -> Result<(), i64> { - let sk = ctx.arg::<*mut bindings::sock>(0).ok_or(1i64)?; - if sk.is_null() { - return Err(1); - } - - let tgid = (unsafe { bpf_get_current_pid_tgid() } >> 32) as u32; - let mut start = TimeStampStartInfo { - comm: [0; TASK_COMM_LEN], - ts_ns: unsafe { bpf_ktime_get_ns() }, - tgid, - }; - unsafe { - let comm_result = bpf_get_current_comm(); - if let Ok(comm) = comm_result { - start.comm.copy_from_slice(&comm); - } - let map_ptr = &raw mut TIME_STAMP_START; - (*map_ptr) - .insert(&(sk as *mut core::ffi::c_void), &start, 0) - .map_err(|_| 1)?; - } - Ok(()) -} - #[kprobe] -fn tcp_rcv_state_process(ctx: ProbeContext) -> u32 { +fn tcp_latency_monitor(ctx: ProbeContext) -> u32 { match on_rcv_state_process(ctx) { Ok(_) => 0, Err(e) => e as u32, } } -fn on_rcv_state_process(ctx: ProbeContext) -> Result<(), i64> { - // On some kernels, kprobe wrapper puts `sk` at arg0; on others arg1. - let sk = ctx.arg::<*mut bindings::sock>(0).unwrap_or(ptr::null_mut()); - let sk = if sk.is_null() { - ctx.arg::<*mut bindings::sock>(1).ok_or(1i64)? - } else { - sk - }; - - if sk.is_null() { - return Err(1); - } - - let skc_daddr_off = 0; - let skc_rcv_saddr_off = 4; - let skc_dport_off = 12; - let skc_num_off = 14; - let skc_family_off = 16; - let skc_state_off = 18; - let skc_v6_daddr_off = 56; - let skc_v6_rcv_saddr_off = 72; - - let state = unsafe { bpf_probe_read_kernel::((sk as usize + skc_state_off) as *const u8) } - .map_err(|_| 1)?; - - if state != TCP_SYN_SENT { - return Ok(()); - } - - let start = unsafe { - let map_ptr = &raw const TIME_STAMP_START; - (*map_ptr).get(&((sk as usize) as *mut core::ffi::c_void)) - } - .ok_or(1i64)?; - let now = unsafe { bpf_ktime_get_ns() }; - let delta = now as i64 - start.ts_ns as i64; - if delta <= 0 { - unsafe { - let map_ptr = &raw mut TIME_STAMP_START; - let _ = (*map_ptr).remove(&((sk as usize) as *mut core::ffi::c_void)); - } - return Ok(()); - } - - let mut ev = TimeStampEvent { - delta_us: (delta as u64) / 1_000, - ts_us: now / 1_000, - tgid: start.tgid, - comm: start.comm, - lport: 0, - dport_be: 0, - af: 0, - saddr_v4: 0, - daddr_v4: 0, - saddr_v6: [0; 4], - daddr_v6: [0; 4], - }; - - // family, ports - ev.af = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_family_off) as *const u16).map_err(|_| 1)? - }; - ev.lport = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_num_off) as *const u16).map_err(|_| 1)? - }; - ev.dport_be = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_dport_off) as *const u16).map_err(|_| 1)? - }; - - if ev.af == AF_INET { - ev.saddr_v4 = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_rcv_saddr_off) as *const u32) - .map_err(|_| 1)? - }; - ev.daddr_v4 = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_daddr_off) as *const u32) - .map_err(|_| 1)? - }; - } else { - // read 16 bytes as four u32 words - for i in 0..4 { - ev.saddr_v6[i] = unsafe { - bpf_probe_read_kernel::( - (sk as usize + skc_v6_rcv_saddr_off + i * 4) as *const u32, - ) - .map_err(|_| 1)? - }; - ev.daddr_v6[i] = unsafe { - bpf_probe_read_kernel::((sk as usize + skc_v6_daddr_off + i * 4) as *const u32) - .map_err(|_| 1)? - }; - } - } - - // emit + cleanup - unsafe { - bpf_perf_event_output( - ctx.as_ptr(), - &raw const TIME_STAMP_EVENTS as *const _ as *mut _, - 0, // BPF_F_CURRENT_CPU - &ev as *const _ as *mut _, - (mem::size_of::() as u32).into(), - ); - let map_ptr = &raw mut TIME_STAMP_START; - let _ = (*map_ptr).remove(&((sk as usize) as *mut core::ffi::c_void)); - } - - Ok(()) -} - #[tracepoint] fn trace_cpu_frequency(ctx: TracePointContext) -> u32 { match trace_cpu_metrics(&ctx) { @@ -286,13 +96,13 @@ fn trace_cpu_idle(ctx: TracePointContext) -> u32 { } fn trace_cpu_metrics(ctx: &TracePointContext) -> Result<(), i64> { - let (bytes_alloc, pid, command) = per_cpu_bytes_alloc(ctx)?; + let (bytes_alloc, tgid, command) = per_cpu_bytes_alloc(ctx)?; //let (cpu_id, cpu_freq) = cpu_frequency(&ctx)?; let cpu_metrics = CpuFrequency { // cpu_id, // cpu_freq, bytes_alloc, - pid, + tgid, command, }; diff --git a/core/src/components/metrics_tracer/src/memory.rs b/core/src/components/metrics_tracer/src/memory.rs index de14c7a..a7ee4d4 100644 --- a/core/src/components/metrics_tracer/src/memory.rs +++ b/core/src/components/metrics_tracer/src/memory.rs @@ -1,4 +1,4 @@ -use aya_ebpf::{EbpfContext, programs::TracePointContext}; +use aya_ebpf::{EbpfContext, helpers::bpf_get_current_pid_tgid, programs::TracePointContext}; /// Read the fields of the `syscalls:sys_enter_mmap` tracepoint. pub fn enter_mmap(ctx: &TracePointContext) -> Result<((u32, u64, u64, [u8; 16])), i64> { @@ -7,7 +7,9 @@ pub fn enter_mmap(ctx: &TracePointContext) -> Result<((u32, u64, u64, [u8; 16])) let addr_offset = 16; let len_offset = 24; - let tgid: u32 = unsafe { ctx.read_at(tgid_offset) }?; + //let tgid: u32 = unsafe { ctx.read_at(tgid_offset) }?; + let pid_tgid: u64 = bpf_get_current_pid_tgid(); + let tgid: u32 = (pid_tgid >> 32) as u32; let addr: u64 = unsafe { ctx.read_at(addr_offset) }?; let len: u64 = unsafe { ctx.read_at(len_offset) }?; let command = ctx.command()?; diff --git a/core/src/components/metrics_tracer/src/mod.rs b/core/src/components/metrics_tracer/src/mod.rs index 56f80cc..e62fb54 100644 --- a/core/src/components/metrics_tracer/src/mod.rs +++ b/core/src/components/metrics_tracer/src/mod.rs @@ -2,3 +2,4 @@ mod bindings; mod cpu; mod data_structures; mod memory; +mod network; diff --git a/core/src/components/metrics_tracer/src/network.rs b/core/src/components/metrics_tracer/src/network.rs new file mode 100644 index 0000000..4cc50bb --- /dev/null +++ b/core/src/components/metrics_tracer/src/network.rs @@ -0,0 +1,217 @@ +use crate::bindings::{self, net_device}; +use crate::data_structures::{NET_METRICS, PacketLossMetrics}; +use crate::data_structures::{ + TASK_COMM_LEN, TIME_STAMP_EVENTS, TIME_STAMP_START, TimeStampEvent, TimeStampStartInfo, +}; +use aya_ebpf::EbpfContext; +use aya_ebpf::helpers::bpf_get_current_pid_tgid; +use aya_ebpf::helpers::generated::{bpf_ktime_get_ns, bpf_perf_event_output}; +use aya_ebpf::helpers::{ + bpf_get_current_comm, bpf_probe_read_kernel, bpf_probe_read_kernel_str_bytes, +}; +use aya_ebpf::macros::{kprobe, map, tracepoint}; +use aya_ebpf::maps::{HashMap, PerfEventArray}; +use aya_ebpf::programs::ProbeContext; +use core::{mem, ptr}; + +const AF_INET: u16 = 2; +const AF_INET6: u16 = 10; +const TCP_SYN_SENT: u8 = 2; + +/// packet loss tracer +pub fn detect_packet_loss(ctx: &ProbeContext) -> Result { + let sk_pointer = ctx.arg::<*const u8>(0).ok_or(1i64)?; + + if sk_pointer.is_null() { + return Err(1); + } + + let tgid = (unsafe { bpf_get_current_pid_tgid() } >> 32) as u32; + let comm = unsafe { bpf_get_current_comm() }.map_err(|_| 1i64)?; + let ts_us: u64 = unsafe { bpf_ktime_get_ns() } / 1_000; + let sk_err_offset = 284; + let sk_err_soft_offset = 600; + let sk_backlog_len_offset = 196; + let sk_write_memory_queued_offset = 376; + let sk_receive_buffer_size_offset = 244; + let sk_ack_backlog_offset = 604; + let sk_drops_offset = 136; + + let sk_err = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_err_offset) as *const i32).map_err(|_| 1)? + }; + let sk_err_soft = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_err_soft_offset) as *const i32) + .map_err(|_| 1)? + }; + let sk_backlog_len = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_backlog_len_offset) as *const i32) + .map_err(|_| 1)? + }; + let sk_write_memory_queued = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_write_memory_queued_offset) as *const i32) + .map_err(|_| 1)? + }; + let sk_receive_buffer_size = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_receive_buffer_size_offset) as *const i32) + .map_err(|_| 1)? + }; + let sk_ack_backlog = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_ack_backlog_offset) as *const u32) + .map_err(|_| 1)? + }; + let sk_drops = unsafe { + bpf_probe_read_kernel::(sk_pointer.add(sk_drops_offset) as *const i32) + .map_err(|_| 1)? + }; + + let packet_loss_metrics = PacketLossMetrics { + tgid: tgid, + comm: comm, + ts_us: ts_us, + sk_err: sk_err, + sk_err_soft: sk_err_soft, + sk_backlog_len: sk_backlog_len, + sk_write_memory_queued: sk_write_memory_queued, + sk_receive_buffer_size: sk_receive_buffer_size, + sk_ack_backlog: sk_ack_backlog, + sk_drops: sk_drops, + }; + + Ok(packet_loss_metrics) +} + +pub fn on_connect(ctx: ProbeContext) -> Result<(), i64> { + let sk = ctx.arg::<*mut bindings::sock>(0).ok_or(1i64)?; + if sk.is_null() { + return Err(1); + } + + let tgid = (unsafe { bpf_get_current_pid_tgid() } >> 32) as u32; + let mut start = TimeStampStartInfo { + comm: [0; TASK_COMM_LEN], + ts_ns: unsafe { bpf_ktime_get_ns() }, + tgid, + }; + unsafe { + let comm_result = bpf_get_current_comm(); + if let Ok(comm) = comm_result { + start.comm.copy_from_slice(&comm); + } + let map_ptr = &raw mut TIME_STAMP_START; + (*map_ptr) + .insert(&(sk as *mut core::ffi::c_void), &start, 0) + .map_err(|_| 1)?; + } + Ok(()) +} + +pub fn on_rcv_state_process(ctx: ProbeContext) -> Result<(), i64> { + // On some kernels, kprobe wrapper puts `sk` at arg0; on others arg1. + let sk = ctx.arg::<*mut bindings::sock>(0).unwrap_or(ptr::null_mut()); + let sk = if sk.is_null() { + ctx.arg::<*mut bindings::sock>(1).ok_or(1i64)? + } else { + sk + }; + + if sk.is_null() { + return Err(1); + } + + let skc_daddr_off = 0; + let skc_rcv_saddr_off = 4; + let skc_dport_off = 12; + let skc_num_off = 14; + let skc_family_off = 16; + let skc_state_off = 18; + let skc_v6_daddr_off = 56; + let skc_v6_rcv_saddr_off = 72; + + let state = unsafe { bpf_probe_read_kernel::((sk as usize + skc_state_off) as *const u8) } + .map_err(|_| 1)?; + + if state != TCP_SYN_SENT { + return Ok(()); + } + + let start = unsafe { + let map_ptr = &raw const TIME_STAMP_START; + (*map_ptr).get(&((sk as usize) as *mut core::ffi::c_void)) + } + .ok_or(1i64)?; + let now = unsafe { bpf_ktime_get_ns() }; + let delta = now as i64 - start.ts_ns as i64; + if delta <= 0 { + unsafe { + let map_ptr = &raw mut TIME_STAMP_START; + let _ = (*map_ptr).remove(&((sk as usize) as *mut core::ffi::c_void)); + } + return Ok(()); + } + + let mut ev = TimeStampEvent { + delta_us: (delta as u64) / 1_000, + ts_us: now / 1_000, + tgid: start.tgid, + comm: start.comm, + lport: 0, + dport_be: 0, + af: 0, + saddr_v4: 0, + daddr_v4: 0, + saddr_v6: [0; 4], + daddr_v6: [0; 4], + }; + + // family, ports + ev.af = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_family_off) as *const u16).map_err(|_| 1)? + }; + ev.lport = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_num_off) as *const u16).map_err(|_| 1)? + }; + ev.dport_be = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_dport_off) as *const u16).map_err(|_| 1)? + }; + + if ev.af == AF_INET { + ev.saddr_v4 = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_rcv_saddr_off) as *const u32) + .map_err(|_| 1)? + }; + ev.daddr_v4 = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_daddr_off) as *const u32) + .map_err(|_| 1)? + }; + } else { + // read 16 bytes as four u32 words + for i in 0..4 { + ev.saddr_v6[i] = unsafe { + bpf_probe_read_kernel::( + (sk as usize + skc_v6_rcv_saddr_off + i * 4) as *const u32, + ) + .map_err(|_| 1)? + }; + ev.daddr_v6[i] = unsafe { + bpf_probe_read_kernel::((sk as usize + skc_v6_daddr_off + i * 4) as *const u32) + .map_err(|_| 1)? + }; + } + } + + // emit + cleanup + unsafe { + bpf_perf_event_output( + ctx.as_ptr(), + &raw const TIME_STAMP_EVENTS as *const _ as *mut _, + 0, // BPF_F_CURRENT_CPU + &ev as *const _ as *mut _, + (mem::size_of::() as u32).into(), + ); + let map_ptr = &raw mut TIME_STAMP_START; + let _ = (*map_ptr).remove(&((sk as usize) as *mut core::ffi::c_void)); + } + + Ok(()) +}