Add feature to read kernel tcp socket metrics (#2)

* Add first tcp_sock_tracer

* First working tcp kernel sockets

* Fix IP read

* Add arrayref to cargo

* Add db/Cargo.lock to gitignore

* Working concept

* Add tcp_sock tracing fields

---------

Co-authored-by: Moritz Flüchter <moritz.fluechter@uni-tuebingen.de>

Author: MoritzFlu

db/.gitignore

@@ -1,2 +1,3 @@
 target/*
 *.sqlite
+Cargo.lock

db/Cargo.toml

@@ -17,6 +17,7 @@ sqlite = { version = "0.36.1", features = ["bundled"] }
 indicatif = "0.17.11"
 env_logger = "0.11.6"
 aya-ebpf = "0.1.1"
+arrayref = "0.3.9"
 
 [net]
 net.git-fetch-with-cli= true 

db/src/bindings/sock.rs

@@ -0,0 +1,125 @@
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+use ts_storage::{DataValue, IpTuple};
+
+use crate::{db_writer::DBOperation, flow_tracker::{EventIndexer, AF_INET}, reader::FromBuffer};
+
+use arrayref::array_ref;
+
+#[repr(C)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash, Default)]
+pub struct sock_trace_entry {
+    pub time: u64,
+    pub addr_v4: u64,
+    pub src_v6: [u8; 16],
+    pub dst_v6: [u8; 16],
+    pub ports: u32,
+    pub family: u16,
+    // SOCK Stats
+    pub pacing_rate: u64,
+    pub max_pacing_rate: u64,
+    // INET_CONN Stats
+    pub backoff: u8,
+    pub rto: u32,
+    // INET_CONN -> icsk_ack
+    pub ato: u32,
+    pub rcv_mss: u16,
+    // TCP_SOCK Stats
+    pub snd_cwnd: u32,
+    pub bytes_acked: u64,
+    pub snd_ssthresh: u32,
+    pub total_retrans: u32,
+    pub probes: u8,
+    pub lost: u32,
+    pub sacked_out: u32,
+    pub retrans: u32,
+    pub rcv_ssthresh: u32,
+    pub rttvar: u32,
+    pub advmss: u16,
+    pub reordering: u32,
+    pub rcv_rtt: u32,
+    pub rcv_space: u32,
+    pub bytes_received: u64,
+    pub segs_out: u32,
+    pub segs_in: u32,
+    // TCP_SOCK -> tcp_options_received
+    pub snd_wscale: u16,
+    pub rcv_wscale: u16,
+    pub div: u32
+}
+
+impl FromBuffer for sock_trace_entry {
+    fn from_buffer(buf: &Vec<u8>) -> Self {
+        unsafe { *(buf.as_ptr() as *const sock_trace_entry) }
+
+    }
+}
+
+impl EventIndexer for sock_trace_entry {
+    fn get_field(&self, index: usize) -> Option<DataValue> {
+        match index {
+            _ => None, // TODO: better error handling
+        }
+    }
+    fn get_default_field(&self, index: usize) -> DataValue {
+        match index {
+            _ => panic!("Tried to access out of bounds index!"), // TODO: better error handling
+        }
+    }
+    fn get_field_name(&self, index: usize) -> &str {
+        match index {
+            _ => panic!("Tried to access out of bounds index!"), // TODO: better error handling
+        }
+    }
+    fn get_ip_tuple(&self) -> IpTuple {
+        let src: IpAddr;
+        let dst: IpAddr;
+
+        //print!("Family: {}",self.family);
+
+
+        if self.family == AF_INET {
+            // TODO: check offsets
+            let mut bytes = self.addr_v4.to_be_bytes();
+
+            let mut srcbytes = array_ref![bytes,0,4].clone();
+            let mut dstbytes = array_ref![bytes,4,4].clone();
+            //srcbytes.reverse();
+
+            srcbytes.reverse();
+            dstbytes.reverse();
+            src = IpAddr::V4(Ipv4Addr::from(srcbytes));
+            dst = IpAddr::V4(Ipv4Addr::from(dstbytes));
+        } else {
+            src = IpAddr::V6(Ipv6Addr::from(self.src_v6));
+            dst = IpAddr::V6(Ipv6Addr::from(self.dst_v6));
+        }
+
+        let port_bytes = self.ports.to_be_bytes();
+
+        let srcbytes = array_ref![port_bytes,0,2].clone();
+        let dstbytes = array_ref![port_bytes,2,2].clone();
+
+        // TODO: check byte order if ports are correct
+        // Dport could be be bytes
+        let sport = u16::from_le_bytes(srcbytes);
+        let dport = u16::from_le_bytes(dstbytes);
+
+        IpTuple {
+            src: src,
+            dst: dst,
+            sport: sport as i64,
+            dport: dport as i64,
+            l4proto: 6,
+        }
+    }
+    fn get_max_index(&self) -> usize {
+        0
+    }
+    fn get_timestamp(&self) -> f64 {
+        self.time as f64
+    }
+    fn as_db_op(self) -> DBOperation {
+        DBOperation::Socket(self)
+    }
+}

db/src/bindings/tcp_probe.rs

@@ -83,7 +83,7 @@ impl EventIndexer for TcpProbe {
     fn get_ip_tuple(&self) -> IpTuple {
         let src: IpAddr;
         let dst: IpAddr;
-
+        
         if self.family == AF_INET {
             //IPv4
             src = IpAddr::V4(Ipv4Addr::from(shorten_to_ipv4(self.saddr)));

db/src/db_writer.rs

@@ -4,12 +4,13 @@ use log::error;
 use ts_storage::{database_factory, sqlite::SQLiteTSDB, DBBackend, IpTuple, TSDBInterface};
 use tokio::sync::mpsc::Receiver;
 
-use crate::{bindings::{tcp_packet::TcpPacket, tcp_probe::TcpProbe}, flow_tracker::{EventIndexer, EventType, FlowTracker}};
+use crate::{bindings::{sock::sock_trace_entry, tcp_packet::TcpPacket, tcp_probe::TcpProbe}, flow_tracker::{EventIndexer, EventType, FlowTracker}};
 
 #[derive(Debug)]
 pub enum DBOperation {
     Packet(TcpPacket),
-    Probe(TcpProbe)
+    Probe(TcpProbe),
+    Socket(sock_trace_entry)
 }
 
 
@@ -94,7 +95,24 @@ impl DBWriter {
 
                     //println!("Probe {:?}",data.ssthresh);
 
+                },
+                DBOperation::Socket(sock) => {
+                    if sock.div != 0xffffffff {
+                        println!(
+                            "Malformed!: {:?},div: {}",sock,sock.div
+                        );
+                        panic!("Malformed packet!");
+                    } else {
+                        if sock.family == 2 && sock.snd_cwnd > 0 {
+                            println!("Time: {}, Stream: {:?} CWND: {:?}",sock.time,sock.get_ip_tuple(),sock.snd_cwnd.to_le_bytes());
+                        }
+                        //println!(
+                        //    "Packet!: {:?}, CWND: {}",sock.get_ip_tuple(),sock.snd_cwnd                        );
+                    }
+                    //let a = sock.get_ip_tuple();
+                    //println!("Socket: {:?}",sock.get_ip_tuple());
                 }
+
             }
         }
 

db/src/main.rs

@@ -6,9 +6,10 @@ mod bindings {
     pub mod tcp_packet;
     pub mod tcp_probe;
     pub mod ctypes;
+    pub mod sock;
 }
 
-use bindings::{tcp_packet::TcpPacket, tcp_probe::TcpProbe};
+use bindings::{sock::sock_trace_entry, tcp_packet::TcpPacket, tcp_probe::TcpProbe};
 use db_writer::{DBOperation, DBWriter};
 use flow_tracker::{EventIndexer, EventType, FlowTracker, TsTracker};
 use log::{error, info};
@@ -99,9 +100,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     // Start all tasks
     let threads = vec![
-        start_file_reader::<TcpPacket>("/tmp/xdp.tcp", tx.clone(), stop_token.clone()).await,
-        start_file_reader::<TcpPacket>("/tmp/tc.tcp", tx.clone(), stop_token.clone()).await,
-        start_file_reader::<TcpProbe>("/tmp/probe.tcp", tx.clone(), stop_token.clone()).await,
+        //start_file_reader::<TcpPacket>("/tmp/xdp.tcp", tx.clone(), stop_token.clone()).await,
+        //start_file_reader::<TcpPacket>("/tmp/tc.tcp", tx.clone(), stop_token.clone()).await,
+        //start_file_reader::<TcpProbe>("/tmp/probe.tcp", tx.clone(), stop_token.clone()).await,
+        start_file_reader::<sock_trace_entry>("/tmp/sock.tcp", tx.clone(), stop_token.clone()).await
     ];
 
     // Wait for file threads to finish!

db/src/reader.rs

@@ -54,6 +54,7 @@ impl<T: EventIndexer + Debug + FromBuffer> FileReader<T> {
         // TODO: I DONT KNOW WHY THIS 4 BYTE MISALIGNMENT HAPPENS, IT JUST DOES
         // FIX THIS!
         let entry_size = core::mem::size_of::<T>() - 4;
+        //let entry_size = 68;
 
         debug!("Entry size: {} bytes for {}", entry_size, self.path);
 

tcbee/Cargo.lock

@@ -1,2 +1,2 @@
 #!/bin/bash
-RUST_LOG=debug cargo run --release --config 'target."cfg(all())".runner="sudo -E"'  -- -q lo
+RUST_LOG=info cargo run --release --config 'target."cfg(all())".runner="sudo -E"'  -- -q lo

tcbee/debu_run.sh

@@ -6,13 +6,15 @@ pub mod eth_header;
 pub mod ip4_header;
 pub mod ip6_header;
 pub mod flow;
+pub mod tcp_sock;
 
 #[cfg(feature = "user")]
 use aya::Pod;
 use tcp_bad_csum::tcp_bad_csum_entry;
 use tcp_probe::tcp_probe_entry;
 use tcp_retransmit_synack::tcp_retransmit_synack_entry;
 use tcp_header::tcp_packet_trace;
+use tcp_sock::sock_trace_entry;
 
 #[repr(C)]
 #[derive(Default)]
@@ -82,10 +84,18 @@ impl EBPFTracePointType for tcp_bad_csum_entry {
     const NAME: &'static str = "tcp_bad_csum";
     const CATEGORY: &'static str = "tcp";
 }
+
+impl EBPFTracePointType for sock_trace_entry {
+    const QUEUE_NAME: &'static str = "TCP_SOCK";
+    const NAME: &'static str = "tcp_sock";
+    const CATEGORY: &'static str = "tcp";
+}
 #[cfg(feature = "user")]
 // Needed to be able to parse as queue entry from eBPF queue
 unsafe impl Pod for tcp_probe_entry {}
 #[cfg(feature = "user")]
 unsafe impl Pod for tcp_retransmit_synack_entry {}
 #[cfg(feature = "user")]
-unsafe impl Pod for tcp_bad_csum_entry {}
+unsafe impl Pod for tcp_bad_csum_entry {}
+#[cfg(feature = "user")]
+unsafe impl Pod for sock_trace_entry {}