Compare commits
2 commits
29ff12f115
...
1e281cf61f
| Author | SHA1 | Date | |
|---|---|---|---|
| 1e281cf61f | |||
| 0439f362dd |
5 changed files with 60 additions and 24 deletions
10
Cargo.lock
generated
10
Cargo.lock
generated
|
|
@ -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",
|
||||||
|
|
|
||||||
|
|
@ -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"
|
||||||
|
|
|
||||||
|
|
@ -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()
|
||||||
|
|
|
||||||
|
|
@ -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(())
|
||||||
|
|
|
||||||
|
|
@ -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;
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue