Not fully tested yet: websocket version of the ingestor.

src/rust/Cargo.lock

@@ -2105,6 +2105,7 @@ dependencies = [
  "tower-http",
  "tracing",
  "tracing-subscriber",
+ "tungstenite",
  "ureq",
  "uuid",
  "zerocopy 0.8.9",
@@ -3850,6 +3851,7 @@ dependencies = [
  "http 1.1.0",
  "httparse",
  "log",
+ "native-tls",
  "rand",
  "sha1",
  "thiserror",

src/rust/lqosd/Cargo.toml

@@ -52,6 +52,7 @@ sha256 = "1.5.0"
 uuid = { version = "1.10.0", features = ["v4", "fast-rng", "serde"] }
 dryoc = {  version ="0.6.0", features = ["serde"] }
 miniz_oxide = "0.8.0"
+tungstenite = { version = "0", features = [ "native-tls" ] } # For WebSockets
 
 # For memory debugging
 allocative = {  version = "0.3.3", features = [ "dashmap" ] }

src/rust/lqosd/src/lts2_sys/lts2_client/ingestor.rs

@@ -10,7 +10,6 @@ pub(crate) use permission::check_submit_permission;
 use crate::lts2_sys::lts2_client::ingestor::commands::IngestorCommand;
 use crate::lts2_sys::lts2_client::ingestor::message_queue::MessageQueue;
 use crate::lts2_sys::lts2_client::ingestor::permission::is_allowed_to_submit;
-use crate::lts2_sys::lts2_client::nacl_blob::KeyStore;
 
 pub fn start_ingestor() -> Sender<IngestorCommand> {
     println!("Starting ingestor");
@@ -24,9 +23,8 @@ fn ingestor_loop(
     rx: std::sync::mpsc::Receiver<IngestorCommand>,
 ) {
     let message_queue = Arc::new(Mutex::new(MessageQueue::new()));
-    let keys = Arc::new(KeyStore::new());
     let my_message_queue = message_queue.clone();
-    std::thread::spawn(move || ticker_timer(my_message_queue, keys.clone()));
+    std::thread::spawn(move || ticker_timer(my_message_queue));
 
     println!("Starting ingestor loop");
     while let Ok(msg) = rx.recv() {
@@ -36,7 +34,7 @@ fn ingestor_loop(
     println!("Ingestor loop exited");
 }
 
-fn ticker_timer(message_queue: Arc<Mutex<MessageQueue>>, keys: Arc<KeyStore>) {
+fn ticker_timer(message_queue: Arc<Mutex<MessageQueue>>) {
     let mut tfd = TimerFd::new().unwrap();
     assert_eq!(tfd.get_state(), TimerState::Disarmed);
     tfd.set_state(TimerState::Periodic{
@@ -55,7 +53,7 @@ fn ticker_timer(message_queue: Arc<Mutex<MessageQueue>>, keys: Arc<KeyStore>) {
         let mut message_queue_lock = message_queue.lock().unwrap();
         if !message_queue_lock.is_empty() && permitted {
             let start = std::time::Instant::now();
-            if let Err(e) = message_queue_lock.send(keys.clone()) {
+            if let Err(e) = message_queue_lock.send() {
                 println!("Failed to send queue: {e:?}");
             }
             println!("Queue send took: {:?}s", start.elapsed().as_secs_f32());

src/rust/lqosd/src/lts2_sys/lts2_client/ingestor/message_queue.rs

@@ -10,15 +10,13 @@ mod site_rtt;
 mod site_cake_drops;
 mod site_cake_marks;
 
-use std::net::TcpStream;
-use anyhow::Result;
-use nacl_blob::KeyStore;
-use std::sync::Arc;
-use std::time::Duration;
+use anyhow::{anyhow, Result};
+use serde::{Deserialize, Serialize};
+use tracing::warn;
 use uuid::Uuid;
 use lqos_config::load_config;
 use crate::lts2_sys::lts2_client::ingestor::commands::IngestorCommand;
-use crate::lts2_sys::lts2_client::{get_remote_host, nacl_blob, remote_commands};
+use crate::lts2_sys::lts2_client::{get_remote_host, remote_commands};
 use crate::lts2_sys::RemoteCommand;
 use crate::lts2_sys::shared_types::{CircuitCakeDrops, CircuitCakeMarks, CircuitRetransmits, CircuitRtt, CircuitThroughput, IngestSession, SiteCakeDrops, SiteCakeMarks, SiteRetransmits, SiteRtt, SiteThroughput};
 
@@ -101,74 +99,147 @@ impl MessageQueue {
         }
     }
 
-    pub(crate) fn send(&mut self, keys: Arc<KeyStore>) -> Result<()> {
+    pub(crate) fn send(&mut self) -> Result<()> {
         let config = load_config()?;
         if !config.long_term_stats.use_insight.unwrap_or(false) {
             self.clear();
             return Ok(());
         }
 
-        use std::net::ToSocketAddrs;
         let remote_host = get_remote_host();
-        let target = &format!("{}:9121", remote_host);
-        let mut to = target.to_socket_addrs()?;
-        let to = to.next().ok_or_else(|| anyhow::anyhow!("Failed to resolve remote host"))?;
-        if let Ok(mut socket) = TcpStream::connect_timeout(&to, Duration::from_secs(5)) {
-            socket.set_nodelay(true)?;
-            if let Err(e) = nacl_blob::transmit_hello(&keys, 0x2763, 1, &mut socket) {
-                println!("Failed to send hello to ingestion server. {e:?}");
+        let target = &format!("ws://{}:9121", remote_host);
+        let Ok((mut socket, _response)) = tungstenite::connect(target) else {
+            warn!("Failed to connect to ingestion server");
+            return Ok(());
+        };
+
+        // Send Hello
+        let Ok((_, _, magic_to_send)) = (WsMessage::Hello { magic: 0x2763 }).to_bytes() else {
+            warn!("Failed to serialize hello message");
+            return Ok(());
+        };
+        if let Err(e) = socket.send(tungstenite::Message::Binary(magic_to_send)) {
+            warn!("Failed to send hello message to server: {}", e);
+            return Ok(());
+        }
+
+        // Wait for Hello Back
+        let Ok(reply) = socket.read() else {
+            warn!("Failed to receive hello response from server");
+            return Ok(());
+        };
+        let Ok(reply) = WsMessage::from_bytes(&reply.into_data()) else {
+            warn!("Failed to deserialize hello response from server");
+            return Ok(());
+        };
+        match reply {
+            WsMessage::Hello { magic } => {
+                if magic != 0x3672 {
+                    warn!("Received invalid magic number from server: {}", magic);
+                    return Ok(());
+                }
+            }
+            _ => {
+                warn!("Received unexpected message from server");
                 return Ok(());
             }
-
-            if let Ok((server_hello, _)) = nacl_blob::receive_hello(&mut socket) {
-                //println!("Received hello from server");
-                // Build the ingestion message
-                let (license_key, node_id, node_name) = {
-                    let lock = load_config().unwrap();
-                    (
-                        lock.long_term_stats.license_key.clone().unwrap_or("".to_string()),
-                        lock.node_id.clone(),
-                        lock.node_name.clone(),
-                    )
-                };
-                if let Ok(key) = Uuid::parse_str(&license_key) {
-                    let mut message = IngestSession {
-                        license_key: key,
-                        node_id,
-                        node_name,
-                        ..Default::default()
-                    };
-                    general::add_general(&mut message, &mut self.general_queue);
-                    circuit_throughput::add_circuit_throughput(&mut message, &mut self.circuit_throughput);
-                    circuit_retransmits::add_circuit_retransmits(&mut message, &mut self.circuit_retransmits);
-                    circuit_rtt::add_circuit_rtt(&mut message, &mut self.circuit_rtt);
-                    circuit_cake_drops::add_circuit_cake_drops(&mut message, &mut self.circuit_cake_drops);
-                    circuit_cake_marks::add_circuit_cake_marks(&mut message, &mut self.circuit_cake_marks);
-                    site_cake_drops::add_site_cake_drops(&mut message, &mut self.site_cake_drops);
-                    site_cake_marks::add_site_cake_marks(&mut message, &mut self.site_cake_marks);
-                    site_retransmits::add_site_retransmits(&mut message, &mut self.site_retransmits);
-                    site_rtt::add_site_rtt(&mut message, &mut self.site_rtt);
-                    site_throughput::add_site_throughput(&mut message, &mut self.site_throughput);
-                    if let Err(e) = nacl_blob::transmit_payload(&keys, &server_hello.public_key, &message, &mut socket) {
-                        println!("Failed to send ingestion message to server. {e:?}");
-                    }
-
-                    // Receive any commands from the remote server
-                    match nacl_blob::receive_payload::<Vec<RemoteCommand>>(&keys, &server_hello.public_key, &mut socket) {
-                        Ok((commands, _size)) => {
-                            remote_commands::enqueue(commands);
-                        }
-                        Err(e) => {
-                            println!("Failed to receive commands from server. {e:?}");
-                        }
-                    }
-                }
-            } else {
-                println!("Failed to receive hello from server");
-            }
-        } else {
-            println!("Failed to connect to ingestion server");
         }
+
+        // Send License
+        let (license_key, node_id, node_name) = {
+            let lock = load_config().unwrap();
+            (
+                lock.long_term_stats.license_key.clone().unwrap_or("".to_string()),
+                lock.node_id.clone(),
+                lock.node_name.clone(),
+            )
+        };
+        let Ok(license_uuid) = Uuid::parse_str(&license_key) else {
+            warn!("Failed to parse license key");
+            return Ok(());
+        };
+        let Ok((_, _, license_to_send)) = (WsMessage::License { license: license_uuid }).to_bytes() else {
+            warn!("Failed to serialize license message");
+            return Ok(());
+        };
+        if let Err(e) = socket.send(tungstenite::Message::Binary(license_to_send)) {
+            warn!("Failed to send license message to server: {}", e);
+            return Ok(());
+        }
+
+        // Wait for CanSubmit
+        let Ok(reply) = socket.read() else {
+            warn!("Failed to receive can submit response from server");
+            return Ok(());
+        };
+        let Ok(reply) = WsMessage::from_bytes(&reply.into_data()) else {
+            warn!("Failed to deserialize can submit response from server");
+            return Ok(());
+        };
+        match reply {
+            WsMessage::CanSubmit => {}
+            _ => {
+                warn!("Received unexpected message from server");
+                return Ok(());
+            }
+        }
+
+        // Submit the actual data
+        let mut message = IngestSession {
+            license_key: license_uuid,
+            node_id: node_id.clone(),
+            node_name,
+            ..Default::default()
+        };
+        general::add_general(&mut message, &mut self.general_queue);
+        circuit_throughput::add_circuit_throughput(&mut message, &mut self.circuit_throughput);
+        circuit_retransmits::add_circuit_retransmits(&mut message, &mut self.circuit_retransmits);
+        circuit_rtt::add_circuit_rtt(&mut message, &mut self.circuit_rtt);
+        circuit_cake_drops::add_circuit_cake_drops(&mut message, &mut self.circuit_cake_drops);
+        circuit_cake_marks::add_circuit_cake_marks(&mut message, &mut self.circuit_cake_marks);
+        site_cake_drops::add_site_cake_drops(&mut message, &mut self.site_cake_drops);
+        site_cake_marks::add_site_cake_marks(&mut message, &mut self.site_cake_marks);
+        site_retransmits::add_site_retransmits(&mut message, &mut self.site_retransmits);
+        site_rtt::add_site_rtt(&mut message, &mut self.site_rtt);
+        site_throughput::add_site_throughput(&mut message, &mut self.site_throughput);
+        let Ok((_, _, data_to_send)) = (WsMessage::DataDump { data: message }).to_bytes() else {
+            warn!("Failed to serialize data message");
+            return Ok(());
+        };
+        if let Err(e) = socket.send(tungstenite::Message::Binary(data_to_send)) {
+            warn!("Failed to send data message to server: {}", e);
+            return Ok(());
+        }
+
+        // Remote Commands
+        let Ok((_, _, request_remote_commands)) = (WsMessage::RequestRemoteCommands { node_id }).to_bytes() else {
+            warn!("Failed to serialize request remote commands message");
+            return Ok(());
+        };
+        if let Err(e) = socket.send(tungstenite::Message::Binary(request_remote_commands)) {
+            warn!("Failed to send request remote commands message to server: {}", e);
+            return Ok(());
+        }
+
+        // Wait for Remote Commands
+        let Ok(reply) = socket.read() else {
+            warn!("Failed to receive remote commands response from server");
+            return Ok(());
+        };
+        let Ok(reply) = WsMessage::from_bytes(&reply.into_data()) else {
+            warn!("Failed to deserialize remote commands response from server");
+            return Ok(());
+        };
+        match reply {
+            WsMessage::RemoteCommands { commands } => {
+                remote_commands::enqueue(commands);
+            }
+            _ => {
+                warn!("Received unexpected message from server");
+                return Ok(());
+            }
+        }
+        drop(socket);
         println!("Finished sending messages to {}", remote_host);
         Ok(())
     }
@@ -186,4 +257,31 @@ impl MessageQueue {
         self.site_cake_marks.clear();
         self.site_rtt.clear();
     }
+}
+
+#[derive(Serialize, Deserialize)]
+enum WsMessage {
+    // Request messages
+    Hello { magic: u32 },
+    License { license: Uuid },
+    DataDump { data: IngestSession },
+    RequestRemoteCommands { node_id: String },
+
+    // Response messages
+    CanSubmit,
+    RemoteCommands { commands: Vec<RemoteCommand> },
+}
+
+impl WsMessage {
+    fn to_bytes(&self) -> anyhow::Result<(usize, usize, Vec<u8>)> {
+        let raw_bytes = serde_cbor::to_vec(self)?;
+        let compressed_bytes = miniz_oxide::deflate::compress_to_vec(&raw_bytes, 10);
+        Ok((raw_bytes.len(), compressed_bytes.len(), compressed_bytes))
+    }
+
+    fn from_bytes(bytes: &[u8]) -> anyhow::Result<Self> {
+        let decompressed_bytes = miniz_oxide::inflate::decompress_to_vec(&bytes)
+            .map_err(|e| anyhow!("Decompression error: {e:?}"))?;
+        Ok(serde_cbor::from_slice(&decompressed_bytes)?)
+    }
 }