feat: handle login and scaling up deployments

Cargo.lock

@@ -367,6 +367,16 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
 
+[[package]]
+name = "errno"
+version = "0.3.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
+dependencies = [
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "event-listener"
 version = "5.4.1"
@@ -900,6 +910,12 @@ version = "0.2.177"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2874a2af47a2325c2001a6e6fad9b16a53b802102b528163885171cf92b15976"
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039"
+
 [[package]]
 name = "lock_api"
 version = "0.4.14"
@@ -1144,7 +1160,9 @@ dependencies = [
  "serde_derive",
  "serde_json",
  "serde_yaml",
+ "strip-ansi-escapes",
  "tokio",
+ "tokio-splice2",
  "tokio-stream",
  "tracing",
  "tracing-error",
@@ -1238,6 +1256,19 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "rustix"
+version = "1.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd15f8a2c5551a84d56efdc1cd049089e409ac19a3072d5037a17fd70719ff3e"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "rustls"
 version = "0.23.35"
@@ -1503,6 +1534,15 @@ dependencies = [
  "windows-sys 0.60.2",
 ]
 
+[[package]]
+name = "strip-ansi-escapes"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a8f8038e7e7969abb3f1b7c2a811225e9296da208539e0f79c5251d6cac0025"
+dependencies = [
+ "vte",
+]
+
 [[package]]
 name = "strsim"
 version = "0.11.1"
@@ -1619,6 +1659,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-splice2"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3512dd280d6eb6810a34477214bb12640c791f148a14c8c8aecd69b4da6c1f25"
+dependencies = [
+ "crossbeam-utils",
+ "pin-project",
+ "rustix",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-stream"
 version = "0.1.17"
@@ -1811,6 +1863,15 @@ version = "0.9.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
 
+[[package]]
+name = "vte"
+version = "0.14.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "231fdcd7ef3037e8330d8e17e61011a2c244126acc0a982f4040ac3f9f0bc077"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "want"
 version = "0.3.1"

Cargo.toml

@@ -38,3 +38,5 @@ clap = { version = "4.0", default-features = false, features = ["std", "cargo",
 futures = { version = "0.3.17", default-features = false }
 either = "1.6.1"
 nix = { version= "0.30.1", features = [ "zerocopy"] }
+tokio-splice2 = "0.3.2"
+strip-ansi-escapes = "0.2.1"

src/kube_cache.rs

@@ -2,10 +2,11 @@ use std::{fmt, sync::Arc};
 
 use k8s_openapi::api::{apps::v1::Deployment, core::v1::Service};
 use kube::{
-    api::{ListParams, ObjectList},
+    api::{ListParams, ObjectList, Patch, PatchParams},
     runtime::reflector::Lookup,
     Api, Client, ResourceExt,
 };
+use serde_json::json;
 use tokio::{net::TcpStream, sync::Mutex};
 
 use crate::{
@@ -61,19 +62,17 @@ impl Cache {
         Some(result.name()?.to_string())
     }
 
-    // pub fn set_dep(
+    pub async fn set_dep_scale(&self, name: &str, num: i32) -> Result<Deployment, kube::Error> {
-    //     &self,
+        let patch = Patch::Merge(json!({"spec":{"replicas": num}}));
-    //     name: &str,
+        let pp = PatchParams::default();
-    //     pp: &PatchParams,
+        self.deployments.patch(name, &pp, &patch).await
-    // ) -> impl std::future::Future<Output = Result<Deployment, kube::Error>> {
+    }
-    //     let patch = Patch;
-    //     self.deployments.patch(name, pp, patch)
-    // }
 }
 
 pub struct KubeServer {
     dep: Deployment,
     srv: Service,
+    server_addr: String,
 }
 impl fmt::Debug for KubeServer {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -96,6 +95,7 @@ impl fmt::Debug for KubeServer {
                     .name
                     .unwrap_or("#error#".to_string()),
             )
+            .field("server_addr", &self.server_addr)
             .finish()
     }
 }
@@ -136,19 +136,23 @@ impl KubeServer {
             )
         })?;
         drop(cache_guard);
-        tracing::info!("found kubernetes deployment & service");
+        tracing::debug!("found kubernetes deployment & service");
 
         return Ok(Self {
             dep: deployment,
             srv: service,
+            server_addr: server_addr.to_string(),
         });
     }
-    fn get_port(&self) -> Option<i32> {
+    pub fn get_port(&self) -> Option<i32> {
         let a = self.srv.clone().spec.unwrap().ports.unwrap();
         let port = a.iter().find(|x| x.name.clone().unwrap() == "mc-router")?;
         port.node_port
     }
     #[tracing::instrument(level = "info")]
+    pub fn get_server_addr(&self) -> String {
+        self.server_addr.clone()
+    }
     pub async fn get_server_status(&self) -> Result<ServerDeploymentStatus, OpaqueError> {
         let mut status = match self.dep.clone().status {
             Some(x) => x,
@@ -238,9 +242,26 @@ impl KubeServer {
                     e
                 )
             })?;
-        tracing::info!("data exchanged while proxing: {:?}", data_amount);
+        tracing::debug!("data exchanged while proxing: {:?}", data_amount);
         Ok(())
     }
+    pub async fn set_scale(
+        &self,
+        cache: Arc<Mutex<Cache>>,
+        num: i32,
+    ) -> Result<Deployment, kube::Error> {
+        let name = self
+            .srv
+            .metadata
+            .clone()
+            .name
+            .unwrap_or("#error#".to_string());
+        let res = cache.lock().await.set_dep_scale(&name, num).await;
+        if res.is_ok() {
+            tracing::info!("scaled replicas of {} to {num}", self.server_addr);
+        }
+        return res;
+    }
 }
 
 fn filter_label_value<R>(dep: &&R, str: &str) -> bool

