tamipes/mc-ingress
1
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity Actions

feat: add abstraction for proxying logic + small fixes

Browse source 
and i dont wanna break this up into multiple commits.
This commit is contained in:
Tamipes 2025-12-10 19:25:59 +01:00
parent 5233570675
commit 2892cc3e7b
5 changed files with 204 additions and 160 deletions
Download patch file Download diff file Expand all files Collapse all files

226
src/kube_cache.rs
View file

@ -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<Deployment>,
services: Api<Service>,
}
impl Cache {
pub fn create() -> Option<Cache> {
impl KubeCache {
/// This initializes the creation of a "kubernetes client"
/// and if it is not possible returns a None.
pub fn create() -> Option<KubeCache> {
let kubeconfig = kube::config::Kubeconfig::read().unwrap();
let client = Client::try_from(kubeconfig).unwrap();
let deployments: Api<Deployment> = Api::default_namespaced(client.clone());
let services: Api<Service> = Api::default_namespaced(client);
return Some(Cache {
return Some(KubeCache {
deployments,
services,
});
}
pub async fn get_dep(&self, name: &str) -> Result<Deployment, kube::Error> {
async fn get_dep(&self, name: &str) -> Result<Deployment, kube::Error> {
self.deployments.get(name).await
}
pub async fn get_srv(&self, name: &str) -> Result<Service, kube::Error> {
async fn get_srv(&self, name: &str) -> Result<Service, kube::Error> {
self.services.get(name).await
}
pub async fn get_deploys(&self) -> ObjectList<Deployment> {
async fn get_deploys(&self) -> ObjectList<Deployment> {
// 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<Service> {
async fn get_srvs(&self) -> ObjectList<Service> {
// 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<Deployment, kube::Error> {
async fn set_dep_scale(&self, name: &str, num: i32) -> Result<Deployment, kube::Error> {
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<Server> for McApi {
#[tracing::instrument(name = "MinecraftAPI::query_server", level = "info", skip(self))]
async fn query_server(&self, addr: String) -> Result<Server, OpaqueError> {
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<Self> {
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<Mutex<kube_cache::Cache>>,
server_addr: &str,
) -> Result<Self, OpaqueError> {
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<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()
}
#[tracing::instrument(level = "info")]
pub async fn get_server_status(&self) -> Result<ServerDeploymentStatus, OpaqueError> {
async fn query_status(&self) -> Result<crate::mc_server::ServerDeploymentStatus, OpaqueError> {
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<TcpStream, OpaqueError> {
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<String> {
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<String> {
Some(self.server_addr.clone())
}
pub async fn set_scale(
&self,
cache: Arc<Mutex<Cache>>,
num: i32,
) -> Result<Deployment, kube::Error> {
}
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<kube::Error> for OpaqueError {
fn from(value: kube::Error) -> Self {
OpaqueError::create(value.to_string().as_str())
}
}

68
src/main.rs
View file

@ -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<T: MinecraftServerHandle>(
mut client_stream: TcpStream,
addr: SocketAddr,
cache: Arc<Mutex<kube_cache::Cache>>,
api: impl MinecraftAPI<T>,
) -> 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<Mutex<Cache>>,
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?;
}
}

66
src/mc_server.rs
View file

@ -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<ServerDeploymentStatus, OpaqueError>;
fn get_internal_address(&self) -> Option<String>;
fn get_addr(&self) -> Option<String>;
async fn query_server_connectable(&self) -> Result<TcpStream, OpaqueError> {
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<T> {
async fn query_server(&self, addr: String) -> Result<T, OpaqueError>;
}
pub enum ServerDeploymentStatus {
Connectable(TcpStream),
Starting,
PodOk,
Offline,
}

2
src/opaque_error.rs
View file

@ -86,7 +86,7 @@ impl From<crate::packets::ParseErrorTrace> 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),
}
}
}