Compare commits

..

2 commits

Author SHA1 Message Date
1e281cf61f chore: rename things to be more to my liking
All checks were successful
/ build (push) Successful in 8m18s
2026-06-15 22:00:44 +02:00
0439f362dd feat: add support for setting multiple address
I did it like this, due to the fact that it would have been a pain
to work out syntax for labels.
So I moved the url to annotations
2026-06-15 21:58:57 +02:00
5 changed files with 60 additions and 24 deletions

10
Cargo.lock generated
View file

@ -745,6 +745,15 @@ dependencies = [
"hashbrown 0.16.1", "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]] [[package]]
name = "itoa" name = "itoa"
version = "1.0.15" version = "1.0.15"
@ -970,6 +979,7 @@ dependencies = [
"either", "either",
"evalexpr", "evalexpr",
"futures", "futures",
"itertools",
"k8s-openapi", "k8s-openapi",
"kube", "kube",
"nix", "nix",

View file

@ -43,3 +43,4 @@ strip-ansi-escapes = "0.2.1"
evalexpr = { version = "13.1.0", features = ["regex"] } evalexpr = { version = "13.1.0", features = ["regex"] }
async-trait = "0.1.89" async-trait = "0.1.89"
tokio-util = { version = "0.7.18", features = ["rt"] } tokio-util = { version = "0.7.18", features = ["rt"] }
itertools = "0.14.0"

View file

@ -2,6 +2,7 @@ use std::{collections::HashMap, fmt, sync::Arc, time::Duration};
use async_trait::async_trait; use async_trait::async_trait;
use futures::StreamExt; use futures::StreamExt;
use itertools::Itertools;
use k8s_openapi::api::{apps::v1::Deployment, core::v1::Service}; use k8s_openapi::api::{apps::v1::Deployment, core::v1::Service};
use kube::{ use kube::{
api::{ListParams, ObjectList, Patch, PatchParams}, api::{ListParams, ObjectList, Patch, PatchParams},
@ -17,12 +18,13 @@ use crate::{
}; };
const MAIN_LABEL: &str = "tami.moe/minecraft"; const MAIN_LABEL: &str = "tami.moe/minecraft";
const PORT_LABEL: &str = "tami.moe/minecraft-port"; 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. /// This is the layer who is respinsible for caching requests.
///
/// TODO: // A heads-up:
/// It should be also clone-able freely, because it deals with // It should be also clone-able freely, because it deals with
/// the underlying async data access. // the underlying async data access.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct KubeCache { pub struct KubeCache {
// When the store gets copied, it points to the same backing store // 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 /// 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. /// 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<()>)> { pub async fn create() -> Option<(KubeCache, JoinHandle<()>)> {
let in_cluster = match std::env::var("KUBERNETES_SERVICE_HOST") { let in_cluster = match std::env::var("KUBERNETES_SERVICE_HOST") {
Ok(_) => true, Ok(_) => true,
@ -59,7 +66,7 @@ impl KubeCache {
.for_each(|_o| std::future::ready(())); .for_each(|_o| std::future::ready(()));
let _res = infinite_watch.await; let _res = infinite_watch.await;
tracing::error!( 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<Arc<Deployment>> { fn get_dep(&self, name: &str) -> Option<Arc<Deployment>> {
// 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) self.dep_cache.find(|x| x.name_any() == name)
} }
async fn get_srv(&self, name: &str) -> Result<Service, kube::Error> { async fn get_srv(&self, name: &str) -> Result<Service, kube::Error> {
@ -297,13 +307,16 @@ impl MinecraftServerHandle for Server {
fn get_internal_address(&self) -> &str { fn get_internal_address(&self) -> &str {
self.internal_address.as_str() self.internal_address.as_str()
} }
fn get_internal_id(&self) -> &str {
fn get_addr(&self) -> String { self.internal_id.as_str()
self.server_addr.clone()
} }
fn get_port(&self) -> String { fn get_join_addr(&self) -> &str {
self.server_port.clone() self.server_addr.as_ref()
}
fn get_join_port(&self) -> &str {
self.server_port.as_ref()
} }
fn get_motd(&self) -> Option<String> { fn get_motd(&self) -> Option<String> {
@ -347,15 +360,28 @@ fn filter_label_value<R>(res: &R, addr: &str, port: &str) -> bool
where where
R: ResourceExt, 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; let mut found_port = false;
res.labels() iters
.iter()
.filter(|(key, value)| match key.as_str() { .filter(|(key, value)| match key.as_str() {
MAIN_LABEL => value.as_str() == addr, MAIN_LABEL => value.as_str() == addr,
PORT_LABEL => { PORT_LABEL => {
found_port = true; found_port = true;
value.as_str() == port 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, _ => false,
}) })
.count() .count()

View file

@ -70,8 +70,9 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone {
async fn stop(&self) -> Result<(), OpaqueError>; async fn stop(&self) -> Result<(), OpaqueError>;
async fn query_status(&self) -> Result<ServerDeploymentStatus, OpaqueError>; async fn query_status(&self) -> Result<ServerDeploymentStatus, OpaqueError>;
fn get_internal_address(&self) -> &str; fn get_internal_address(&self) -> &str;
fn get_addr(&self) -> String; fn get_internal_id(&self) -> &str;
fn get_port(&self) -> String; fn get_join_addr(&self) -> &str;
fn get_join_port(&self) -> &str;
fn get_motd(&self) -> Option<String>; fn get_motd(&self) -> Option<String>;
async fn query_server_connectable(&self) -> Result<TcpStream, OpaqueError> { async fn query_server_connectable(&self) -> Result<TcpStream, OpaqueError> {
@ -122,7 +123,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone {
ServerDeploymentStatus::Connectable(mut tcp_stream) => { ServerDeploymentStatus::Connectable(mut tcp_stream) => {
let handshake = crate::packets::serverbound::handshake::Handshake::create( let handshake = crate::packets::serverbound::handshake::Handshake::create(
crate::types::VarInt::from(746).ok_or("could not create VarInt WTF?")?, 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::UShort::from(1234),
crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?, crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?,
) )
@ -162,7 +163,6 @@ pub trait MinecraftAPI<T>: Send + Sync + 'static {
async fn query_server(&self, addr: &str, port: &str) -> Result<T, OpaqueError>; async fn query_server(&self, addr: &str, port: &str) -> Result<T, OpaqueError>;
fn get_map(&self) -> Arc<tokio::sync::Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>; fn get_map(&self) -> Arc<tokio::sync::Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>;
// TODO: move the implementation to here, but
/// This should be callable even if there is already a watcher, /// This should be callable even if there is already a watcher,
/// and it should handle the collision itself while returning OK(). /// and it should handle the collision itself while returning OK().
async fn start_watch( async fn start_watch(
@ -171,18 +171,18 @@ pub trait MinecraftAPI<T>: Send + Sync + 'static {
frequency: std::time::Duration, frequency: std::time::Duration,
) -> Result<(), OpaqueError> ) -> Result<(), OpaqueError>
where 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() { if !handle.is_finished() {
return Ok(()); 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 api = self.clone();
let handle = tokio::spawn( let handle = tokio::spawn(
async move { async move {
@ -208,7 +208,7 @@ pub trait MinecraftAPI<T>: Send + Sync + 'static {
drop(status_json); drop(status_json);
let map = api.get_map(); let map = api.get_map();
let mut guard = map.lock().await; let mut guard = map.lock().await;
guard.remove(&full_addr_clone); guard.remove(&internal_id_clone);
drop(guard); drop(guard);
drop(map); drop(map);
if let Err(err) = server.stop().await { if let Err(err) = server.stop().await {
@ -226,7 +226,7 @@ pub trait MinecraftAPI<T>: Send + Sync + 'static {
); );
let map = self.get_map(); let map = self.get_map();
let mut guard = map.lock().await; let mut guard = map.lock().await;
guard.insert(inter_addr.clone(), handle); guard.insert(internal_id, handle);
drop(guard); drop(guard);
Ok(()) Ok(())

View file

@ -1,5 +1,4 @@
use evalexpr::*; use evalexpr::*;
use std::collections::HashMap;
use std::env; use std::env;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::time::Duration; use std::time::Duration;