Add support for DuckDB Backend, some fixes to recording (#9)

* Implement DuckDB CUrsor

* Storing current state, not working

* It works but values written are wrong for time series

* Change sequence creation to IF EXISTS

* Fix kernel trace file names in process

* Add duckdb to viz tool

* Add duckdb to path selector

* Improve flow selection

* Fix dport mapping for sock recording

* Udpate

* Fix DataValue parsing from DuckDB Union

* Update README.md

* Move queries to extra file

* Add test db files to gitignore

* Add duckdb dependency to github workflow

* Switch to bundled version

---------

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

Author: MoritzFlu

.github/workflows/tcbee.yml

@@ -19,7 +19,7 @@ jobs:
     - name: Install Dependencies
       run: |
         sudo apt update
-        sudo apt install -y llvm clang libelf-dev libclang-dev pkg-config fontconfig libfontconfig1-dev 
+        sudo apt install -y llvm clang libelf-dev libclang-dev pkg-config fontconfig libfontconfig1-dev
         cargo install bpf-linker
 
     - name: TCBee-Record Build

README.md

@@ -40,10 +40,9 @@ The current Todo-List includes
 
 - Documentation for the tools and interfaces
 - Add plugins for the calculation of common TCP congestion metrics
-- Implement InfluxDB interface for faster processing 
+- Add full IPv6 support for kernel hooks
 - Test and benchmark bottlenecks (eBPF Ringbuf size, File writer, etc.)
 - Cleanup of eBPF and user space code
-- Change UI based on selected traces
 - ...
 
 The current version is tested for linux kernel 6.13.6 and may not work on older or newer kernel versions.
@@ -56,7 +55,7 @@ TCBee
 
 * provides a command-line program to record flows and track current data rates
 * monitors both packet headers for incoming and outgoing packets
-* hooks onto the linux kernel functions `tcp_sendmsg` and `tcp_recvmsg` to read kernel metrics
+* hooks onto the linux kernel tcp sending and receiving functions to read kernel metrics **per packet**
 * stores recorded data in a structured flow database
 * provides a simple plugin interface to calculate metrics from recorded data and save the results
 * comes with a visualization tool to analyse and compare TCP flow metrics

tcbee-process/run.sh

@@ -1,3 +1,3 @@
 #!/bin/bash
 # Runs the rust program as sudo, needed privileges
-RUST_LOG=debug cargo run --release --config 'target."cfg(all())".runner="sudo -E"'
+RUST_LOG=debug cargo run --release -- --output ./db.duck -t

tcbee-process/src/bindings/sock.rs

@@ -186,7 +186,7 @@ impl EventIndexer for sock_trace_entry {
 
         // TODO: check byte order if ports are correct
         // Dport could be be bytes
-        let sport = u16::from_le_bytes(srcbytes);
+        let sport = u16::from_be_bytes(srcbytes);
         let dport = u16::from_le_bytes(dstbytes);
 
         IpTuple {

tcbee-process/src/db_writer.rs

@@ -28,12 +28,12 @@ pub struct DBWriter {
 
 impl DBWriter {
     pub fn new(
-        file: &str,
+        backend: DBBackend,
         rx: Receiver<DBOperation>,
         status: ProgressBar,
     ) -> Result<DBWriter, Box<dyn Error>> {
         let db: Box<dyn TSDBInterface + Send> =
-            database_factory::<SQLiteTSDB>(DBBackend::SQLite(file.to_string()))?;
+            database_factory::<SQLiteTSDB>(backend)?;
 
         let streams: HashMap<IpTuple, FlowTracker> = HashMap::new();
 

tcbee-process/src/main.rs

@@ -10,7 +10,7 @@ mod bindings {
     pub mod cwnd;
 }
 
-use argparse::{ArgumentParser, Store};
+use argparse::{ArgumentParser, Store, StoreOption, StoreTrue};
 use bindings::{sock::sock_trace_entry, cwnd::cwnd_trace_entry, tcp_packet::TcpPacket, tcp_probe::TcpProbe};
 use db_writer::{DBOperation, DBWriter};
 use flow_tracker::{EventIndexer, EventType, FlowTracker, TsTracker};
@@ -24,6 +24,7 @@ use tokio::{
     task::{self, JoinHandle},
 };
 use tokio_util::sync::CancellationToken;
+use ts_storage::DBBackend;
 
 use core::num;
 use std::{
@@ -95,17 +96,60 @@ async fn start_file_reader<
 async fn main() -> Result<(), Box<dyn Error>> {
     env_logger::init();
 
-    let mut dir: String = "/tmp/".to_string();
+    let mut source: String = "/tmp/".to_string();
+    let mut output: String = "".to_string();
+    let mut sqlite: bool = false;
+    let mut duckdb: bool = false;
+
     {
         let mut argparser = ArgumentParser::new();
-        argparser.refer(&mut dir).add_option(
-            &["-d", "--dir"],
+        argparser.refer(&mut source).add_option(
+            &["-s", "--source"],
+            Store,
+            "Directory to search for TCBee recording *.tcp files!",
+        );
+        argparser.refer(&mut output).add_option(
+            &["-o", "--output"],
             Store,
-            "Directory to read recording results from. Defaults to /tmp/",
+            "Path for outoput database file",
+        );
+        argparser.refer(&mut sqlite).add_option(
+            &["-q", "--sqlite"],
+            StoreTrue,
+            "Store result to SQLITE",
         );
+        argparser.refer(&mut duckdb).add_option(
+            &["-d", "--duckdb"],
+            StoreTrue,
+            "Store result to DuckDB, better performance",
+        );
+
         argparser.parse_args_or_exit();
     }
 
+    if !sqlite && !duckdb {
+        print!("Please select either --sqlite or --duckdb");
+        return Ok(());
+    }
+    if sqlite && duckdb {
+        print!("Please select either --sqlite or --duckdb");
+        return Ok(());
+    }
+
+    if output.is_empty() {
+        if sqlite {
+            output = "/tmp/db.sqlite".to_string();
+        }
+        if duckdb {
+            output = "/tmp/db.duck".to_string();
+        }
+    }
+
+    let mut backend = DBBackend::SQLite(output.clone());
+    if duckdb {
+        backend = DBBackend::DuckDB(output);
+    }
+
     let progress_bars = MultiProgress::new();
 
     let status = progress_bars.add(ProgressBar::new(5));
@@ -123,7 +167,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     println!("Starting readers, initial processing may be slow due to setup of streams!");
 
     // Create DB Backend handler
-    let db_res = DBWriter::new("db.sqlite", rx,status);
+    let db_res = DBWriter::new(backend, rx,status);
     if db_res.is_err() {
         panic!("Could not open Database! Error: {}", db_res.err().unwrap())
     }
@@ -146,49 +190,49 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     let threads = vec![
         start_file_reader::<TcpPacket>(
-            prepend_string("xdp.tcp".to_string(),&dir),
+            prepend_string("xdp.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<TcpPacket>(
-            prepend_string("tc.tcp".to_string(),&dir),
+            prepend_string("tc.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<TcpProbe>(
-            prepend_string("probe.tcp".to_string(),&dir),
+            prepend_string("probe.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<sock_trace_entry>(
-            prepend_string("sock_send.tcp".to_string(),&dir),
+            prepend_string("send_sock.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<sock_trace_entry>(
-            prepend_string("sock_recv.tcp".to_string(),&dir),
+            prepend_string("recv_sock.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<cwnd_trace_entry>(
-            prepend_string("recv_cwnd.tcp".to_string(),&dir),
+            prepend_string("recv_cwnd.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,
         )
         .await,
         start_file_reader::<cwnd_trace_entry>(
-            prepend_string("send_cwnd.tcp".to_string(),&dir),
+            prepend_string("send_cwnd.tcp".to_string(),&source),
             tx.clone(),
             stop_token.clone(),
             &progress_bars,

tcbee-viz/src/modules/backend/intermediate_backend.rs

@@ -16,7 +16,7 @@ use crate::modules::{
 use crate::TSDBInterface;
 use iced::widget::canvas::Cache;
 use plotters::style::RGBAColor;
-use ts_storage::{database_factory, sqlite::SQLiteTSDB, DBBackend, DataValue, Flow};
+use ts_storage::{database_factory, duckdb::DuckDBTSDB, sqlite::SQLiteTSDB, DBBackend, DataValue, Flow};
 use ts_storage::{DataPoint, TimeSeries};
 use std::{cell::RefCell, f64::{MAX, MIN}, path::PathBuf, slice::Iter, sync::RwLock};
 
@@ -29,6 +29,7 @@ use super::{app_settings::ApplicationSettings, struct_tcp_flow_wrapper::TcpFlowW
 pub enum DataSource {
     Influx,
     Sqllite,
+    DuckDB,
     None,
 }
 
@@ -45,6 +46,7 @@ impl ToString for DataSource {
         match self {
             DataSource::Influx => String::from("Influx"),
             DataSource::Sqllite => String::from("Sqllite"),
+            DataSource::DuckDB => String::from("DuckDB"),
             DataSource::None => String::from("Nothing selected"),
         }
     }
@@ -245,9 +247,10 @@ impl IntermediateBackend {
     // DEBUG
     pub fn receive_flow_formatted(&self, flow: &Flow) -> String {
         format!(
-            "ID:{:?} Port: (src: {:?}) IP(src: {:?} : dst: {:?}) ",
+            "ID:{:?} Port: {:?}-{:?} IP: {:?}-{:?}",
             flow.get_id().unwrap(),
             flow.tuple.sport,
+            flow.tuple.dport,
             flow.tuple.src,
             flow.tuple.dst
         )
@@ -276,7 +279,19 @@ impl IntermediateBackend {
                     database_factory::<SQLiteTSDB>(DBBackend::SQLite(path_db.clone()))
                         .expect("could not parse database"),
                 );
-                println!("initialized database of time {:?}", source);
+                println!("initialized SQLITE database of time {:?}", source);
+                IntermediateBackend {
+                    source_type: source.clone(),
+                    database_interface: Some(db_interface),
+                    database_path: Some(PathBuf::from(path_db))
+                }
+            },
+            DataSource::DuckDB => {
+                let db_interface: Arc<Box<dyn TSDBInterface>> = Arc::new(
+                    database_factory::<DuckDBTSDB>(DBBackend::DuckDB(path_db.clone()))
+                        .expect("could not parse database"),
+                );
+                println!("initialized DUCKDB database of time {:?}", source);
                 IntermediateBackend {
                     source_type: source.clone(),
                     database_interface: Some(db_interface),

tcbee-viz/src/modules/backend/lib_system_io.rs

@@ -10,6 +10,7 @@ pub fn receive_source_from_path(path: &PathBuf) -> Option<DataSource> {
     match extension {
         "sqlite" => Some(DataSource::Sqllite),
         "influx" => Some(DataSource::Influx),
+        "duck" => Some(DataSource::DuckDB),
         _ => None,
     }
 }

tcbee-viz/src/modules/ui/lib_screens/screen_home.rs

@@ -71,7 +71,7 @@ impl ScreenHome {
         let headline = text("Selecting Database").size(TEXT_HEADLINE_0_SIZE);
         let description = text(
             "
-This program aims to replace tcptrace+xplot for analyzing PCAP-Files containing TCP-Flows.\n
+This program aims to visualize recorded metrics TCP-Flows for an explorative analysis.\n
 Its able to visualize single and multiple flows with their corresponding timeseries information.\n
 Modification of the database is supplied via an extendable feature-system.\n
             ",
@@ -149,7 +149,7 @@ impl ScreenHome {
                 // FIXME maybe async this operation?
 
                 let file_selection = FileDialog::new()
-                    .add_filter("text", &["sqlite"])
+                    .add_filter("*.sqlite or *.duck", &["sqlite","duck"])
                     .set_directory("~/")
                     .pick_file();
                 // FIXME improve error handling

tcbee-viz/src/modules/ui/lib_widgets/app_widgets.rs

@@ -6,31 +6,29 @@
 
 use std::sync::{Arc, RwLock};
 
-use crate::modules::{
+use crate::{modules::{
     backend::{
         app_settings::ApplicationSettings, intermediate_backend::IntermediateBackend, lib_system_io::receive_file_metadata, plot_data_preprocessing::convert_rgba_to_iced_color, struct_tcp_flow_wrapper::TcpFlowWrapper
     },
     ui::{
         lib_styling::app_style_settings::{
-            HORIZONTAL_LINE_SECONDARY_HEIGHT, PADDING_AROUND_CONTENT, SLIDER_STEP_SIZE,
-            SPACE_BETWEEN_ELEMENTS, SPLIT_CHART_MAX_HEIGHT, SPLIT_CHART_MIN_HEIGHT,
-            TEXT_HEADLINE_0_SIZE, TEXT_HEADLINE_1_SIZE, TEXT_HEADLINE_2_SIZE,
+            HORIZONTAL_LINE_PRIMARY_HEIGHT, HORIZONTAL_LINE_SECONDARY_HEIGHT, PADDING_AROUND_CONTENT, SLIDER_STEP_SIZE, SPACE_BETWEEN_ELEMENTS, SPLIT_CHART_MAX_HEIGHT, SPLIT_CHART_MIN_HEIGHT, TEXT_HEADLINE_0_SIZE, TEXT_HEADLINE_1_SIZE, TEXT_HEADLINE_2_SIZE
         },
         lib_widgets::lib_graphs::{
             struct_processed_plot_data::ProcessedPlotData,
             struct_string_series_wrapper::{view_wrapper, StringSeriesWrapper},
             struct_zoom_bounds::ZoomBound2D,
         },
     },
-};
+}, Message};
 use iced::{
-    widget::{
+    theme::palette::Background, widget::{
         button, checkbox, radio, scrollable, slider, text, Button, Checkbox, Column, Row, Rule,
-        Space,
-    },
-    Alignment, Element, Length,
+        Space, Text,
+    }, Alignment, Element, Length
 };
 
+/// OTHER FUNCTIONS
 pub fn display_current_mouse_position<'a, Message: 'a>(
     maybe_position: Option<(f64, f64)>,
 ) -> Column<'a, Message> {
@@ -82,18 +80,38 @@ fn generate_selections_for_flows<'a, Message: 'a + Clone>(
     let interface = &ref_backend.database_interface;
 
     if let Some(db_connection) = interface {
+
+        let header_1: Row<'_, _> = Row::<Message>::new()
+            .push(Text::new("ID").width(Length::FillPortion(1))) // Space for radio button
+            .push(Text::new("Source").width(Length::FillPortion(3)))
+            .push(Text::new("Destination").width(Length::FillPortion(3)));
+
+
+        new_col = new_col.push(header_1);
+
         let all_flows = db_connection.list_flows().expect("could not find flows");
         for entry in all_flows {
-            new_col = new_col.push(
-                radio(
-                    ref_backend.receive_flow_formatted(&entry),
-                    // self.backend_interface.receive_flow_formatted(&entry),
-                    entry.get_id().expect("no flow id").clone(),
+
+            let tuple = &entry.tuple;
+            let first_flow_row = Row::<Message>::new()
+                .push(Text::new(entry.get_id().unwrap()).width(Length::FillPortion(1)))
+                .push(Text::new(tuple.src.to_string()).width(Length::FillPortion(3)))
+                .push(Text::new(tuple.dst.to_string()).width(Length::FillPortion(3)));
+            let second_flow_row = Row::<Message>::new()
+                .push(
+                    radio(
+                    "",
+                    entry.get_id().expect("no flow id"),
                     focused_flow.flow_id,
-                    // self.backend_interface.receive_active_flow_id(),
                     message_on_click.clone(),
-                ), //   text(format!("flow: {:?}",entry))
-            );
+                    ) 
+                    .width(Length::FillPortion(1))
+                )
+                .push(Text::new(tuple.sport.to_string()).width(Length::FillPortion(3)))
+                .push(Text::new(tuple.dport.to_string()).width(Length::FillPortion(3)));
+                
+
+            new_col = new_col.push(Rule::horizontal(HORIZONTAL_LINE_SECONDARY_HEIGHT)).push(first_flow_row).push(second_flow_row);
         }
     }
     new_col