Work in progress. Add endian conversions. Add Rust interface for iterating the flows list. Add a temporary interface that dumps flow data to the console to prove that it works.

src/rust/lqos_sys/src/bpf/common/flows.h

@@ -1,3 +1,5 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+
 // TCP flow monitor system
 
 #include <linux/bpf.h>
@@ -24,6 +26,8 @@ struct flow_key_t {
     __u16 dst_port;
     __u8 protocol;
     __u8 pad;
+    __u8 pad1;
+    __u8 pad2;
 };
 
 // TCP connection flow entry
@@ -108,19 +112,23 @@ static __always_inline struct flow_key_t build_flow_key(
         return (struct flow_key_t) {
             .src = dissector->src_ip,
             .dst = dissector->dst_ip,
-            .src_port = dissector->src_port,
-            .dst_port = dissector->dst_port,
+            .src_port = bpf_htons(dissector->src_port),
+            .dst_port = bpf_htons(dissector->dst_port),
             .protocol = dissector->ip_protocol,
-            .pad = 0
+            .pad = 0,
+            .pad1 = 0,
+            .pad2 = 0
         };
     } else {
         return (struct flow_key_t) {
             .src = dissector->dst_ip,
             .dst = dissector->src_ip,
-            .src_port = dissector->dst_port,
-            .dst_port = dissector->src_port,
+            .src_port = bpf_htons(dissector->dst_port),
+            .dst_port = bpf_htons(dissector->src_port),
             .protocol = dissector->ip_protocol,
-            .pad = 0
+            .pad = 0,
+            .pad1 = 0,
+            .pad2 = 0
         };    
     }
 }
@@ -179,6 +187,7 @@ static __always_inline void process_icmp(
             return;
         }
         data = bpf_map_lookup_elem(&flowbee, &key);
+        if (data == NULL) return;
     }
     update_flow_rates(dissector, direction, data, now);
 }
@@ -199,6 +208,7 @@ static __always_inline void process_udp(
             return;
         }
         data = bpf_map_lookup_elem(&flowbee, &key);
+        if (data == NULL) return;
     }
     update_flow_rates(dissector, direction, data, now);
 }

src/rust/lqos_sys/src/bpf/lqos_kern.c

@@ -422,4 +422,23 @@ int heimdall_reader(struct bpf_iter__bpf_map_elem *ctx) {
     return 0;
 }
 
+SEC("iter/bpf_map_elem")
+int flow_reader(struct bpf_iter__bpf_map_elem *ctx)
+{
+    // The sequence file
+    struct seq_file *seq = ctx->meta->seq;
+    struct flow_data_t *counter = ctx->value;
+    struct flow_key_t *ip = ctx->key;
+
+    // Bail on end
+    if (counter == NULL || ip == NULL) {
+        return 0;
+    }
+
+    //BPF_SEQ_PRINTF(seq, "%d %d\n", counter->next_entry, counter->rtt[0]);
+    bpf_seq_write(seq, ip, sizeof(struct flow_key_t));
+    bpf_seq_write(seq, counter, sizeof(struct flow_data_t));
+    return 0;
+}
+
 char _license[] SEC("license") = "GPL";

src/rust/lqos_sys/src/bpf_iterator.rs

@@ -1,6 +1,5 @@
 use crate::{
-  kernel_wrapper::BPF_SKELETON, lqos_kernel::bpf, HostCounter,
-  RttTrackingEntry, heimdall_data::{HeimdallKey, HeimdallData},
+  flowbee_data::{FlowbeeData, FlowbeeKey}, heimdall_data::{HeimdallData, HeimdallKey}, kernel_wrapper::BPF_SKELETON, lqos_kernel::bpf, HostCounter, RttTrackingEntry
 };
 use lqos_utils::XdpIpAddress;
 use once_cell::sync::Lazy;
@@ -149,7 +148,17 @@ where
           let (_head, values, _tail) =
             unsafe { &value_slice.align_to::<VALUE>() };
 
+          if !key.is_empty() && !values.is_empty() {
             callback(&key[0], &values[0]);
+          } else {
+            log::error!("Empty key or value found in iterator");
+            if key.is_empty() {
+              log::error!("Empty key");
+            }
+            if values.is_empty() {
+              log::error!("Empty value");
+            }
+          }
 
           index += Self::KEY_SIZE + Self::VALUE_SIZE;
         }
@@ -191,6 +200,10 @@ static mut HEIMDALL_TRACKER: Lazy<
   Option<BpfMapIterator<HeimdallKey, HeimdallData>>,
 > = Lazy::new(|| None);
 
