diff --git a/Cargo.lock b/Cargo.lock index f26aa05..8c755c0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -60,9 +60,9 @@ dependencies = [ "bitflags 1.3.2", "bytes", "futures-util", - "http", - "http-body", - "hyper", + "http 0.2.12", + "http-body 0.4.6", + "hyper 0.14.32", "itoa", "matchit", "memchr", @@ -90,8 +90,8 @@ dependencies = [ "async-trait", "bytes", "futures-util", - "http", - "http-body", + "http 0.2.12", + "http-body 0.4.6", "mime", "rustversion", "tower-layer", @@ -125,6 +125,15 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f68f53c83ab957f72c32642f3868eec03eb974d1fb82e453128456482613d36" +[[package]] +name = "bstr" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234113d19d0d7d613b40e86fb654acf958910802bcceab913a4f9e7cda03b1a4" +dependencies = [ + "memchr", +] + [[package]] name = "bumpalo" version = "3.17.0" @@ -170,6 +179,51 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "cityhash-rs" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93a719913643003b84bd13022b4b7e703c09342cd03b679c4641c7d2e50dc34d" + +[[package]] +name = "clickhouse" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9894248c4c5a4402f76a56c273836a0c32547ec8a68166aedee7e01b7b8d102" +dependencies = [ + "bstr", + "bytes", + "cityhash-rs", + "clickhouse-derive", + "futures", + "futures-channel", + "http-body-util", + "hyper 1.6.0", + "hyper-util", + "lz4_flex", + "quanta", + "replace_with", + "sealed", + "serde", + "static_assertions", + "thiserror 1.0.69", + "time", + "tokio", + "url", +] + +[[package]] +name = "clickhouse-derive" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d70f3e2893f7d3e017eeacdc9a708fbc29a10488e3ebca21f9df6a5d2b616dbb" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn", +] + [[package]] name = "combine" version = "4.6.7" @@ -216,6 +270,15 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "deranged" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c9e6a11ca8224451684bc0d7d5a7adbf8f2fd6887261a1cfc3c0432f9d4068e" +dependencies = [ + "powerfmt", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -277,6 +340,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.31" @@ -284,6 +362,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -292,6 +371,34 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +[[package]] +name = "futures-executor" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "futures-sink" version = "0.3.31" @@ -316,9 +423,13 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", "slab", @@ -414,6 +525,17 @@ dependencies = [ "itoa", ] +[[package]] +name = "http" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f4a85d31aea989eead29a3aaf9e1115a180df8282431156e533de47660892565" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http-body" version = "0.4.6" @@ -421,7 +543,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" dependencies = [ "bytes", - "http", + "http 0.2.12", + "pin-project-lite", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http 1.3.1", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http 1.3.1", + "http-body 1.0.1", "pin-project-lite", ] @@ -453,8 +598,8 @@ dependencies = [ "futures-channel", "futures-core", "futures-util", - "http", - "http-body", + "http 0.2.12", + "http-body 0.4.6", "httparse", "httpdate", "itoa", @@ -466,6 +611,45 @@ dependencies = [ "want", ] +[[package]] +name = "hyper" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc2b571658e38e0c01b1fdca3bbbe93c00d3d71693ff2770043f8c29bc7d6f80" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http 1.3.1", + "http-body 1.0.1", + "httparse", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-util" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf9f1e950e0d9d1d3c47184416723cf29c0d1f93bd8cccf37e4beb6b44f31710" +dependencies = [ + "bytes", + "futures-channel", + "futures-util", + "http 1.3.1", + "http-body 1.0.1", + "hyper 1.6.0", + "libc", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + [[package]] name = "icu_collections" version = "1.5.0" @@ -660,9 +844,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.170" +version = "0.2.172" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "875b3680cb2f8f71bdcf9a30f38d48282f5d3c95cbf9b3fa57269bb5d5c06828" +checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa" [[package]] name = "litemap" @@ -686,6 +870,12 @@ version = "0.4.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30bde2b3dc3671ae49d8e2e9f044c7c005836e7a023ee57cffa25ab82764bb9e" +[[package]] +name = "lz4_flex" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75761162ae2b0e580d7e7c390558127e5f01b4194debd6221fd8c207fc80e3f5" + [[package]] name = "matchit" version = "0.7.3" @@ -740,6 +930,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-conv" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" + [[package]] name = "num-integer" version = "0.1.46" @@ -846,6 +1042,12 @@ version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "350e9b48cbc6b0e028b0473b114454c6316e57336ee184ceab6e53f72c178b3e" +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.20" @@ -884,6 +1086,7 @@ name = "quiclime" version = "0.1.0" dependencies = [ "axum", + "clickhouse", "env_logger", "eyre", "governor", @@ -896,6 +1099,7 @@ dependencies = [ "serde", "serde_json", "thiserror 1.0.69", + "time", "tokio", ] @@ -1069,6 +1273,12 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" +[[package]] +name = "replace_with" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51743d3e274e2b18df81c4dc6caf8a5b8e15dbe799e0dca05c7617380094e884" + [[package]] name = "ring" version = "0.17.11" @@ -1214,6 +1424,17 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "sealed" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22f968c5ea23d555e670b449c1c5e7b2fc399fdaec1d304a17cd48e288abc107" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "security-framework" version = "2.11.1" @@ -1258,6 +1479,17 @@ dependencies = [ "syn", ] +[[package]] +name = "serde_derive_internals" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d26a20a969b9e3fdf2fc2d9f21eda6c40e2de84c9408bb5d3b05d499aae711" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "serde_json" version = "1.0.139" @@ -1315,9 +1547,9 @@ checksum = "7fcf8323ef1faaee30a44a340193b1ac6814fd9b7b4e88e9d4519a3e4abe1cfd" [[package]] name = "socket2" -version = "0.5.8" +version = "0.5.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c970269d99b64e60ec3bd6ad27270092a5394c4e309314b18ae3fe575695fbe8" +checksum = "4f5fd57c80058a56cf5c777ab8a126398ece8e442983605d280a44ce79d0edef" dependencies = [ "libc", "windows-sys 0.52.0", @@ -1338,6 +1570,12 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "subtle" version = "2.6.1" @@ -1421,6 +1659,25 @@ dependencies = [ "syn", ] +[[package]] +name = "time" +version = "0.3.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a7619e19bc266e0f9c5e6686659d394bc57973859340060a69221e57dbc0c40" +dependencies = [ + "deranged", + "num-conv", + "powerfmt", + "serde", + "time-core", +] + +[[package]] +name = "time-core" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9e9a38711f559d9e3ce1cdb06dd7c5b8ea546bc90052da6d06bb76da74bb07c" + [[package]] name = "tinystr" version = "0.7.6" @@ -1539,6 +1796,17 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" +[[package]] +name = "url" +version = "2.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32f8b686cadd1473f4bd0117a5d28d36b1ade384ea9b5069a1c40aefed7fda60" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", +] + [[package]] name = "utf16_iter" version = "1.0.5" diff --git a/Cargo.toml b/Cargo.toml index d0f3760..eb2b434 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,6 +8,7 @@ license = "MIT OR Apache-2.0" [dependencies] axum = "0.6.18" +clickhouse = { version = "0.13.2", features = ["inserter", "time"] } env_logger = "0.10.0" eyre = "0.6.12" governor = "0.10.0" @@ -20,6 +21,7 @@ rustls-pemfile = "2" serde = { version = "1.0.164", features = ["derive"] } serde_json = "1.0.97" thiserror = "1.0.40" +time = "0.3.41" tokio = { version = "1.28.2", features = ["rt-multi-thread", "fs", "macros", "io-util", "net"] } [profile.release] diff --git a/flake.nix b/flake.nix index d8fbcb7..cf843a6 100644 --- a/flake.nix +++ b/flake.nix @@ -115,6 +115,36 @@ example = "/path/to/key.pem"; description = lib.mdDoc "Path to TLS key to use for quiclime connections."; }; + + clickhouseUrl = mkOption { + type = types.str; + example = "http://clickhouse:8123"; + description = lib.mdDoc "Clickhouse URL to submit metrics to."; + }; + + clickhouseUser = mkOption { + type = types.str; + example = "quiclime"; + description = lib.mdDoc "Clickhouse user."; + }; + + clickhousePasswordPath = mkOption { + type = types.str; + example = "/clickhouse_password"; + description = lib.mdDoc "Path to file containing the Clickhouse user's password."; + }; + + clickhouseDatabase = mkOption { + type = types.str; + example = "default"; + description = lib.mdDoc "Name of Clickhouse database."; + }; + + clickhouseTable = mkOption { + type = types.str; + example = "mc_connections"; + description = lib.mdDoc "Name of Clickhouse table."; + }; }; }; @@ -141,6 +171,11 @@ QUICLIME_BIND_ADDR_WEB = cfg.controlAddr; QUICLIME_CERT_PATH = cfg.cert; QUICLIME_KEY_PATH = cfg.key; + CLICKHOUSE_URL = cfg.clickhouseUrl; + CLICKHOUSE_USER = cfg.clickhouseUser; + CLICKHOUSE_PASSWORD_PATH = cfg.clickhousePasswordPath; + CLICKHOUSE_DB = cfg.clickhouseDatabase; + CLICKHOUSE_TABLE = cfg.clickhouseTable; }; }; diff --git a/src/main.rs b/src/main.rs index e5d506a..55d4eb4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,21 +2,30 @@ #![allow(clippy::cast_possible_truncation)] #![allow(clippy::cast_possible_wrap)] -use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration}; +use std::{ + convert::Infallible, + net::{Ipv6Addr, SocketAddr}, + sync::Arc, + time::Duration, +}; use axum::{ http::StatusCode, routing::{get, post}, }; +use clickhouse::{inserter::Inserter, Row}; use eyre::{eyre, Context}; use log::{error, info, warn}; use netty::{Handshake, ReadError}; +use parking_lot::Mutex; use quinn::{ crypto::rustls::QuicServerConfig, rustls::pki_types::{CertificateDer, PrivateKeyDer}, ConnectionError, Endpoint, Incoming, ServerConfig, TransportConfig, }; use routing::{RoutingError, RoutingTable}; +use serde::Serialize; +use time::OffsetDateTime; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::TcpStream, @@ -64,6 +73,16 @@ async fn create_server_config() -> eyre::Result { Ok(config) } +#[derive(Row, Serialize)] +struct Connection { + #[serde(with = "clickhouse::serde::time::datetime")] + established: OffsetDateTime, + region: &'static str, + client: Ipv6Addr, + intent: &'static str, + successful: bool, +} + #[tokio::main] async fn main() -> eyre::Result<()> { env_logger::init(); @@ -78,11 +97,31 @@ async fn main() -> eyre::Result<()> { let routing_table = Box::leak(Box::new(routing::RoutingTable::new( std::env::var("QUICLIME_BASE_DOMAIN").context("Reading QUICLIME_BASE_DOMAIN")?, ))); + + let client = clickhouse::Client::default() + .with_url(std::env::var("CLICKHOUSE_URL").context("Reading CLICKHOUSE_URL")?) + .with_user(std::env::var("CLICKHOUSE_USER").context("Reading CLICKHOUSE_USER")?) + .with_password( + tokio::fs::read_to_string( + std::env::var("CLICKHOUSE_PASSWORD_PATH") + .context("Reading CLICKHOUSE_PASSWORD_PATH")?, + ) + .await + .context("Reading from CLICKHOUSE_PASSWORD_PATH")?, + ) + .with_database(std::env::var("CLICKHOUSE_DB").context("Reading CLICKHOUSE_DB")?); + let inserter: clickhouse::inserter::Inserter = client + .inserter(&std::env::var("CLICKHOUSE_TABLE").context("Reading CLICKHOUSE_TABLE")?)? + .with_timeouts(Some(Duration::from_secs(5)), Some(Duration::from_secs(20))) + .with_max_bytes(50_000_000) + .with_max_rows(750_000) + .with_period(Some(Duration::from_secs(15))); + let inserter = Arc::new(Mutex::new(inserter)); #[allow(unreachable_code)] tokio::try_join!( listen_quic(endpoint, routing_table), listen_control(endpoint, routing_table), - listen_minecraft(routing_table) + listen_minecraft(routing_table, inserter) )?; Ok(()) } @@ -219,6 +258,7 @@ async fn listen_control( async fn try_handle_minecraft( mut connection: TcpStream, routing_table: &'static RoutingTable, + inserter: Arc>>, ) -> eyre::Result<()> { let peer = connection.peer_addr()?; info!("Minecraft client connected from: {}", peer); @@ -237,6 +277,21 @@ async fn try_handle_minecraft( match routing_table.route_limited(&address, peer.ip()).await { Ok(val) => val, Err(RoutingError::InvalidDomain) => { + if let Err(e) = inserter.lock().write(&Connection { + established: OffsetDateTime::now_utc(), + region: routing_table.base_domain(), + client: match peer.ip() { + std::net::IpAddr::V4(ipv4_addr) => ipv4_addr.to_ipv6_mapped(), + std::net::IpAddr::V6(ipv6_addr) => ipv6_addr, + }, + intent: match handshake.next_state { + netty::HandshakeType::Status => "status", + netty::HandshakeType::Login => "login", + }, + successful: false, + }) { + error!("Failed to send telemetry: {e:?}"); + } return politely_disconnect(connection, handshake).await; } Err(RoutingError::RateLimited) => { @@ -244,6 +299,21 @@ async fn try_handle_minecraft( return impolitely_disconnect(connection, handshake).await; } }; + if let Err(e) = inserter.lock().write(&Connection { + established: OffsetDateTime::now_utc(), + region: routing_table.base_domain(), + client: match peer.ip() { + std::net::IpAddr::V4(ipv4_addr) => ipv4_addr.to_ipv6_mapped(), + std::net::IpAddr::V6(ipv6_addr) => ipv6_addr, + }, + intent: match handshake.next_state { + netty::HandshakeType::Status => "status", + netty::HandshakeType::Login => "login", + }, + successful: true, + }) { + error!("Failed to send telemetry: {e:?}"); + } handshake.send(&mut send_host).await?; let (mut recv_client, mut send_client) = connection.split(); tokio::select! { @@ -349,13 +419,20 @@ async fn impolitely_disconnect( Ok(()) } -async fn handle_minecraft(connection: TcpStream, routing_table: &'static RoutingTable) { - if let Err(e) = try_handle_minecraft(connection, routing_table).await { +async fn handle_minecraft( + connection: TcpStream, + routing_table: &'static RoutingTable, + inserter: Arc>>, +) { + if let Err(e) = try_handle_minecraft(connection, routing_table, inserter).await { error!("Error handling Minecraft connection: {:#}", e); }; } -async fn listen_minecraft(routing_table: &'static RoutingTable) -> eyre::Result { +async fn listen_minecraft( + routing_table: &'static RoutingTable, + inserter: Arc>>, +) -> eyre::Result { let server = tokio::net::TcpListener::bind( std::env::var("QUICLIME_BIND_ADDR_MC") .context("Reading QUICLIME_BIND_ADDR_MC")? @@ -365,7 +442,11 @@ async fn listen_minecraft(routing_table: &'static RoutingTable) -> eyre::Result< loop { match server.accept().await { Ok((connection, _)) => { - tokio::spawn(handle_minecraft(connection, routing_table)); + tokio::spawn(handle_minecraft( + connection, + routing_table, + inserter.clone(), + )); } Err(e) => { error!("Error accepting minecraft connection: {:#}", e); diff --git a/src/routing.rs b/src/routing.rs index 7faeb67..542e11a 100644 --- a/src/routing.rs +++ b/src/routing.rs @@ -102,6 +102,10 @@ impl RoutingTable { parent: self, } } + + pub fn base_domain(&self) -> &str { + &self.base_domain + } } #[allow(clippy::module_name_repetitions)]