diff --git a/Cargo.lock b/Cargo.lock index 26c44ab..38d904d 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 81379d3..1d8e1e7 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/kube_cache.rs b/src/kube_cache.rs index 1654d4c..e26c147 100644 --- a/src/kube_cache.rs +++ b/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( - // &self, - // name: &str, - // pp: &PatchParams, - // ) -> impl std::future::Future> { - // let patch = Patch; - // self.deployments.patch(name, pp, patch) - // } + pub async fn set_dep_scale(&self, name: &str, num: i32) -> Result { + let patch = Patch::Merge(json!({"spec":{"replicas": num}})); + let pp = PatchParams::default(); + self.deployments.patch(name, &pp, &patch).await + } } 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 { + pub fn get_port(&self) -> Option { 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 { 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>, + num: i32, + ) -> Result { + 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(dep: &&R, str: &str) -> bool diff --git a/src/main.rs b/src/main.rs index d2bc56d..e863f3f 100644 --- a/src/main.rs +++ b/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( - &mut client_stream, - cache, - handshake.get_server_address(), - &handshake, - ) - .await?; + handle_status(&mut client_stream, &handshake, kube_server).await?; } - packets::ProtocolState::Login => todo!(), - packets::ProtocolState::Transfer => todo!(), - _ => todo!(), + packets::ProtocolState::Login => { + handle_login(&mut client_stream, &handshake, kube_server, cache.clone()).await? + } + 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>, - 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>, +) -> 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(()) +} diff --git a/src/opaque_error.rs b/src/opaque_error.rs index 7d602b3..1876bdd 100644 --- a/src/opaque_error.rs +++ b/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| { - if acc.len() != 0 { - acc.push_str("::"); + let str = vec.iter().rfold(String::new(), |mut acc, x| { + let first = acc.len() != 0; + 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, + } } }