src/main.rs

@@ -3,11 +3,13 @@
 //! with labels and namespace selectors supported.
 use std::{net::SocketAddr, sync::Arc};
 
+use futures::TryFutureExt;
+use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::Mutex;
 use tracing_subscriber::{prelude::*, EnvFilter};
 
-use crate::kube_cache::{KubeServer, ServerDeploymentStatus};
+use crate::kube_cache::{Cache, KubeServer, ServerDeploymentStatus};
 use crate::opaque_error::OpaqueError;
 use crate::packets::clientbound::status::StatusStructNew;
 use crate::packets::serverbound::handshake::Handshake;
@@ -55,7 +57,7 @@ async fn main() {
             if let Err(e) = process_connection(socket, addr, acc).await {
                 tracing::error!(
                     message = format!("Client disconnected"),
-                    addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
+                    // addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()),
                     trace = format!("{}", e.get_span_trace()),
                     err = format!("{}", e.context)
                 );
@@ -85,7 +87,7 @@ async fn process_connection(
 
     // --- Handshake ---
     let handshake;
-    let server_state;
+    let next_server_state;
     if client_packet.id.get_int() != 0 {
         return Err(OpaqueError::create(
             "Client HANDSHAKE -> bad packet; Disconnecting...",
@@ -95,38 +97,41 @@ async fn process_connection(
         .await
         .ok_or_else(|| "Handshake request from client failed to parse".to_string())?;
 
-    server_state = handshake.get_next_state();
+    next_server_state = handshake.get_next_state();
 
-    match server_state {
+    let kube_server = KubeServer::create(cache.clone(), &handshake.get_server_address()).await?;
+    tracing::debug!(
+        "kube server status: {:?}",
+        kube_server.get_server_status().await?
+    );
+
+    match next_server_state {
         packets::ProtocolState::Status => {
-            handle_status(
+            handle_status(&mut client_stream, &handshake, kube_server).await?;
-                &mut client_stream,
-                cache,
-                handshake.get_server_address(),
-                &handshake,
-            )
-            .await?;
         }
-        packets::ProtocolState::Login => todo!(),
+        packets::ProtocolState::Login => {
-        packets::ProtocolState::Transfer => todo!(),
+            handle_login(&mut client_stream, &handshake, kube_server, cache.clone()).await?
-        _ => todo!(),
+        }
+        packets::ProtocolState::Transfer => {
+            return Err(OpaqueError::create("Transfer; Not yet implemented!"))
+        }
+        _ => unreachable!(),
     };
     Ok(())
 }
 
-#[tracing::instrument(level = "info", skip(client_stream, cache, handshake))]
+#[tracing::instrument(level = "info", fields(server_addr = kube_server.get_server_addr()),skip(client_stream, handshake, kube_server))]
 async fn handle_status(
     client_stream: &mut TcpStream,
-    cache: Arc<Mutex<kube_cache::Cache>>,
-    server_addr: String,
     handshake: &Handshake,
+    kube_server: KubeServer,
 ) -> Result<(), OpaqueError> {
     tracing::debug!(handshake = ?handshake);
     let client_packet = Packet::parse(client_stream)
         .await
         .ok_or_else(|| "Could not parse client_packet".to_string())?;
     match client_packet.id.get_int() {
-        0 => tracing::info!("Client STATUS: {:#x} Status Request", 0),
+        0 => tracing::info!("status request"),
         _ => {
             return Err(OpaqueError::create(&format!(
                 "Client STATUS: {:#x} Unknown Id -> Shutdown",
@@ -135,15 +140,10 @@ async fn handle_status(
         }
     };
 
-    let kube_server = KubeServer::create(cache, &server_addr).await?;
-
-    let status: ServerDeploymentStatus = kube_server.get_server_status().await?;
-    tracing::info!("kube server status: {:?}", status);
-
     let commit_hash: &'static str = env!("COMMIT_HASH");
     let mut status_struct = StatusStructNew::create();
     status_struct.version.protocol = handshake.protocol_version.get_int();
-    match status {
+    match kube_server.get_server_status().await? {
         ServerDeploymentStatus::Connectable => {
             return kube_server
                 .proxy_status(handshake, &client_packet, client_stream)
@@ -170,3 +170,87 @@ async fn handle_status(
 
     Ok(())
 }
+
+#[tracing::instrument(level = "info", fields(server_addr = kube_server.get_server_addr()),skip(client_stream, handshake, kube_server,cache))]
+async fn handle_login(
+    client_stream: &mut TcpStream,
+    handshake: &Handshake,
+    kube_server: KubeServer,
+    cache: Arc<Mutex<Cache>>,
+) -> Result<(), OpaqueError> {
+    // let client_packet = Packet::parse(client_stream).await.unwrap();
+    tracing::info!("login request");
+    match kube_server.get_server_status().await? {
+        ServerDeploymentStatus::Connectable => {
+            // referenced from:
+            // https://github.com/hanyu-dev/tokio-splice2/blob/fc47199fffde8946b0acf867d1fa0b2222267a34/examples/proxy.rs
+            let io_sl2sr = tokio_splice2::context::SpliceIoCtx::prepare()
+                .map_err(|e| format!("tokio_splice2::context::SpliceIoCtx err={}", e.to_string()))?
+                .into_io();
+
+            let io_sr2sl = tokio_splice2::context::SpliceIoCtx::prepare()
+                .map_err(|e| format!("tokio_splice2::context::SpliceIoCtx err={}", e.to_string()))?
+                .into_io();
+
+            let port = kube_server
+                .get_port()
+                .ok_or_else(|| "failed to get port from service")?;
+            let mut server_stream = TcpStream::connect(format!("localhost:{}", port))
+                .await
+                .map_err(|_| "Failed to connect to minecraft server")?;
+
+            handshake
+                .send_packet(&mut server_stream)
+                .await
+                .map_err(|_| "Failed to forward handshake packet to minecraft server")?;
+
+            tracing::info!("proxying with splice");
+            let traffic = tokio_splice2::io::SpliceBidiIo { io_sl2sr, io_sr2sl }
+                .execute(client_stream, &mut server_stream)
+                .await;
+            tracing::debug!("data exchanged: tx: {} rx: {}", traffic.tx, traffic.rx);
+        }
+        ServerDeploymentStatus::PodOk | ServerDeploymentStatus::Starting => {
+            let _client_packet = Packet::parse(client_stream).await;
+            if _client_packet.is_none() {
+                return Err(OpaqueError::create(
+                    "Client LOGIN START -> bad packet; Disconnecting...",
+                ));
+            }
+
+            let disconnect_packet =
+                packets::clientbound::login::Disconnect::set_reason("Starting...§d<3§r".to_owned())
+                    .await
+                    .ok_or_else(|| "failed to *create* disconnect packet")?;
+            disconnect_packet
+                .send_packet(client_stream)
+                .await
+                .map_err(|_| "failed to *send* disconnect packet")?;
+            client_stream.flush().await.map_err(|e| e.to_string())?;
+        }
+        ServerDeploymentStatus::Offline => {
+            let _client_packet = Packet::parse(client_stream).await;
+            if _client_packet.is_none() {
+                return Err(OpaqueError::create(
+                    "Client LOGIN START -> bad packet; Disconnecting...",
+                ));
+            }
+
+            kube_server
+                .set_scale(cache, 1)
+                .map_err(|e| format!("Failed to set depoloyment scale: err = {:?}", e))
+                .await?;
+            let disconnect_packet = packets::clientbound::login::Disconnect::set_reason(
+                "Okayy_starting_it...§d<3§r".to_owned(),
+            )
+            .await
+            .ok_or_else(|| "failed to *create* disconnect packet")?;
+            disconnect_packet
+                .send_packet(client_stream)
+                .await
+                .map_err(|_| "failed to *send* disconnect packet")?;
+            client_stream.flush().await.map_err(|e| e.to_string())?;
+        }
+    }
+    Ok(())
+}

src/opaque_error.rs

@@ -38,19 +38,29 @@ impl OpaqueError {
         }
     }
     pub fn get_span_trace(&self) -> String {
-        let mut vec = Vec::new();
+        let mut vec: Vec<(&str, String)> = Vec::new();
 
         self.span_trace.with_spans(|metadata, _fields| {
-            vec.push(metadata.name());
+            vec.push((metadata.name(), _fields.into()));
             true
         });
-        vec.iter().rfold(String::new(), |mut acc, x| {
+        let str = vec.iter().rfold(String::new(), |mut acc, x| {
-            if acc.len() != 0 {
+            let first = acc.len() != 0;
-                acc.push_str("::");
+            if first {
+                acc.push_str(":");
             }
-            acc.push_str(x);
+
+            acc.push_str(x.0);
+            acc.push_str("{");
+            acc.push_str(&x.1);
+            acc.push_str("}");
+
             acc
-        })
+        });
+        match String::from_utf8(strip_ansi_escapes::strip(str.clone())) {
+            Ok(x) => x,
+            Err(_) => str,
+        }
     }
 }