diff --git a/README.md b/README.md index f73b2c2..17c2f00 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,2 @@ # Env variables: -- BIND_PORT(default: 25565): the port the server should bind to +- BIND_ADDR(default: 0.0.0.0:25565): the address the server should bind to diff --git a/src/kube_cache.rs b/src/kube_cache.rs index 8b7f4a2..d427e95 100644 --- a/src/kube_cache.rs +++ b/src/kube_cache.rs @@ -1,4 +1,4 @@ -use std::{fmt, sync::Arc}; +use std::fmt; use k8s_openapi::api::{apps::v1::Deployment, core::v1::Service}; use kube::{ @@ -7,44 +7,49 @@ use kube::{ Api, Client, ResourceExt, }; use serde_json::json; -use tokio::{net::TcpStream, sync::Mutex}; use crate::{ - kube_cache, - packets::{serverbound::handshake::Handshake, Packet, SendPacket}, + mc_server::{MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus}, OpaqueError, }; -#[derive(Debug)] -pub struct Cache { +/// This is the layer who is respinsible for caching requests. +/// +/// TODO: +/// It should be also clone-able freely, because it deals with +/// the underlying async data access. +#[derive(Debug, Clone)] +pub struct KubeCache { deployments: Api, services: Api, } -impl Cache { - pub fn create() -> Option { +impl KubeCache { + /// This initializes the creation of a "kubernetes client" + /// and if it is not possible returns a None. + pub fn create() -> Option { let kubeconfig = kube::config::Kubeconfig::read().unwrap(); let client = Client::try_from(kubeconfig).unwrap(); let deployments: Api = Api::default_namespaced(client.clone()); let services: Api = Api::default_namespaced(client); - return Some(Cache { + return Some(KubeCache { deployments, services, }); } - pub async fn get_dep(&self, name: &str) -> Result { + async fn get_dep(&self, name: &str) -> Result { self.deployments.get(name).await } - pub async fn get_srv(&self, name: &str) -> Result { + async fn get_srv(&self, name: &str) -> Result { self.services.get(name).await } - pub async fn get_deploys(&self) -> ObjectList { + async fn get_deploys(&self) -> ObjectList { // let lp: ListParams = ListParams::default(); let lp: ListParams = ListParams::default().labels("tami.moe/minecraft"); self.deployments.list(&lp).await.unwrap() } - pub async fn get_srvs(&self) -> ObjectList { + async fn get_srvs(&self) -> ObjectList { // let lp: ListParams = ListParams::default(); let lp: ListParams = ListParams::default().labels("tami.moe/minecraft"); self.services.list(&lp).await.unwrap() @@ -62,19 +67,76 @@ impl Cache { Some(result.name()?.to_string()) } - pub async fn set_dep_scale(&self, name: &str, num: i32) -> Result { + 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 { +#[derive(Clone)] +pub struct McApi { + cache: KubeCache, +} + +impl MinecraftAPI for McApi { + #[tracing::instrument(name = "MinecraftAPI::query_server", level = "info", skip(self))] + async fn query_server(&self, addr: String) -> Result { + let dep_name = match self.cache.query_dep_addr(&addr).await { + Some(x) => x, + None => { + return Err(OpaqueError::create(&format!( + "Failed to find deployment name by addr" + ))) + } + }; + let srv_name = match self.cache.query_srv_addr(&addr).await { + Some(x) => x, + None => { + return Err(OpaqueError::create(&format!( + "Failed to find service name by addr" + ))) + } + }; + + let deployment = self.cache.get_dep(&dep_name).await.map_err(|x| { + format!( + "Failed to query cache for deployment with dep_name err:{}", + x.to_string() + ) + })?; + let service = self.cache.get_srv(&srv_name).await.map_err(|x| { + format!( + "Failed to query cache for service with dep_name err:{}", + x.to_string() + ) + })?; + tracing::debug!("found kubernetes deployment & service"); + + return Ok(Server { + dep: deployment, + srv: service, + server_addr: addr.to_string(), + cache: self.cache.clone(), + }); + } +} + +impl McApi { + pub fn create() -> Option { + Some(Self { + cache: KubeCache::create()?, + }) + } +} + +pub struct Server { dep: Deployment, srv: Service, server_addr: String, + cache: KubeCache, } -impl fmt::Debug for KubeServer { +impl fmt::Debug for Server { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("KubeServer") .field( @@ -99,62 +161,21 @@ impl fmt::Debug for KubeServer { .finish() } } -impl KubeServer { - #[tracing::instrument(name = "KubeServer::create", level = "info", skip(cache))] - pub async fn create( - cache: Arc>, - server_addr: &str, - ) -> Result { - let cache_guard = cache.lock().await; - let dep_name = match cache_guard.query_dep_addr(server_addr).await { - Some(x) => x, - None => { - return Err(OpaqueError::create(&format!( - "Failed to find deployment name by addr" - ))) - } - }; - let srv_name = match cache_guard.query_srv_addr(server_addr).await { - Some(x) => x, - None => { - return Err(OpaqueError::create(&format!( - "Failed to find service name by addr" - ))) - } - }; +impl MinecraftServerHandle for Server { + async fn start(&self) -> Result<(), OpaqueError> { + self.set_scale(1).await.map_err(|e| { + OpaqueError::create(&format!("failed to set deployment scale: err = {:?}", e)) + }) + } - let deployment = cache_guard.get_dep(&dep_name).await.map_err(|x| { - format!( - "Failed to query cache for deployment with dep_name err:{}", - x.to_string() - ) - })?; - let service = cache_guard.get_srv(&srv_name).await.map_err(|x| { - format!( - "Failed to query cache for service with dep_name err:{}", - x.to_string() - ) - })?; - drop(cache_guard); - tracing::debug!("found kubernetes deployment & service"); + async fn stop(&self) -> Result<(), OpaqueError> { + self.set_scale(1).await.map_err(|e| { + OpaqueError::create(&format!("failed to set deployment scale: err = {:?}", e)) + }) + } - return Ok(Self { - dep: deployment, - srv: service, - server_addr: server_addr.to_string(), - }); - } - 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() - } - #[tracing::instrument(level = "info")] - pub async fn get_server_status(&self) -> Result { + async fn query_status(&self) -> Result { let mut status = match self.dep.clone().status { Some(x) => x, None => { @@ -198,63 +219,31 @@ impl KubeServer { return Ok(ServerDeploymentStatus::Offline); } } - async fn query_server_connectable(&self) -> Result { - let port = self - .get_port() - .ok_or_else(|| "failed to get port from service")?; - let server_stream = TcpStream::connect(format!("localhost:{port}")) - .await - .map_err(|_| "failed to connect to minecraft server")?; - tracing::trace!( - "successfully connected to backend server; (connectibility check) {:?}", - server_stream.peer_addr() - ); - Ok(server_stream) + fn get_internal_address(&self) -> Option { + let a = self.srv.clone().spec.unwrap().ports.unwrap(); + let port = a.iter().find(|x| x.name.clone().unwrap() == "mc-router")?; + Some(format!("localhost:{}", port.node_port?)) } - pub async fn proxy_status( - &self, - handshake: &Handshake, - status_request: &Packet, - client_stream: &mut TcpStream, - server_stream: &mut TcpStream, - ) -> Result<(), OpaqueError> { - handshake - .send_packet(server_stream) - .await - .map_err(|_| "failed to forward handshake packet to minecraft server")?; - status_request - .send_packet(server_stream) - .await - .map_err(|_| "failed to forward status request packet to minecraft server")?; - let data_amount = tokio::io::copy_bidirectional(client_stream, server_stream) - .await - .map_err(|e| { - format!( - "error during bidirectional copy between server and client; err={:?}", - e - ) - })?; - tracing::trace!("data exchanged while proxying status: {:?}", data_amount); - Ok(()) + fn get_addr(&self) -> Option { + Some(self.server_addr.clone()) } - pub async fn set_scale( - &self, - cache: Arc>, - num: i32, - ) -> Result { +} + +impl Server { + async fn set_scale(&self, num: i32) -> Result<(), 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; + let res = self.cache.set_dep_scale(&name, num).await; if res.is_ok() { tracing::info!("scaled replicas of {} to {num}", self.server_addr); } - return res; + Ok(()) } } @@ -265,12 +254,6 @@ where dep.labels().values().filter(|x| x.as_str() == str).count() > 0 } -pub enum ServerDeploymentStatus { - Connectable(TcpStream), - Starting, - PodOk, - Offline, -} impl fmt::Debug for ServerDeploymentStatus { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { @@ -281,3 +264,8 @@ impl fmt::Debug for ServerDeploymentStatus { } } } +impl From for OpaqueError { + fn from(value: kube::Error) -> Self { + OpaqueError::create(value.to_string().as_str()) + } +} diff --git a/src/main.rs b/src/main.rs index baadcc0..2131103 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,12 +1,10 @@ use std::env; -use std::{net::SocketAddr, sync::Arc}; +use std::net::SocketAddr; -use futures::TryFutureExt; use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::Mutex; use tracing_subscriber::{prelude::*, EnvFilter}; -use crate::kube_cache::{Cache, KubeServer, ServerDeploymentStatus}; +use crate::mc_server::{MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus}; use crate::opaque_error::OpaqueError; use crate::packets::clientbound::status::StatusStructNew; use crate::packets::serverbound::handshake::Handshake; @@ -30,31 +28,31 @@ async fn main() { .with(filter_layer) .with(tracing_error::ErrorLayer::default()) .init(); + tracing::info!("mc-ingress"); let revision: &'static str = env!("COMMIT_HASH"); tracing::info!(revision); - let cache = kube_cache::Cache::create().unwrap(); - let arc_cache = Arc::new(Mutex::new(cache)); + let api = kube_cache::McApi::create().unwrap(); tracing::info!("initialized kube api"); - let port = match env::var("BIND_PORT") { + let addr = match env::var("BIND_ADDR") { Ok(x) => x, - Err(_) => "25565".to_string(), + Err(_) => "0.0.0.0:25565".to_string(), }; - let listener = TcpListener::bind(format!("0.0.0.0:{port}")).await.unwrap(); - tracing::info!(port, "started tcp server"); + let listener = TcpListener::bind(addr.clone()).await.unwrap(); + tracing::info!(addr, "started tcp server"); loop { let (socket, addr) = listener.accept().await.unwrap(); - let acc = arc_cache.clone(); + let api = api.clone(); tokio::spawn(async move { tracing::debug!( addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()), "Client connected" ); - if let Err(e) = process_connection(socket, addr, acc).await { + if let Err(e) = process_connection(socket, addr, api).await { tracing::error!( // addr = format!("{}:{}", addr.ip().to_string(), addr.port().to_string()), trace = format!("{}", e.get_span_trace()), @@ -71,21 +69,22 @@ async fn main() { } } -#[tracing::instrument(level = "info", skip(cache, client_stream))] -async fn process_connection( +#[tracing::instrument(level = "info", skip(api, client_stream))] +async fn process_connection( mut client_stream: TcpStream, addr: SocketAddr, - cache: Arc>, + api: impl MinecraftAPI, ) -> Result<(), OpaqueError> { let client_packet = Packet::parse(&mut client_stream).await?; // --- Handshake --- let handshake; let next_server_state; - if client_packet.id.get_int() != 0 { - return Err(OpaqueError::create( - "Client HANDSHAKE -> bad packet; Disconnecting...", - )); + let packet_id = client_packet.id.get_int(); + if packet_id != 0 { + return Err(OpaqueError::create(&format!( + "Client HANDSHAKE -> bad packet; id={packet_id} Disconnecting..." + ))); } handshake = packets::serverbound::handshake::Handshake::parse(client_packet) .await @@ -93,18 +92,15 @@ async fn process_connection( next_server_state = handshake.get_next_state(); - let kube_server = KubeServer::create(cache.clone(), &handshake.get_server_address()).await?; - tracing::debug!( - "kube server status: {:?}", - kube_server.get_server_status().await? - ); + let server = api.query_server(handshake.get_server_address()).await?; + tracing::debug!("kube server status: {:?}", server.query_status().await?); match next_server_state { packets::ProtocolState::Status => { - handle_status(&mut client_stream, &handshake, kube_server).await?; + handle_status(&mut client_stream, &handshake, server).await?; } packets::ProtocolState::Login => { - handle_login(&mut client_stream, &handshake, kube_server, cache.clone()).await? + handle_login(&mut client_stream, &handshake, server).await? } packets::ProtocolState::Transfer => { return Err(OpaqueError::create( @@ -117,11 +113,11 @@ async fn process_connection( Ok(()) } -#[tracing::instrument(level = "info", fields(server_addr = kube_server.get_server_addr()),skip(client_stream, handshake, kube_server))] +#[tracing::instrument(level = "info", fields(server_addr = server.get_addr()),skip(client_stream, handshake, server))] async fn handle_status( client_stream: &mut TcpStream, handshake: &Handshake, - kube_server: KubeServer, + server: impl MinecraftServerHandle, ) -> Result<(), OpaqueError> { tracing::debug!(handshake = ?handshake); let client_packet = Packet::parse(client_stream).await?; @@ -135,11 +131,11 @@ async fn handle_status( let commit_hash: &'static str = env!("COMMIT_HASH"); let mut status_struct = StatusStructNew::create(); status_struct.version.protocol = handshake.protocol_version.get_int(); - let status = kube_server.get_server_status().await?; + let status = server.query_status().await?; tracing::info!(status = ?status, "status request"); match status { ServerDeploymentStatus::Connectable(mut server_stream) => { - return kube_server + return server .proxy_status(handshake, &client_packet, client_stream, &mut server_stream) .await } @@ -165,14 +161,13 @@ async fn handle_status( Ok(()) } -#[tracing::instrument(level = "info", fields(server_addr = kube_server.get_server_addr()),skip(client_stream, handshake, kube_server,cache))] +#[tracing::instrument(level = "info", fields(server_addr = server.get_addr()),skip(client_stream, handshake, server))] async fn handle_login( client_stream: &mut TcpStream, handshake: &Handshake, - kube_server: KubeServer, - cache: Arc>, + server: impl MinecraftServerHandle, ) -> Result<(), OpaqueError> { - match kube_server.get_server_status().await? { + match server.query_status().await? { ServerDeploymentStatus::Connectable(mut server_stream) => { // referenced from: // https://github.com/hanyu-dev/tokio-splice2/blob/fc47199fffde8946b0acf867d1fa0b2222267a34/examples/proxy.rs @@ -206,10 +201,7 @@ async fn handle_login( mc_server::send_disconnect(client_stream, "Starting...§d<3§r").await?; } ServerDeploymentStatus::Offline => { - kube_server - .set_scale(cache, 1) - .map_err(|e| format!("Failed to set depoloyment scale: err = {:?}", e)) - .await?; + server.start().await?; mc_server::send_disconnect(client_stream, "Okayy_starting_it...§d<3§r").await?; } } diff --git a/src/mc_server.rs b/src/mc_server.rs index d5dd5f0..138e123 100644 --- a/src/mc_server.rs +++ b/src/mc_server.rs @@ -1,7 +1,7 @@ use tokio::{io::AsyncWriteExt, net::TcpStream}; use crate::{ - packets::{Packet, SendPacket}, + packets::{serverbound::handshake::Handshake, Packet, SendPacket}, OpaqueError, }; @@ -43,3 +43,67 @@ pub async fn send_disconnect( client_stream.flush().await.map_err(|e| e.to_string())?; Ok(()) } + +pub trait MinecraftServerHandle { + async fn start(&self) -> Result<(), OpaqueError>; + async fn stop(&self) -> Result<(), OpaqueError>; + async fn query_status(&self) -> Result; + fn get_internal_address(&self) -> Option; + fn get_addr(&self) -> Option; + + async fn query_server_connectable(&self) -> Result { + let port = self + .get_internal_address() + .ok_or_else(|| "failed to get internal address from server")?; + let server_stream = TcpStream::connect(port) + .await + .map_err(|_| "failed to connect to minecraft server")?; + + tracing::trace!( + "successfully connected to backend server; (connectibility check) {:?}", + server_stream.peer_addr() + ); + Ok(server_stream) + } + + /// Takes the already received packets, sends them to the server, + /// and proceeds to proxy the rest of the connection transparently. + async fn proxy_status( + &self, + handshake: &Handshake, + status_request: &Packet, + client_stream: &mut TcpStream, + server_stream: &mut TcpStream, + ) -> Result<(), OpaqueError> { + handshake + .send_packet(server_stream) + .await + .map_err(|_| "failed to forward handshake packet to minecraft server")?; + status_request + .send_packet(server_stream) + .await + .map_err(|_| "failed to forward status request packet to minecraft server")?; + + let data_amount = tokio::io::copy_bidirectional(client_stream, server_stream) + .await + .map_err(|e| { + format!( + "error during bidirectional copy between server and client; err={:?}", + e + ) + })?; + tracing::trace!("data exchanged while proxying status: {:?}", data_amount); + Ok(()) + } +} + +pub trait MinecraftAPI { + async fn query_server(&self, addr: String) -> Result; +} + +pub enum ServerDeploymentStatus { + Connectable(TcpStream), + Starting, + PodOk, + Offline, +} diff --git a/src/opaque_error.rs b/src/opaque_error.rs index f1aa578..ad70182 100644 --- a/src/opaque_error.rs +++ b/src/opaque_error.rs @@ -86,7 +86,7 @@ impl From for OpaqueError { fn from(value: crate::packets::ParseErrorTrace) -> Self { Self { span_trace: value.context, - context: format!("packet parse error: {:?}", value.inner), + context: format!("{:?}", value.inner), } } }