diff --git a/Cargo.lock b/Cargo.lock index 599e7b8..2644623 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -745,6 +745,15 @@ dependencies = [ "hashbrown 0.16.1", ] +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.15" @@ -970,6 +979,7 @@ dependencies = [ "either", "evalexpr", "futures", + "itertools", "k8s-openapi", "kube", "nix", diff --git a/Cargo.toml b/Cargo.toml index 85040c9..63cec66 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,3 +43,4 @@ strip-ansi-escapes = "0.2.1" evalexpr = { version = "13.1.0", features = ["regex"] } async-trait = "0.1.89" tokio-util = { version = "0.7.18", features = ["rt"] } +itertools = "0.14.0" diff --git a/src/kube_cache.rs b/src/kube_cache.rs index 199109d..c929128 100644 --- a/src/kube_cache.rs +++ b/src/kube_cache.rs @@ -2,6 +2,7 @@ use std::{collections::HashMap, fmt, sync::Arc, time::Duration}; use async_trait::async_trait; use futures::StreamExt; +use itertools::Itertools; use k8s_openapi::api::{apps::v1::Deployment, core::v1::Service}; use kube::{ api::{ListParams, ObjectList, Patch, PatchParams}, @@ -17,12 +18,13 @@ use crate::{ }; const MAIN_LABEL: &str = "tami.moe/minecraft"; const PORT_LABEL: &str = "tami.moe/minecraft-port"; +const URL_ANNOTATION: &str = "tami.moe/minecraft-url"; /// 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. + +// A heads-up: +// It should be also clone-able freely, because it deals with +// the underlying async data access. #[derive(Debug, Clone)] pub struct KubeCache { // When the store gets copied, it points to the same backing store @@ -39,6 +41,11 @@ impl KubeCache { /// /// It also returns a JoinHandle which is tied to the api watcher /// and if it returns that means that the kubecache is not responsive now. + /// + /// # Example: + /// ```no_run + /// let (kube_cache, api_task) = kube_cache::KubeCache::create().await.unwrap(); + /// ``` pub async fn create() -> Option<(KubeCache, JoinHandle<()>)> { let in_cluster = match std::env::var("KUBERNETES_SERVICE_HOST") { Ok(_) => true, @@ -59,7 +66,7 @@ impl KubeCache { .for_each(|_o| std::future::ready(())); let _res = infinite_watch.await; tracing::error!( - "Deployments watcher ended. This should not happen. (Program should exit now)" + "deployments watcher ended; this should not happen; (program should exit now)" ); }); @@ -76,6 +83,9 @@ impl KubeCache { )); } fn get_dep(&self, name: &str) -> Option> { + // dep.name_any() has the caveat of returning "" if it can't query for + // a name, resulting in unexpected behaviour if that happens and also + // a query for "" comes as well. self.dep_cache.find(|x| x.name_any() == name) } async fn get_srv(&self, name: &str) -> Result { @@ -297,13 +307,16 @@ impl MinecraftServerHandle for Server { fn get_internal_address(&self) -> &str { self.internal_address.as_str() } - - fn get_addr(&self) -> String { - self.server_addr.clone() + fn get_internal_id(&self) -> &str { + self.internal_id.as_str() } - fn get_port(&self) -> String { - self.server_port.clone() + fn get_join_addr(&self) -> &str { + self.server_addr.as_ref() + } + + fn get_join_port(&self) -> &str { + self.server_port.as_ref() } fn get_motd(&self) -> Option { @@ -347,15 +360,28 @@ fn filter_label_value(res: &R, addr: &str, port: &str) -> bool where R: ResourceExt, { + let labels_iter = res.labels().iter(); + let annotations_iter = res.annotations().iter(); + let iters = labels_iter.merge(annotations_iter); + let mut found_port = false; - res.labels() - .iter() + iters .filter(|(key, value)| match key.as_str() { MAIN_LABEL => value.as_str() == addr, PORT_LABEL => { found_port = true; value.as_str() == port } + URL_ANNOTATION => value + .lines() + .find(|str| { + if str.find(':').is_some() { + str == &format!("{addr}:{port}").as_str() + } else { + str == &addr + } + }) + .is_some(), _ => false, }) .count() diff --git a/src/mc_server.rs b/src/mc_server.rs index 3794acf..e403b71 100644 --- a/src/mc_server.rs +++ b/src/mc_server.rs @@ -70,8 +70,9 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { async fn stop(&self) -> Result<(), OpaqueError>; async fn query_status(&self) -> Result; fn get_internal_address(&self) -> &str; - fn get_addr(&self) -> String; - fn get_port(&self) -> String; + fn get_internal_id(&self) -> &str; + fn get_join_addr(&self) -> &str; + fn get_join_port(&self) -> &str; fn get_motd(&self) -> Option; async fn query_server_connectable(&self) -> Result { @@ -122,7 +123,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone { ServerDeploymentStatus::Connectable(mut tcp_stream) => { let handshake = crate::packets::serverbound::handshake::Handshake::create( crate::types::VarInt::from(746).ok_or("could not create VarInt WTF?")?, - crate::types::VarString::from(self.get_addr()), + crate::types::VarString::from(self.get_join_addr().to_string()), crate::types::UShort::from(1234), crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?, ) @@ -162,7 +163,6 @@ pub trait MinecraftAPI: Send + Sync + 'static { async fn query_server(&self, addr: &str, port: &str) -> Result; fn get_map(&self) -> Arc>>>; - // TODO: move the implementation to here, but /// This should be callable even if there is already a watcher, /// and it should handle the collision itself while returning OK(). async fn start_watch( @@ -171,18 +171,18 @@ pub trait MinecraftAPI: Send + Sync + 'static { frequency: std::time::Duration, ) -> Result<(), OpaqueError> where - Self: Send + Sync + 'static + Clone, + Self: Clone, { - let inter_addr = server.get_internal_address().to_string(); + let internal_id = server.get_internal_id().to_string(); - if let Some(handle) = self.get_map().lock().await.get(&inter_addr) { + if let Some(handle) = self.get_map().lock().await.get(&internal_id) { if !handle.is_finished() { return Ok(()); } } - let span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", inter_addr, join_addr = server.get_addr(), join_port = server.get_port()); + let span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", internal_id, join_addr = server.get_join_addr(), join_port = server.get_join_port()); - let full_addr_clone = inter_addr.clone(); + let internal_id_clone = internal_id.clone(); let api = self.clone(); let handle = tokio::spawn( async move { @@ -208,7 +208,7 @@ pub trait MinecraftAPI: Send + Sync + 'static { drop(status_json); let map = api.get_map(); let mut guard = map.lock().await; - guard.remove(&full_addr_clone); + guard.remove(&internal_id_clone); drop(guard); drop(map); if let Err(err) = server.stop().await { @@ -226,7 +226,7 @@ pub trait MinecraftAPI: Send + Sync + 'static { ); let map = self.get_map(); let mut guard = map.lock().await; - guard.insert(inter_addr.clone(), handle); + guard.insert(internal_id, handle); drop(guard); Ok(()) diff --git a/src/proxy.rs b/src/proxy.rs index 8a23bc2..24a6aca 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -1,5 +1,4 @@ use evalexpr::*; -use std::collections::HashMap; use std::env; use std::net::SocketAddr; use std::time::Duration;