feat: add url type

src/kube_cache.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, fmt, sync::Arc, time::Duration};
+use std::{collections::HashMap, fmt, sync::Arc};
 
 use async_trait::async_trait;
 use futures::StreamExt;
@@ -13,7 +13,7 @@ use serde_json::json;
 use tokio::task::JoinHandle;
 
 use crate::{
-    mc_server::{sanitize_addr, MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus},
+    mc_server::{MinecraftAPI, MinecraftServerHandle, ServerDeploymentStatus, Url},
     OpaqueError,
 };
 const MAIN_LABEL: &str = "tami.moe/minecraft";
@@ -106,11 +106,11 @@ impl KubeCache {
         self.srv_api.list(&lp).await.unwrap()
     }
 
-    pub async fn query_dep(&self, addr: &str, port: &str) -> Option<Arc<Deployment>> {
+    pub async fn query_dep(&self, url: &Url) -> Option<Arc<Deployment>> {
-        self.dep_cache.find(|x| filter_label_value(x, addr, port))
+        self.dep_cache.find(|x| filter_label_value(x, url))
     }
 
-    pub async fn query_srv(&self, addr: &str, port: &str) -> Option<Service> {
+    pub async fn query_srv(&self, url: &Url) -> Option<Service> {
         let deploys = self.list_srvs().await;
         deploys.into_iter().find(|x| {
             let in_cluster = match x.spec.as_ref().unwrap().type_.as_ref() {
@@ -118,7 +118,7 @@ impl KubeCache {
                 None => false,
             };
             let incorrect_type = in_cluster ^ self.running_in_cluster;
-            !incorrect_type && filter_label_value(x, addr, port)
+            !incorrect_type && filter_label_value(x, url)
         })
     }
 
@@ -137,15 +137,11 @@ pub struct McApi {
 
 #[async_trait]
 impl MinecraftAPI<Server> for McApi {
-    #[tracing::instrument(
+    #[tracing::instrument(name = "MinecraftAPI::query_server", level = "info", skip(self, url))]
-        name = "MinecraftAPI::query_server",
+    async fn query_server(&self, url: &Url) -> Result<Server, OpaqueError> {
-        level = "info",
+        // let addr = sanitize_addr(&addr);
-        skip(self, addr, port)
-    )]
-    async fn query_server(&self, addr: &str, port: &str) -> Result<Server, OpaqueError> {
-        let addr = sanitize_addr(&addr);
 
-        let deployment = match self.cache.query_dep(&addr, &port).await {
+        let deployment = match self.cache.query_dep(url).await {
             Some(x) => x,
             None => {
                 return Err(OpaqueError::create_with_kind(
@@ -154,7 +150,7 @@ impl MinecraftAPI<Server> for McApi {
                 ))
             }
         };
-        let service = match self.cache.query_srv(&addr, &port).await {
+        let service = match self.cache.query_srv(url).await {
             Some(x) => x,
             None => {
                 return Err(OpaqueError::create_with_kind(
@@ -204,8 +200,7 @@ impl MinecraftAPI<Server> for McApi {
             }
         };
         return Ok(Server {
-            server_addr: addr.to_string(),
+            server_url: url.to_owned(),
-            server_port: port.to_string(),
             internal_address: inter_addr,
             resource_name,
             cache: self.cache.clone(),
@@ -234,8 +229,7 @@ impl McApi {
 
 #[derive(Clone)]
 pub struct Server {
-    server_addr: String,
+    server_url: Url,
-    server_port: String,
     internal_address: String,
     resource_name: String,
     cache: KubeCache,
@@ -244,10 +238,7 @@ impl fmt::Debug for Server {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("KubeServer")
             .field("internal_id", &self.resource_name)
-            .field(
+            .field("join_url", &self.server_url)
-                "join_addr",
-                &format!("{}:{}", self.server_addr, self.server_port),
-            )
             .finish()
     }
 }
@@ -323,12 +314,8 @@ impl MinecraftServerHandle for Server {
         self.resource_name.as_str()
     }
 
-    fn get_join_addr(&self) -> &str {
+    fn get_join_url(&self) -> &Url {
-        self.server_addr.as_ref()
+        &self.server_url
-    }
-
-    fn get_join_port(&self) -> &str {
-        self.server_port.as_ref()
     }
 
     fn get_motd(&self) -> Option<String> {
@@ -359,39 +346,38 @@ impl MinecraftServerHandle for Server {
 impl Server {
     async fn set_scale(&self, num: i32) -> Result<(), kube::Error> {
         let _res = self.cache.set_dep_scale(&self.resource_name, num).await?;
-        tracing::info!(
+        tracing::info!("scaled replicas of {} to {num}", self.server_url,);
-            "scaled replicas of {}:{} to {num}",
-            self.server_addr,
-            self.server_port
-        );
         Ok(())
     }
 }
 
-fn filter_label_value<R>(res: &R, addr: &str, port: &str) -> bool
+fn filter_label_value<R>(res: &R, url: &Url) -> bool
 where
     R: ResourceExt,
 {
+    let addr = url.get_authority();
+    let port = url.get_port();
+
     let labels_iter = res.labels().iter();
     let annotations_iter = res.annotations().iter();
     let iters = labels_iter.merge(annotations_iter);
 
-    let mut found_port = false;
     iters
         .filter(|(key, value)| match key.as_str() {
             MAIN_LABEL => value.as_str() == addr,
-            PORT_LABEL => {
+            PORT_LABEL => match port {
-                found_port = true;
+                Some(p) => value.as_str() == p,
-                value.as_str() == port
+                None => false,
-            }
+            },
             URL_ANNOTATION => value
                 .lines()
                 .find(|str| {
-                    if str.find(':').is_some() {
+                    // if str.find(':').is_some() {
-                        str == &format!("{addr}:{port}").as_str()
+                    //     str == &format!("{addr}:{port}").as_str()
-                    } else {
+                    // } else {
-                        str == &addr
+                    //     str == &addr
-                    }
+                    // }
+                    str.parse::<Url>().ok().as_ref() == Some(url)
                 })
                 .is_some(),
             _ => false,

src/mc_server/mod.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, sync::Arc};
+use std::{borrow::Cow, collections::HashMap, str::FromStr, sync::Arc};
 
 use async_trait::async_trait;
 use tokio::net::TcpStream;
@@ -20,8 +20,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone {
     async fn query_status(&self) -> Result<ServerDeploymentStatus, OpaqueError>;
     fn get_internal_address(&self) -> &str;
     fn get_internal_id(&self) -> &str;
-    fn get_join_addr(&self) -> &str;
+    fn get_join_url(&self) -> &Url;
-    fn get_join_port(&self) -> &str;
     fn get_motd(&self) -> Option<String>;
 
     async fn query_server_connectable(&self) -> Result<TcpStream, OpaqueError> {
@@ -72,7 +71,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_join_addr().to_string()),
+                    crate::types::VarString::from(self.get_join_url().get_authority().to_string()),
                     crate::types::UShort::from(1234),
                     crate::types::VarInt::from(1).ok_or("could not create VarInt WTF?")?,
                 )
@@ -109,7 +108,7 @@ pub trait MinecraftServerHandle: Send + Sync + 'static + Clone {
 
 #[async_trait]
 pub trait MinecraftAPI<T>: Send + Sync + 'static {
-    async fn query_server(&self, addr: &str, port: &str) -> Result<T, OpaqueError>;
+    async fn query_server(&self, url: &Url) -> Result<T, OpaqueError>;
     fn get_map(&self) -> Arc<tokio::sync::Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>;
 
     /// This should be callable even if there is already a watcher,
@@ -129,7 +128,7 @@ pub trait MinecraftAPI<T>: Send + Sync + 'static {
                 return Ok(());
             }
         }
-        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 span = tracing::span!(parent: None,tracing::Level::INFO, "server_watcher", internal_id, join_url =% server.get_join_url());
 
         let internal_id_clone = internal_id.clone();
         let api = self.clone();
@@ -190,7 +189,114 @@ pub enum ServerDeploymentStatus {
     Unavailable(String),
 }
 
+/// Minecraft specific URL handler.
+///
+/// (I do not know if this exhaustively handles all kinds of URL's
+/// supported by the minecraft client)
+///
+/// names are inspired by https://en.wikipedia.org/wiki/URL
+#[derive(Clone, Debug)]
+pub struct Url {
+    authority: Box<str>,
+    port: Port,
+}
+impl Url {
+    pub fn get_port(&self) -> Option<&str> {
+        match &self.port {
+            Port::Defaullt => None,
+            Port::Other(x) => Some(x.as_ref()),
+        }
+    }
+
+    /// This returns what is usually the DNS address or ip address.
+    pub fn get_authority(&self) -> &str {
+        self.authority.as_ref()
+    }
+
+    pub fn new(addr: &str, port: &str) -> Url {
+        let addr = sanitize_addr(addr);
+        Url {
+            authority: addr.to_string().into_boxed_str(),
+            port: Port::new(Some(port.to_string().into_boxed_str())),
+        }
+    }
+}
+
+impl FromStr for Url {
+    type Err = MinecraftURLParseError;
+
+    fn from_str(addr: &str) -> Result<Self, Self::Err> {
+        let addr = sanitize_addr(addr);
+        let mut iter = addr.split(":");
+
+        let authority = iter
+            .next()
+            .ok_or(MinecraftURLParseError::UnreachableIterFailed)?;
+        let port = iter.next().map(|str| str.to_string().into_boxed_str());
+
+        if let Some(_) = iter.next() {
+            return Err(MinecraftURLParseError::TooManyColons);
+        }
+
+        Ok(Self {
+            authority: authority.to_string().into_boxed_str(),
+            port: Port::new(port),
+        })
+    }
+}
+#[derive(Debug, Clone, PartialEq)]
+enum Port {
+    Defaullt,
+    Other(Box<str>),
+}
+impl Port {
+    pub fn new(p: Option<Box<str>>) -> Self {
+        match p {
+            Some(p) => {
+                if p.as_ref() == "25565" {
+                    Self::Defaullt
+                } else {
+                    Self::Other(p)
+                }
+            }
+            None => Self::Defaullt,
+        }
+    }
+}
+#[derive(Debug)]
+pub enum MinecraftURLParseError {
+    TooManyColons,
+    UnreachableIterFailed,
+}
+
+impl PartialEq<&Url> for Url {
+    fn eq(&self, other: &&Url) -> bool {
+        self.port == other.port && self.authority == other.authority
+    }
+}
+impl PartialEq<Url> for Url {
+    fn eq(&self, other: &Url) -> bool {
+        self.port == other.port && self.authority == other.authority
+    }
+}
+
+impl std::fmt::Display for Url {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        if let Port::Other(p) = &self.port {
+            write!(f, "{}:{}", self.authority, p)
+        } else {
+            write!(f, "{}", self.authority)
+        }
+    }
+}
+
+// impl Debug for MinecraftURL<'_> {}
 pub fn sanitize_addr(addr: &str) -> &str {
+    // Modded minecraft clients send null terminated strings,
+    // after which they have extra data. This just removes them
+    // from the addr lookup
+    let addr = terminate_at_null(addr);
+
     // Thanks to a buggy minecraft, when the client sends a join
     // from a SRV DNS record, it will not use the address typed
     // in the game, but use the address redicted *to* by the
@@ -207,10 +313,6 @@ pub fn sanitize_addr(addr: &str) -> &str {
     //         (the address used in the protocol)
     let addr = addr.trim_end_matches(".");
 
-    // Modded minecraft clients send null terminated strings,
-    // after which they have extra data. This just removes them
-    // from the addr lookup
-    let addr = terminate_at_null(addr);
     addr
 }
 

src/proxy.rs

@@ -138,21 +138,20 @@ async fn process_connection<T: MinecraftServerHandle>(
         // this is needed because of the filter
         None => return Ok(()),
     };
+    let join_url = mc_server::Url::new(
+        &handshake.get_server_address(),
+        &handshake.server_port.get_value().to_string(),
+    );
 
     let next_server_state = handshake.get_next_state();
     match next_server_state {
         packets::ProtocolState::Status => {
-            handle_status(&mut client_stream, &handshake, api).await?;
+            handle_status(&mut client_stream, &handshake, api, &join_url).await?;
         }
         packets::ProtocolState::Login => {
             // This block of packet parsing is needed here, so the span with the
             // username is correctly propagated due to the async nature of things
-            let span = tracing::span!(
+            let span = tracing::span!(tracing::Level::INFO, "login_username_extract", %join_url);
-                tracing::Level::INFO,
-                "login_username_extract",
-                join_addr = handshake.get_server_address(),
-                join_port = handshake.server_port.get_value()
-            );
 
             let packet = Packet::parse(&mut client_stream)
                 .instrument(span.clone())
@@ -161,7 +160,7 @@ async fn process_connection<T: MinecraftServerHandle>(
                 .instrument(span.clone())
                 .await
                 .ok_or("Failed to parse login start packet".to_string())?;
-            handle_login(&mut client_stream, &handshake, login_packet, api).await?
+            handle_login(&mut client_stream, &handshake, login_packet, api, &join_url).await?
         }
         packets::ProtocolState::Transfer => {
             return Err(OpaqueError::create(
@@ -174,11 +173,12 @@ async fn process_connection<T: MinecraftServerHandle>(
     Ok(())
 }
 
-#[tracing::instrument(level = "info", fields(join_addr = handshake.get_server_address(),join_port = handshake.server_port.get_value()),skip(client_stream, handshake, api))]
+#[tracing::instrument(level = "info", fields(join_url =% join_url),skip(client_stream, handshake, api, join_url))]
 async fn handle_status<T: MinecraftServerHandle>(
     client_stream: &mut TcpStream,
     handshake: &Handshake,
     api: impl MinecraftAPI<T>,
+    join_url: &mc_server::Url,
 ) -> Result<(), OpaqueError> {
     let client_packet = Packet::parse(client_stream).await?;
     if client_packet.id.get_int() != 0 {
@@ -188,24 +188,17 @@ async fn handle_status<T: MinecraftServerHandle>(
         )));
     };
 
-    let join_addr = handshake.get_server_address();
     let mut status_struct = StatusStructNew::create();
     status_struct.version.protocol = handshake.protocol_version.get_int();
 
-    let server = match api
+    let server = match api.query_server(join_url).await {
-        .query_server(
-            &handshake.get_server_address(),
-            &handshake.server_port.get_value().to_string(),
-        )
-        .await
-    {
         Ok(x) => x,
         Err(e) => {
             let span = tracing::span!(tracing::Level::INFO, "unavailable", err = e.get_kind());
             status_struct.players.max = 0;
             status_struct.players.online = 0;
             status_struct.description.text = format!(
-                "Could not find §kserver§r: §f§o{join_addr}§r\nMinecraft Ingress    - {BYE_MESSAGE}"
+                "Could not find §kserver§r: §f§o{join_url}§r\nMinecraft Ingress    - {BYE_MESSAGE}"
             );
 
             mc_server::helpers::complete_status_request(client_stream, status_struct)
@@ -265,22 +258,18 @@ async fn handle_status<T: MinecraftServerHandle>(
     return mc_server::helpers::handle_ping(client_stream).await;
 }
 
-#[tracing::instrument(level = "info", fields(join_addr = handshake.get_server_address(),join_port = handshake.server_port.get_value(),username = login_start.name.get_value()),skip(client_stream, handshake, api, login_start))]
+#[tracing::instrument(level = "info", fields(join_url =% join_url), skip(client_stream, handshake, api, login_start, join_url))]
 async fn handle_login<T: MinecraftServerHandle>(
     client_stream: &mut TcpStream,
     handshake: &Handshake,
     login_start: LoginStart,
     api: impl MinecraftAPI<T> + Send + Sync + 'static + Clone,
+    join_url: &mc_server::Url,
 ) -> Result<(), OpaqueError>
 where
     T: Send + Sync + 'static,
 {
-    let server = api
+    let server = api.query_server(join_url).await;
-        .query_server(
-            &handshake.get_server_address(),
-            &handshake.server_port.get_value().to_string(),
-        )
-        .await;
 
     let status = match server.as_ref() {
         Ok(x) => x.query_status().await?,