+static mut FLOWBEE_TRACKER: Lazy<
+  Option<BpfMapIterator<FlowbeeKey, FlowbeeData>>,
+> = Lazy::new(|| None);
+
 pub unsafe fn iterate_throughput(
   callback: &mut dyn FnMut(&XdpIpAddress, &[HostCounter]),
 ) {
@@ -235,6 +248,9 @@ pub unsafe fn iterate_rtt(
   if let Some(iter) = RTT_TRACKER.as_mut() {
     let _ = iter.for_each(callback);
   }
+
+  // TEMPORARY
+  iterate_flows();
 }
 
 /// Iterate through the heimdall map and call the callback for each entry.
@@ -262,3 +278,31 @@ pub fn iterate_heimdall(
     }
   }
 }
+
+/// Iterate through the Flows 2 system tracker, retrieving all flows
+pub fn iterate_flows() {
+  unsafe {
+    if FLOWBEE_TRACKER.is_none() {
+      let lock = BPF_SKELETON.lock().unwrap();
+      if let Some(skeleton) = lock.as_ref() {
+        let skeleton = skeleton.get_ptr();
+        if let Ok(iter) = unsafe {
+          BpfMapIterator::new(
+            (*skeleton).progs.flow_reader,
+            (*skeleton).maps.flowbee,
+          )
+        } {
+          *FLOWBEE_TRACKER = Some(iter);
+        }
+      }
+    }
+  
+    let mut callback = |key: &FlowbeeKey, data: &FlowbeeData| {
+      log::info!("Flow: {:#?} -> {:#?}", key, data);
+    };
+
+    if let Some(iter) = FLOWBEE_TRACKER.as_mut() {
+      let _ = iter.for_each(&mut callback);
+    }
+  }
+}

src/rust/lqos_sys/src/flowbee_data.rs

@@ -0,0 +1,63 @@
+use lqos_utils::XdpIpAddress;
+use zerocopy::FromBytes;
+
+/// Representation of the eBPF `flow_key_t` type.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Hash, FromBytes)]
+#[repr(C)]
+pub struct FlowbeeKey {
+  /// Mapped `XdpIpAddress` source for the flow.
+  pub remote_ip: XdpIpAddress,
+  /// Mapped `XdpIpAddress` destination for the flow
+  pub local_ip: XdpIpAddress,
+  /// Source port number, or ICMP type.
+  pub src_port: u16,
+  /// Destination port number.
+  pub dst_port: u16,
+  /// IP protocol (see the Linux kernel!)
+  pub ip_protocol: u8,
+  /// Padding to align the structure to 16 bytes.
+  padding: u8,
+  padding1: u8,
+  padding2: u8,
+}
+
+/// Mapped representation of the eBPF `flow_data_t` type.
+#[derive(Debug, Clone, Default, FromBytes)]
+#[repr(C)]
+pub struct FlowbeeData {
+  /// Time (nanos) when the connection was established
+  pub start_time: u64,
+  /// Time (nanos) when the connection was last seen
+  pub last_seen: u64,
+  /// Bytes transmitted
+  pub bytes_sent: [u64; 2],
+  /// Packets transmitted
+  pub packets_sent: [u64; 2],
+  /// Clock for the next rate estimate
+  pub next_count_time: [u64; 2],
+  /// Clock for the previous rate estimate
+  pub last_count_time: [u64; 2],
+  /// Bytes at the next rate estimate
+  pub next_count_bytes: [u64; 2],
+  /// Rate estimate
+  pub rate_estimate_bps: [u64; 2],
+  /// Sequence number of the last packet
+  pub last_sequence: [u32; 2],
+  /// Acknowledgement number of the last packet
+  pub last_ack: [u32; 2],
+  /// Retry Counters
+  pub retries: [u32; 2],
+  /// Timestamp values
+  pub tsval: [u32; 2],
+  /// Timestamp echo values
+  pub tsecr: [u32; 2],
+  /// When did the timestamp change?
+  pub ts_change_time: [u64; 2],
+  /// When should we calculate RTT (to avoid flooding)
+  pub ts_calc_time: [u64; 2],
+  /// Most recent RTT
+  pub last_rtt: [u64; 2],
+  /// Has the connection ended?
+  /// 0 = Alive, 1 = FIN, 2 = RST
+  pub end_status: u32,
+}

src/rust/lqos_sys/src/lib.rs

@@ -22,6 +22,7 @@ mod bpf_iterator;
 /// Data shared between eBPF and Heimdall that needs local access
 /// for map control.
 pub mod heimdall_data;
+pub mod flowbee_data;
 
 pub use ip_mapping::{
   add_ip_to_tc, clear_ips_from_tc, del_ip_from_tc, list_mapped_ips,