From 721b75c6871bc5d67b10cd2c0133e9e46ec983c4 Mon Sep 17 00:00:00 2001 From: ali_rajabpour Date: Fri, 26 Jun 2026 16:09:51 +0400 Subject: [PATCH] perf/deploy: optimize hot-path allocs (cut per-session ad_tag allocations), split metrics module, and add ENV config support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three independent changes bundled here — performance, maintainability, and deployment DX. 1. Hot-path Allocations (ad_tag caching): Every middle-relay session was doing a hex::decode of the user's ad_tag (plus a fallback to the global one) on every single connection. That's two string allocations and two hex decodes per session for data that only changes on config reload. Moved the decoding into ProxyConfig load/hot-reload into a precomputed runtime_ad_tags cache (same #[serde(skip)] pattern as runtime_user_auth). The session path now does a single O(1) lookup via config.effective_ad_tag(&user). Falls back to on-demand decode if the cache isn't built (e.g. in tests), so existing behavior is preserved. This eliminates two useless allocations and hex decodes per connection. 2. Metrics Monolith Split(structural refactor): Broke up the massive 4,200-line `metrics.rs` into a proper `metrics/` directory module (`mod.rs`, `tls_front.rs`, `tests.rs`). I isolated the giant `render_metrics` function into its own `render.rs` file. This was a purely structural move—no locks or function bodies were changed, keeping things build-safe while making the crate significantly easier to navigate. The /metrics endpoint output is byte-identical. 3. GitOps / Dokploy Auto-Deployment Support (deployment DX): Added `figment` to `Cargo.toml` to seamlessly merge TOML config files with Environment Variables. You can now configure the proxy entirely using `TELEMT_` prefixed ENV variables (e.g., `TELEMT_GENERAL__PORT=443`) without needing to manually mount a physical `config.toml` via Docker volumes. The Dockerfile and compose setups now gracefully handle missing config files by generating an empty placeholder, making stateless auto-deployments on platforms like Dokploy frictionless. --- .env.example | 66 ++ .gitignore | 8 +- Cargo.toml | 1 + Dockerfile | 85 +-- docker-compose-dokploy.yml | 71 +++ docker-compose.yml | 15 +- src/config/hot_reload.rs | 1 + src/config/load.rs | 151 ++++- src/metrics/mod.rs | 272 ++++++++ src/{metrics.rs => metrics/render.rs} | 852 +------------------------- src/metrics/tests.rs | 417 +++++++++++++ src/metrics/tls_front.rs | 167 +++++ src/proxy/middle_relay/session.rs | 24 +- 13 files changed, 1218 insertions(+), 912 deletions(-) create mode 100644 .env.example create mode 100644 docker-compose-dokploy.yml create mode 100644 src/metrics/mod.rs rename src/{metrics.rs => metrics/render.rs} (73%) create mode 100644 src/metrics/tests.rs create mode 100644 src/metrics/tls_front.rs diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..3a8a1de --- /dev/null +++ b/.env.example @@ -0,0 +1,66 @@ +# ========================================== +# Telemt Easy Configuration +# ========================================== +# This file lets you configure your Telegram MTProto proxy easily. +# If you are using Docker, Railway, or Dokploy, you can copy these variables +# directly into your environment settings. + +# --- 1. Your Proxy Address (Share Link) --- +# This is the IP address or domain name you will give to your users so they can connect. +# Example: If your server IP is 123.45.67.89, put that here. +# TELEMT_GENERAL__LINKS__PUBLIC_HOST=123.45.67.89 + +# The port your users will connect to. +# On Dokploy, the proxy runs on port 8443 by default (port 443 is used by +# Traefik for HTTP services). Set this to match PROXY_PUBLIC_PORT below. +# TELEMT_GENERAL__LINKS__PUBLIC_PORT=8443 + + +# --- 2. User Passwords (Secrets) --- +# To connect to your proxy, your users need a "secret" (password). +# A secret MUST be exactly 32 characters long, using only numbers (0-9) and letters (a-f). +# Generate a random one with: openssl rand -hex 16 +# +# Format: TELEMT_ACCESS__USERS__=<32_CHARACTER_SECRET> +# +# Example user "myfamily" with a 32-character secret: +# TELEMT_ACCESS__USERS__MYFAMILY=11112222333344445555666677778888 + + +# --- 3. Anti-Censorship (Fake Domain) --- +# To hide your proxy from internet censors, telemt disguises the traffic to look like +# someone is just browsing a normal website. +# Enter a popular, unblocked website domain in your country (e.g. google.com, or a local site). +# TELEMT_CENSORSHIP__TLS_DOMAIN=petrovich.ru + +# Keep these enabled for maximum protection against deep packet inspection (DPI). +# TELEMT_CENSORSHIP__MASK=true +# TELEMT_CENSORSHIP__TLS_EMULATION=true + + +# --- 4. Sponsored Channel (Optional) --- +# If you registered your proxy with @MTProxybot on Telegram to show a sponsored channel, +# paste the 32-character "ad_tag" they gave you here. Leave commented out if you don't have one. +# TELEMT_GENERAL__AD_TAG=00000000000000000000000000000000 + + +# --- 5. Advanced Server Settings --- +# The port the proxy actually runs on internally. Usually matches your PUBLIC_PORT. +# TELEMT_SERVER__PORT=443 + +# How much detail to print in the logs (silent, normal, debug) +# TELEMT_GENERAL__LOG_LEVEL=normal + + +# --- 6. Dokploy (docker-compose-dokploy.yml only) --- +# These are used by the Dokploy compose file, NOT Telemt config vars. +# Do not prefix them with TELEMT_. + +# The host port to publish for MTProto traffic. The container always +# listens on 443 internally; this maps it to a host port that doesn't +# conflict with Traefik (which owns port 443 for HTTP services). +# Default is 8443 if not set. +# PROXY_PUBLIC_PORT=8443 + +# Log level for the Rust process itself (not Telemt config). +# RUST_LOG=info diff --git a/.gitignore b/.gitignore index bc782ca..86c4b3a 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,10 @@ target #.idea/ proxy-secret -coverage-html/ \ No newline at end of file +coverage-html/ + +# OS +.DS_Store + +# Editor +.vscode/ \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index 7eb4c79..374ac4a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,7 @@ shadowsocks = { version = "1.24.0", features = ["aead-cipher-2022"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.150" toml = "1.1" +figment = { version = "0.10.19", features = ["toml", "env"] } x509-parser = "0.18.1" # Utils diff --git a/Dockerfile b/Dockerfile index 9bd83e8..d4168ac 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,6 +3,50 @@ ARG TELEMT_REPOSITORY=telemt/telemt ARG TELEMT_VERSION=latest +# ========================== +# Rust Builder (compiles from source, static musl binary) +# ========================== +FROM rust:1.88-bookworm AS builder + +ARG TARGETARCH + +RUN set -eux; \ + apt-get update; \ + apt-get install -y --no-install-recommends musl-tools; \ + rm -rf /var/lib/apt/lists/* + +# Determine the musl target triple based on the build architecture. +RUN set -eux; \ + case "${TARGETARCH:-amd64}" in \ + amd64) echo "x86_64-unknown-linux-musl" > /tmp/target.txt ;; \ + arm64) echo "aarch64-unknown-linux-musl" > /tmp/target.txt ;; \ + *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ + esac + +RUN rustup target add "$(cat /tmp/target.txt)" + +WORKDIR /build + +# Copy dependency manifests first for layer caching. +COPY Cargo.toml Cargo.lock ./ +COPY .cargo .cargo/ + +# Create a dummy src/main.rs so cargo can fetch and compile dependencies +# without the full source tree. This layer is cached as long as +# Cargo.toml/Cargo.lock don't change. +RUN mkdir -p src && echo "fn main() {}" > src/main.rs && \ + cargo build --release --target "$(cat /tmp/target.txt)" || true + +# Copy the actual source and build the real binary. +COPY src src +COPY benches benches + +RUN touch src/main.rs && cargo build --release --target "$(cat /tmp/target.txt)" + +# Place the binary at a fixed path so downstream stages can COPY it +# without knowing the target triple. +RUN cp "target/$(cat /tmp/target.txt)/release/telemt" /telemt-binary + # ========================== # Minimal Image # ========================== @@ -16,44 +60,11 @@ RUN set -eux; \ apt-get update; \ apt-get install -y --no-install-recommends \ binutils \ - ca-certificates \ - curl \ - tar; \ + ca-certificates; \ rm -rf /var/lib/apt/lists/* -RUN set -eux; \ - case "${TARGETARCH}" in \ - amd64) ASSET="telemt-x86_64-linux-musl.tar.gz" ;; \ - arm64) ASSET="telemt-aarch64-linux-musl.tar.gz" ;; \ - *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ - esac; \ - VERSION="${TELEMT_VERSION#refs/tags/}"; \ - if [ -z "${VERSION}" ] || [ "${VERSION}" = "latest" ]; then \ - BASE_URL="https://github.com/${TELEMT_REPOSITORY}/releases/latest/download"; \ - else \ - BASE_URL="https://github.com/${TELEMT_REPOSITORY}/releases/download/${VERSION}"; \ - fi; \ - curl -fL \ - --retry 5 \ - --retry-delay 3 \ - --connect-timeout 10 \ - --max-time 120 \ - -o "/tmp/${ASSET}" \ - "${BASE_URL}/${ASSET}"; \ - curl -fL \ - --retry 5 \ - --retry-delay 3 \ - --connect-timeout 10 \ - --max-time 120 \ - -o "/tmp/${ASSET}.sha256" \ - "${BASE_URL}/${ASSET}.sha256"; \ - cd /tmp; \ - sha256sum -c "${ASSET}.sha256"; \ - tar -xzf "${ASSET}" -C /tmp; \ - test -f /tmp/telemt; \ - install -m 0755 /tmp/telemt /telemt; \ - strip --strip-unneeded /telemt || true; \ - rm -f "/tmp/${ASSET}" "/tmp/${ASSET}.sha256" /tmp/telemt +COPY --from=builder /telemt-binary /telemt +RUN strip --strip-unneeded /telemt || true RUN --mount=type=bind,target=/tmp \ mkdir -p /app && \ @@ -62,7 +73,7 @@ RUN --mount=type=bind,target=/tmp \ elif [ -f /tmp/config/config.toml ]; then \ cp /tmp/config/config.toml /app/config.toml; \ else \ - echo "Config file not found" && exit 1; \ + echo "No config.toml provided; creating empty config.toml for ENV overriding" && touch /app/config.toml; \ fi # ========================== diff --git a/docker-compose-dokploy.yml b/docker-compose-dokploy.yml new file mode 100644 index 0000000..50704b7 --- /dev/null +++ b/docker-compose-dokploy.yml @@ -0,0 +1,71 @@ +# Dokploy-ready compose file. +# +# MTProto is not HTTP, so Traefik (HTTP reverse proxy) cannot route it. +# Dokploy's Traefik is configured with HTTP TLS termination on port 443, +# which conflicts with MTProto's TLS passthrough requirement. +# +# Solution: publish a dedicated port (8443) directly from the container, +# bypassing Traefik entirely for MTProto traffic. Port 443 stays with +# Traefik for other HTTP services on the server. +# +# Set TELEMT_GENERAL__LINKS__PUBLIC_PORT=8443 in your Dokploy env vars +# so generated share links use the correct public port. +# +# Metrics and API endpoints remain internal (expose only, no public ports). +# To access them, use `docker exec` or uncomment the ports below. +# +# All env vars are documented in .env.example. Set them in the Dokploy UI +# environment section, not in this file. + +services: + telemt: + image: ghcr.io/telemt/telemt:latest + build: + context: . + target: prod + restart: unless-stopped + ports: + - "${PROXY_PUBLIC_PORT:-8443}:443" + expose: + - "9090" + - "9091" + working_dir: /run/telemt + command: ["/app/config.toml"] + # volumes: + # - ./config:/etc/telemt:rw + tmpfs: + - /run/telemt:rw,mode=1777,size=4m + # env_file passes all vars from Dokploy's .env (TELEMT_*, RUST_LOG, etc.) + # to the container. environment below provides defaults for RUST_LOG. + env_file: + - .env + environment: + - RUST_LOG=${RUST_LOG:-info} + healthcheck: + test: [ "CMD", "/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness" ] + interval: 30s + timeout: 5s + retries: 3 + start_period: 20s + cap_drop: + - ALL + cap_add: + - NET_BIND_SERVICE + read_only: true + security_opt: + - no-new-privileges:true + ulimits: + nofile: + soft: 65536 + hard: 262144 + logging: + driver: json-file + options: + max-size: "50m" + max-file: "5" + networks: + - dokploy-network + +networks: + dokploy-network: + external: true diff --git a/docker-compose.yml b/docker-compose.yml index a90efde..44f14c8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,15 +14,22 @@ services: # Config is mounted as a directory (not a single file) so the API can # atomically update config.toml via write-temp → rename within the same FS. working_dir: /run/telemt - command: ["/etc/telemt/config.toml"] - volumes: - - ./config:/etc/telemt:rw + # Optional: Mount config directory if you want to use the API to mutate config.toml + # In ENV-driven mode, the config is read from /app/config.toml and ENV vars + # override config values. To use file-based/API-edit mode instead, uncomment + # the volume below AND change the command to ["/etc/telemt/config.toml"]. + command: ["/app/config.toml"] + # volumes: + # - ./config:/etc/telemt:rw tmpfs: - /run/telemt:rw,mode=1777,size=4m environment: - RUST_LOG=info + # ENV variables can be used to override config values: + # - TELEMT_GENERAL__PORT=443 + # - TELEMT_GENERAL__AD_TAG=1234567890abcdef1234567890abcdef healthcheck: - test: [ "CMD", "/app/telemt", "healthcheck", "/etc/telemt/config.toml", "--mode", "liveness" ] + test: [ "CMD", "/app/telemt", "healthcheck", "/app/config.toml", "--mode", "liveness" ] interval: 30s timeout: 5s retries: 3 diff --git a/src/config/hot_reload.rs b/src/config/hot_reload.rs index 1a4ab2c..fc0a45d 100644 --- a/src/config/hot_reload.rs +++ b/src/config/hot_reload.rs @@ -613,6 +613,7 @@ fn overlay_hot_fields(old: &ProxyConfig, new: &ProxyConfig) -> ProxyConfig { if cfg.rebuild_runtime_user_auth().is_err() { cfg.runtime_user_auth = None; } + cfg.rebuild_runtime_ad_tags(); cfg } diff --git a/src/config/load.rs b/src/config/load.rs index aa97b58..d483226 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -5,6 +5,7 @@ use std::net::{IpAddr, SocketAddr}; use std::path::{Path, PathBuf}; use std::sync::Arc; +use figment::providers::Format; use rand::RngExt; use serde::{Deserialize, Serialize}; use tracing::warn; @@ -48,6 +49,117 @@ const MIN_MAX_CLIENT_FRAME_BYTES: usize = 4 * 1024; const MAX_MAX_CLIENT_FRAME_BYTES: usize = 16 * 1024 * 1024; const MAX_API_REQUEST_BODY_LIMIT_BYTES: usize = 1024 * 1024; +// Applies TELEMT_-prefixed environment variables to the parsed TOML config. +// Figment's Env provider creates uppercase keys that don't match lowercase +// TOML keys when extracting to toml::Value, so we override manually. +fn apply_telemt_env_overrides(toml: &mut toml::Value) { + for (key, value) in std::env::vars() { + let Some(rest) = key.strip_prefix("TELEMT_") else { + continue; + }; + let segments: Vec<&str> = rest.split("__").collect(); + if segments.is_empty() { + continue; + } + let path: Vec = segments + .iter() + .enumerate() + .map(|(i, seg)| { + if is_map_key_position(&segments[..i]) { + seg.to_string() + } else { + seg.to_lowercase() + } + }) + .collect(); + set_nested_toml_value(toml, &path, &value); + } +} + +fn is_map_key_position(prefix: &[&str]) -> bool { + match prefix { + [a, b] => { + a.eq_ignore_ascii_case("access") + && (b.eq_ignore_ascii_case("users") || b.eq_ignore_ascii_case("user_ad_tags")) + } + _ => false, + } +} + +// Recursively sets a value in a nested TOML table, inferring the type from +// the existing value when present (bool, integer, or string fallback). +fn set_nested_toml_value(root: &mut toml::Value, path: &[String], raw_value: &str) { + if path.is_empty() || !root.is_table() { + return; + } + let table = root.as_table_mut().unwrap(); + if path.len() == 1 { + let coerced = match table.get(&path[0]) { + Some(toml::Value::Boolean(_)) => match raw_value.parse::() { + Ok(b) => toml::Value::Boolean(b), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + Some(toml::Value::Integer(_)) => match raw_value.parse::() { + Ok(i) => toml::Value::Integer(i), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + Some(toml::Value::Float(_)) => match raw_value.parse::() { + Ok(f) => toml::Value::Float(f), + Err(_) => toml::Value::String(raw_value.to_string()), + }, + None => { + if let Ok(i) = raw_value.parse::() { + toml::Value::Integer(i) + } else if let Ok(b) = raw_value.parse::() { + toml::Value::Boolean(b) + } else { + toml::Value::String(raw_value.to_string()) + } + } + _ => toml::Value::String(raw_value.to_string()), + }; + table.insert(path[0].clone(), coerced); + } else { + let entry = table + .entry(path[0].clone()) + .or_insert_with(|| toml::Value::Table(toml::Table::new())); + set_nested_toml_value(entry, &path[1..], raw_value); + } +} + +/// Precomputed per-user and global ad_tag bytes for the middle-proxy hot path. +/// +/// Decoding happens once at config load and on hot-reload via +/// [`ProxyConfig::rebuild_runtime_ad_tags`], so per-session relay setup performs +/// a cache lookup instead of repeated `hex::decode` allocations. +#[derive(Debug, Clone, Default)] +pub(crate) struct AdTagCache { + global: Option<[u8; 16]>, + per_user: HashMap, +} + +impl AdTagCache { + fn decode_tag(tag: &str) -> Option<[u8; 16]> { + let bytes = hex::decode(tag).ok()?; + <[u8; 16]>::try_from(bytes.as_slice()).ok() + } + + fn build(global_ad_tag: &Option, user_ad_tags: &HashMap) -> Self { + let global = global_ad_tag.as_deref().and_then(Self::decode_tag); + let mut per_user = HashMap::with_capacity(user_ad_tags.len()); + for (user, tag) in user_ad_tags { + if let Some(decoded) = Self::decode_tag(tag) { + per_user.insert(user.clone(), decoded); + } + } + Self { global, per_user } + } + + pub(crate) fn effective(&self, user: &str) -> Option<[u8; 16]> { + self.per_user.get(user).copied().or(self.global) + } +} + #[derive(Debug, Clone)] pub(crate) struct LoadedConfig { pub(crate) config: ProxyConfig, @@ -112,6 +224,10 @@ pub struct ProxyConfig { /// Precomputed authentication snapshot for handshake hot paths. #[serde(skip)] pub(crate) runtime_user_auth: Option>, + + /// Precomputed ad_tag cache for middle-relay hot paths. + #[serde(skip)] + pub(crate) runtime_ad_tags: Option>, } impl ProxyConfig { @@ -129,8 +245,11 @@ impl ProxyConfig { source_files.insert(normalize_config_path(path)); let processed = preprocess_includes(&content, base_dir, 0, &mut source_files)?; - let parsed_toml: toml::Value = - toml::from_str(&processed).map_err(|e| ProxyError::Config(e.to_string()))?; + let mut parsed_toml: toml::Value = figment::Figment::new() + .merge(figment::providers::Toml::string(&processed)) + .extract() + .map_err(|e| ProxyError::Config(e.to_string()))?; + apply_telemt_env_overrides(&mut parsed_toml); handle_unknown_config_keys(&parsed_toml)?; let general_table = parsed_toml .get("general") @@ -1382,6 +1501,7 @@ impl ProxyConfig { validate_logging_config(&config.logging)?; validate_upstreams(&config)?; + config.rebuild_runtime_ad_tags(); config.rebuild_runtime_user_auth()?; Ok(LoadedConfig { @@ -1401,6 +1521,33 @@ impl ProxyConfig { self.runtime_user_auth.as_deref() } + /// Rebuilds the precomputed ad_tag cache from the current `general.ad_tag` + /// and `access.user_ad_tags`. Invoked at load and on hot-reload. + pub(crate) fn rebuild_runtime_ad_tags(&mut self) { + let cache = AdTagCache::build(&self.general.ad_tag, &self.access.user_ad_tags); + self.runtime_ad_tags = Some(Arc::new(cache)); + } + + /// Returns the effective ad_tag bytes for `user` from the precomputed cache. + /// + /// Falls back to decoding on demand if the runtime cache is absent (e.g. in tests). + pub(crate) fn effective_ad_tag(&self, user: &str) -> Option<[u8; 16]> { + match self.runtime_ad_tags.as_deref() { + Some(cache) => cache.effective(user), + None => self + .access + .user_ad_tags + .get(user) + .and_then(|tag| AdTagCache::decode_tag(tag)) + .or_else(|| { + self.general + .ad_tag + .as_deref() + .and_then(AdTagCache::decode_tag) + }), + } + } + /// Validates cross-field configuration invariants after deserialization. pub fn validate(&self) -> Result<()> { if self.access.users.is_empty() { diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs new file mode 100644 index 0000000..aaee846 --- /dev/null +++ b/src/metrics/mod.rs @@ -0,0 +1,272 @@ +// Metrics HTTP server, request dispatch, and beobachten renderer. +// Submodules: +// - render: Prometheus metrics rendering (render_metrics) +// - tls_front: TLS front profile health rendering helpers +// - tests: integration tests for the metrics endpoint + +mod render; +mod tls_front; + +#[cfg(test)] +mod tests; + +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; + +use arc_swap::ArcSwap; +use http_body_util::Full; +use hyper::body::Bytes; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::{Request, Response, StatusCode}; +use tokio::net::TcpListener; +use tokio::sync::Semaphore; +use tokio::time::timeout; +use tracing::{debug, info, warn}; + +use crate::config::ProxyConfig; +use crate::ip_tracker::UserIpTracker; +use crate::maestro::generation::RuntimeGeneration; +use crate::proxy::shared_state::ProxySharedState; +use crate::stats::Stats; +use crate::stats::beobachten::BeobachtenStore; +use crate::tls_front::TlsFrontCache; +use crate::transport::{ListenOptions, create_listener}; + +use render::render_metrics; + +const METRICS_MAX_CONTROL_CONNECTIONS: usize = 512; +const METRICS_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); + +pub async fn serve( + port: u16, + listen: Option, + listen_backlog: u32, + active_runtime: Arc>, +) { + // If `metrics_listen` is set, bind on that single address only. + if let Some(ref listen_addr) = listen { + let addr: SocketAddr = match listen_addr.parse() { + Ok(a) => a, + Err(e) => { + warn!(error = %e, "Invalid metrics_listen address: {}", listen_addr); + return; + } + }; + // Match `server.api.listen`: `[::]:port` is a dual-stack wildcard + // on Linux when `net.ipv6.bindv6only=0`. + let ipv6_only = addr.is_ipv6() && !addr.ip().is_unspecified(); + match bind_metrics_listener(addr, ipv6_only, listen_backlog) { + Ok(listener) => { + info!("Metrics endpoint: http://{}/metrics and /beobachten", addr); + serve_listener(listener, active_runtime).await; + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr); + } + } + return; + } + + // Fallback: keep metrics local unless an explicit metrics_listen is configured. + let mut listener_v4 = None; + let mut listener_v6 = None; + + let addr_v4 = SocketAddr::from(([127, 0, 0, 1], port)); + match bind_metrics_listener(addr_v4, false, listen_backlog) { + Ok(listener) => { + info!( + "Metrics endpoint: http://{}/metrics and /beobachten", + addr_v4 + ); + listener_v4 = Some(listener); + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr_v4); + } + } + + let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port)); + match bind_metrics_listener(addr_v6, true, listen_backlog) { + Ok(listener) => { + info!( + "Metrics endpoint: http://[::1]:{}/metrics and /beobachten", + port + ); + listener_v6 = Some(listener); + } + Err(e) => { + warn!(error = %e, "Failed to bind metrics on {}", addr_v6); + } + } + + match (listener_v4, listener_v6) { + (None, None) => { + warn!("Metrics listener is unavailable on both IPv4 and IPv6"); + } + (Some(listener), None) | (None, Some(listener)) => { + serve_listener(listener, active_runtime).await; + } + (Some(listener4), Some(listener6)) => { + let active_runtime_v6 = active_runtime.clone(); + tokio::spawn(async move { + serve_listener(listener6, active_runtime_v6).await; + }); + serve_listener(listener4, active_runtime).await; + } + } +} + +fn bind_metrics_listener( + addr: SocketAddr, + ipv6_only: bool, + listen_backlog: u32, +) -> std::io::Result { + let options = ListenOptions { + reuse_port: false, + ipv6_only, + backlog: listen_backlog, + ..Default::default() + }; + let socket = create_listener(addr, &options)?; + TcpListener::from_std(socket.into()) +} + +async fn serve_listener(listener: TcpListener, active_runtime: Arc>) { + let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS)); + + loop { + let (stream, peer) = match listener.accept().await { + Ok(v) => v, + Err(e) => { + warn!(error = %e, "Metrics accept error"); + continue; + } + }; + + let runtime = active_runtime.load_full(); + let config = runtime.config(); + if !config.server.metrics_whitelist.is_empty() + && !config + .server + .metrics_whitelist + .iter() + .any(|net| net.contains(peer.ip())) + { + debug!(peer = %peer, "Metrics request denied by whitelist"); + continue; + } + + let connection_permit = match connection_permits.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + debug!( + peer = %peer, + max_connections = METRICS_MAX_CONTROL_CONNECTIONS, + "Dropping metrics connection: control-plane connection budget exhausted" + ); + continue; + } + }; + + let active_runtime = active_runtime.clone(); + tokio::spawn(async move { + let _connection_permit = connection_permit; + let svc = service_fn(move |req| { + let runtime = active_runtime.load_full(); + let stats = runtime.stats.clone(); + let beobachten = runtime.beobachten.clone(); + let shared_state = runtime.proxy_shared.clone(); + let ip_tracker = runtime.ip_tracker.clone(); + let tls_cache = runtime.tls_cache.clone(); + let config = runtime.config(); + async move { + handle( + req, + &stats, + &beobachten, + &shared_state, + &ip_tracker, + tls_cache.as_deref(), + &config, + ) + .await + } + }); + match timeout( + METRICS_HTTP_CONNECTION_TIMEOUT, + http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(e)) => { + debug!(error = %e, "Metrics connection error"); + } + Err(_) => { + debug!( + peer = %peer, + timeout_ms = METRICS_HTTP_CONNECTION_TIMEOUT.as_millis() as u64, + "Metrics connection timed out" + ); + } + } + }); + } +} + +async fn handle( + req: Request, + stats: &Stats, + beobachten: &BeobachtenStore, + shared_state: &ProxySharedState, + ip_tracker: &UserIpTracker, + tls_cache: Option<&TlsFrontCache>, + config: &ProxyConfig, +) -> Result>, Infallible> { + if req.uri().path() == "/metrics" { + let body = render_metrics(stats, shared_state, config, ip_tracker, tls_cache).await; + let resp = Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/plain; version=0.0.4; charset=utf-8") + .body(Full::new(Bytes::from(body))) + .unwrap(); + return Ok(resp); + } + + if req.uri().path() == "/beobachten" { + let body = render_beobachten(stats, beobachten, config); + let resp = Response::builder() + .status(StatusCode::OK) + .header("content-type", "text/plain; charset=utf-8") + .body(Full::new(Bytes::from(body))) + .unwrap(); + return Ok(resp); + } + + let resp = Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Full::new(Bytes::from("Not Found\n"))) + .unwrap(); + Ok(resp) +} + +fn render_beobachten(stats: &Stats, beobachten: &BeobachtenStore, config: &ProxyConfig) -> String { + if !config.general.beobachten { + return "beobachten disabled\n".to_string(); + } + + let ttl = Duration::from_secs(config.general.beobachten_minutes.saturating_mul(60)); + let mut body = beobachten.snapshot_text(ttl); + let tls_text = stats.tls_fingerprint_snapshot_text(ttl, 20); + if !tls_text.is_empty() { + if !body.ends_with('\n') { + body.push('\n'); + } + body.push('\n'); + body.push_str(&tls_text); + } + body +} diff --git a/src/metrics.rs b/src/metrics/render.rs similarity index 73% rename from src/metrics.rs rename to src/metrics/render.rs index fb35b84..7015502 100644 --- a/src/metrics.rs +++ b/src/metrics/render.rs @@ -1,432 +1,19 @@ use std::collections::{BTreeSet, HashMap}; -use std::convert::Infallible; -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::Duration; - -use arc_swap::ArcSwap; -use http_body_util::Full; -use hyper::body::Bytes; -use hyper::server::conn::http1; -use hyper::service::service_fn; -use hyper::{Request, Response, StatusCode}; -use tokio::net::TcpListener; -use tokio::sync::Semaphore; -use tokio::time::timeout; -use tracing::{debug, info, warn}; use crate::config::ProxyConfig; use crate::ip_tracker::UserIpTracker; -use crate::maestro::generation::RuntimeGeneration; use crate::proxy::shared_state::ProxySharedState; use crate::stats::Stats; -use crate::stats::beobachten::BeobachtenStore; use crate::tls_front::TlsFrontCache; use crate::tls_front::cache; use crate::tls_front::fetcher; -use crate::transport::{ListenOptions, create_listener}; + +use super::tls_front::render_tls_front_profile_health; // Keeps `/metrics` response size bounded when per-user telemetry is enabled. const USER_LABELED_METRICS_MAX_USERS: usize = 4096; -// Keeps TLS-front per-domain health series bounded for large generated configs. -const TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS: usize = 256; -const METRICS_MAX_CONTROL_CONNECTIONS: usize = 512; -const METRICS_HTTP_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); -pub async fn serve( - port: u16, - listen: Option, - listen_backlog: u32, - active_runtime: Arc>, -) { - // If `metrics_listen` is set, bind on that single address only. - if let Some(ref listen_addr) = listen { - let addr: SocketAddr = match listen_addr.parse() { - Ok(a) => a, - Err(e) => { - warn!(error = %e, "Invalid metrics_listen address: {}", listen_addr); - return; - } - }; - // Match `server.api.listen`: `[::]:port` is a dual-stack wildcard - // on Linux when `net.ipv6.bindv6only=0`. - let ipv6_only = addr.is_ipv6() && !addr.ip().is_unspecified(); - match bind_metrics_listener(addr, ipv6_only, listen_backlog) { - Ok(listener) => { - info!("Metrics endpoint: http://{}/metrics and /beobachten", addr); - serve_listener(listener, active_runtime).await; - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr); - } - } - return; - } - - // Fallback: keep metrics local unless an explicit metrics_listen is configured. - let mut listener_v4 = None; - let mut listener_v6 = None; - - let addr_v4 = SocketAddr::from(([127, 0, 0, 1], port)); - match bind_metrics_listener(addr_v4, false, listen_backlog) { - Ok(listener) => { - info!( - "Metrics endpoint: http://{}/metrics and /beobachten", - addr_v4 - ); - listener_v4 = Some(listener); - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr_v4); - } - } - - let addr_v6 = SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port)); - match bind_metrics_listener(addr_v6, true, listen_backlog) { - Ok(listener) => { - info!( - "Metrics endpoint: http://[::1]:{}/metrics and /beobachten", - port - ); - listener_v6 = Some(listener); - } - Err(e) => { - warn!(error = %e, "Failed to bind metrics on {}", addr_v6); - } - } - - match (listener_v4, listener_v6) { - (None, None) => { - warn!("Metrics listener is unavailable on both IPv4 and IPv6"); - } - (Some(listener), None) | (None, Some(listener)) => { - serve_listener(listener, active_runtime).await; - } - (Some(listener4), Some(listener6)) => { - let active_runtime_v6 = active_runtime.clone(); - tokio::spawn(async move { - serve_listener(listener6, active_runtime_v6).await; - }); - serve_listener(listener4, active_runtime).await; - } - } -} - -fn bind_metrics_listener( - addr: SocketAddr, - ipv6_only: bool, - listen_backlog: u32, -) -> std::io::Result { - let options = ListenOptions { - reuse_port: false, - ipv6_only, - backlog: listen_backlog, - ..Default::default() - }; - let socket = create_listener(addr, &options)?; - TcpListener::from_std(socket.into()) -} - -async fn serve_listener(listener: TcpListener, active_runtime: Arc>) { - let connection_permits = Arc::new(Semaphore::new(METRICS_MAX_CONTROL_CONNECTIONS)); - - loop { - let (stream, peer) = match listener.accept().await { - Ok(v) => v, - Err(e) => { - warn!(error = %e, "Metrics accept error"); - continue; - } - }; - - let runtime = active_runtime.load_full(); - let config = runtime.config(); - if !config.server.metrics_whitelist.is_empty() - && !config - .server - .metrics_whitelist - .iter() - .any(|net| net.contains(peer.ip())) - { - debug!(peer = %peer, "Metrics request denied by whitelist"); - continue; - } - - let connection_permit = match connection_permits.clone().try_acquire_owned() { - Ok(permit) => permit, - Err(_) => { - debug!( - peer = %peer, - max_connections = METRICS_MAX_CONTROL_CONNECTIONS, - "Dropping metrics connection: control-plane connection budget exhausted" - ); - continue; - } - }; - - let active_runtime = active_runtime.clone(); - tokio::spawn(async move { - let _connection_permit = connection_permit; - let svc = service_fn(move |req| { - let runtime = active_runtime.load_full(); - let stats = runtime.stats.clone(); - let beobachten = runtime.beobachten.clone(); - let shared_state = runtime.proxy_shared.clone(); - let ip_tracker = runtime.ip_tracker.clone(); - let tls_cache = runtime.tls_cache.clone(); - let config = runtime.config(); - async move { - handle( - req, - &stats, - &beobachten, - &shared_state, - &ip_tracker, - tls_cache.as_deref(), - &config, - ) - .await - } - }); - match timeout( - METRICS_HTTP_CONNECTION_TIMEOUT, - http1::Builder::new().serve_connection(hyper_util::rt::TokioIo::new(stream), svc), - ) - .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - debug!(error = %e, "Metrics connection error"); - } - Err(_) => { - debug!( - peer = %peer, - timeout_ms = METRICS_HTTP_CONNECTION_TIMEOUT.as_millis() as u64, - "Metrics connection timed out" - ); - } - } - }); - } -} - -async fn handle( - req: Request, - stats: &Stats, - beobachten: &BeobachtenStore, - shared_state: &ProxySharedState, - ip_tracker: &UserIpTracker, - tls_cache: Option<&TlsFrontCache>, - config: &ProxyConfig, -) -> Result>, Infallible> { - if req.uri().path() == "/metrics" { - let body = render_metrics(stats, shared_state, config, ip_tracker, tls_cache).await; - let resp = Response::builder() - .status(StatusCode::OK) - .header("content-type", "text/plain; version=0.0.4; charset=utf-8") - .body(Full::new(Bytes::from(body))) - .unwrap(); - return Ok(resp); - } - - if req.uri().path() == "/beobachten" { - let body = render_beobachten(stats, beobachten, config); - let resp = Response::builder() - .status(StatusCode::OK) - .header("content-type", "text/plain; charset=utf-8") - .body(Full::new(Bytes::from(body))) - .unwrap(); - return Ok(resp); - } - - let resp = Response::builder() - .status(StatusCode::NOT_FOUND) - .body(Full::new(Bytes::from("Not Found\n"))) - .unwrap(); - Ok(resp) -} - -fn render_beobachten(stats: &Stats, beobachten: &BeobachtenStore, config: &ProxyConfig) -> String { - if !config.general.beobachten { - return "beobachten disabled\n".to_string(); - } - - let ttl = Duration::from_secs(config.general.beobachten_minutes.saturating_mul(60)); - let mut body = beobachten.snapshot_text(ttl); - let tls_text = stats.tls_fingerprint_snapshot_text(ttl, 20); - if !tls_text.is_empty() { - if !body.ends_with('\n') { - body.push('\n'); - } - body.push('\n'); - body.push_str(&tls_text); - } - body -} - -fn tls_front_domains(config: &ProxyConfig) -> Vec { - let mut domains = Vec::with_capacity(1 + config.censorship.tls_domains.len()); - if !config.censorship.tls_domain.is_empty() { - domains.push(config.censorship.tls_domain.clone()); - } - for domain in &config.censorship.tls_domains { - if !domain.is_empty() && !domains.contains(domain) { - domains.push(domain.clone()); - } - } - domains -} - -fn prometheus_label_value(value: &str) -> String { - value.replace('\\', "\\\\").replace('"', "\\\"") -} - -async fn render_tls_front_profile_health( - out: &mut String, - config: &ProxyConfig, - tls_cache: Option<&TlsFrontCache>, -) { - use std::fmt::Write; - - let domains = tls_front_domains(config); - let (health, suppressed) = match (config.censorship.tls_emulation, tls_cache) { - (true, Some(cache)) => { - cache - .profile_health_snapshot(&domains, TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS) - .await - } - _ => (Vec::new(), domains.len()), - }; - - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_domains TLS front configured profile domains by export status" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_domains gauge"); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"configured\"}} {}", - domains.len() - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"emitted\"}} {}", - health.len() - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_domains{{status=\"suppressed\"}} {}", - suppressed - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_info TLS front profile source and feature flags per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_info gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_quality_info TLS front profile quality and key-share group per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_quality_info gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_age_seconds Age of cached TLS front profile data per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_age_seconds gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_server_hello_bytes TLS front cached ServerHello record body bytes per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_server_hello_bytes gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_server_hello_extensions TLS front cached visible ServerHello extension count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_server_hello_extensions gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_app_data_records TLS front cached app-data record count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_app_data_records gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_ticket_records TLS front cached ticket-like tail record count per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_ticket_records gauge"); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_change_cipher_spec_records TLS front cached ChangeCipherSpec record count per configured domain" - ); - let _ = writeln!( - out, - "# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge" - ); - let _ = writeln!( - out, - "# HELP telemt_tls_front_profile_app_data_bytes TLS front cached total app-data bytes per configured domain" - ); - let _ = writeln!(out, "# TYPE telemt_tls_front_profile_app_data_bytes gauge"); - - for item in health { - let domain = prometheus_label_value(&item.domain); - let _ = writeln!( - out, - "telemt_tls_front_profile_info{{domain=\"{}\",source=\"{}\",is_default=\"{}\",has_cert_info=\"{}\",has_cert_payload=\"{}\"}} 1", - domain, item.source, item.is_default, item.has_cert_info, item.has_cert_payload - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_quality_info{{domain=\"{}\",quality=\"{}\",key_share_group=\"{}\"}} 1", - domain, item.quality, item.key_share_group - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_age_seconds{{domain=\"{}\"}} {}", - domain, item.age_seconds - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_server_hello_bytes{{domain=\"{}\"}} {}", - domain, item.server_hello_record_len - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_server_hello_extensions{{domain=\"{}\"}} {}", - domain, item.server_hello_extensions - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_app_data_records{{domain=\"{}\"}} {}", - domain, item.app_data_records - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_ticket_records{{domain=\"{}\"}} {}", - domain, item.ticket_records - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_change_cipher_spec_records{{domain=\"{}\"}} {}", - domain, item.change_cipher_spec_count - ); - let _ = writeln!( - out, - "telemt_tls_front_profile_app_data_bytes{{domain=\"{}\"}} {}", - domain, item.total_app_data_len - ); - } -} - -async fn render_metrics( +pub(super) async fn render_metrics( stats: &Stats, shared_state: &ProxySharedState, config: &ProxyConfig, @@ -3858,436 +3445,3 @@ async fn render_metrics( out } - -#[cfg(test)] -mod tests { - use super::*; - use http_body_util::BodyExt; - use std::net::IpAddr; - use std::time::SystemTime; - - use crate::tls_front::types::{ - CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource, - }; - - #[tokio::test] - async fn test_render_metrics_format() { - let stats = Arc::new(Stats::new()); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - config - .access - .user_max_unique_ips - .insert("alice".to_string(), 4); - - stats.increment_connects_all(); - stats.increment_connects_all(); - stats.increment_connects_bad_with_class("tls_handshake_bad_client"); - stats.increment_handshake_timeouts(); - stats.increment_handshake_failure_class("timeout"); - shared_state - .handshake - .auth_expensive_checks_total - .fetch_add(9, std::sync::atomic::Ordering::Relaxed); - shared_state - .handshake - .auth_budget_exhausted_total - .fetch_add(2, std::sync::atomic::Ordering::Relaxed); - stats.increment_upstream_connect_attempt_total(); - stats.increment_upstream_connect_attempt_total(); - stats.increment_upstream_connect_success_total(); - stats.increment_upstream_connect_fail_total(); - stats.increment_upstream_connect_failfast_hard_error_total(); - stats.observe_upstream_connect_attempts_per_request(2); - stats.observe_upstream_connect_duration_ms(220, true); - stats.observe_upstream_connect_duration_ms(1500, false); - stats.increment_me_rpc_proxy_req_signal_sent_total(); - stats.increment_me_rpc_proxy_req_signal_failed_total(); - stats.increment_me_rpc_proxy_req_signal_skipped_no_meta_total(); - stats.increment_me_rpc_proxy_req_signal_response_total(); - stats.increment_me_rpc_proxy_req_signal_close_sent_total(); - stats.increment_me_idle_close_by_peer_total(); - stats.increment_relay_idle_soft_mark_total(); - stats.increment_relay_idle_hard_close_total(); - stats.increment_relay_pressure_evict_total(); - stats.increment_relay_protocol_desync_close_total(); - stats.increment_me_d2c_batches_total(); - stats.add_me_d2c_batch_frames_total(3); - stats.add_me_d2c_batch_bytes_total(2048); - stats.increment_me_d2c_flush_reason(crate::stats::MeD2cFlushReason::AckImmediate); - stats.increment_me_d2c_data_frames_total(); - stats.increment_me_d2c_ack_frames_total(); - stats.add_me_d2c_payload_bytes_total(1800); - stats.increment_me_d2c_write_mode(crate::stats::MeD2cWriteMode::Coalesced); - stats.increment_me_d2c_quota_reject_total(crate::stats::MeD2cQuotaRejectStage::PostWrite); - stats.observe_me_d2c_frame_buf_shrink(4096); - stats.increment_me_endpoint_quarantine_total(); - stats.increment_me_endpoint_quarantine_unexpected_total(); - stats.increment_me_endpoint_quarantine_draining_suppressed_total(); - stats.increment_user_connects("alice"); - stats.increment_user_curr_connects("alice"); - stats.add_user_octets_from("alice", 1024); - stats.add_user_octets_to("alice", 2048); - stats.increment_user_msgs_from("alice"); - stats.increment_user_msgs_to("alice"); - stats.increment_user_msgs_to("alice"); - tracker - .check_and_add("alice", "203.0.113.10".parse().unwrap()) - .await - .unwrap(); - - let output = render_metrics(&stats, shared_state.as_ref(), &config, &tracker, None).await; - - assert!(output.contains(&format!( - "telemt_build_info{{version=\"{}\"}} 1", - env!("CARGO_PKG_VERSION") - ))); - assert!(output.contains("telemt_connections_total 2")); - assert!(output.contains("telemt_connections_bad_total 1")); - assert!(output.contains( - "telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1" - )); - assert!(output.contains("telemt_handshake_timeouts_total 1")); - assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); - assert!(output.contains("telemt_auth_expensive_checks_total 9")); - assert!(output.contains("telemt_auth_budget_exhausted_total 2")); - assert!(output.contains("telemt_upstream_connect_attempt_total 2")); - assert!(output.contains("telemt_upstream_connect_success_total 1")); - assert!(output.contains("telemt_upstream_connect_fail_total 1")); - assert!(output.contains("telemt_upstream_connect_failfast_hard_error_total 1")); - assert!(output.contains("telemt_upstream_connect_attempts_per_request{bucket=\"2\"} 1")); - assert!( - output - .contains("telemt_upstream_connect_duration_success_total{bucket=\"101_500ms\"} 1") - ); - assert!( - output.contains("telemt_upstream_connect_duration_fail_total{bucket=\"gt_1000ms\"} 1") - ); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_sent_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_failed_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_skipped_no_meta_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_response_total 1")); - assert!(output.contains("telemt_me_rpc_proxy_req_signal_close_sent_total 1")); - assert!(output.contains("telemt_me_idle_close_by_peer_total 1")); - assert!(output.contains("telemt_relay_idle_soft_mark_total 1")); - assert!(output.contains("telemt_relay_idle_hard_close_total 1")); - assert!(output.contains("telemt_relay_pressure_evict_total 1")); - assert!(output.contains("telemt_relay_protocol_desync_close_total 1")); - assert!(output.contains("telemt_me_d2c_batches_total 1")); - assert!(output.contains("telemt_me_d2c_batch_frames_total 3")); - assert!(output.contains("telemt_me_d2c_batch_bytes_total 2048")); - assert!(output.contains("telemt_me_d2c_flush_reason_total{reason=\"ack_immediate\"} 1")); - assert!(output.contains("telemt_me_d2c_data_frames_total 1")); - assert!(output.contains("telemt_me_d2c_ack_frames_total 1")); - assert!(output.contains("telemt_me_d2c_payload_bytes_total 1800")); - assert!(output.contains("telemt_me_d2c_write_mode_total{mode=\"coalesced\"} 1")); - assert!(output.contains("telemt_me_d2c_quota_reject_total{stage=\"post_write\"} 1")); - assert!(output.contains("telemt_me_d2c_frame_buf_shrink_total 1")); - assert!(output.contains("telemt_me_d2c_frame_buf_shrink_bytes_total 4096")); - assert!(output.contains("telemt_me_endpoint_quarantine_total 1")); - assert!(output.contains("telemt_me_endpoint_quarantine_unexpected_total 1")); - assert!(output.contains("telemt_me_endpoint_quarantine_draining_suppressed_total 1")); - assert!(output.contains("telemt_user_connections_total{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_connections_current{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_octets_from_client{user=\"alice\"} 1024")); - assert!(output.contains("telemt_user_octets_to_client{user=\"alice\"} 2048")); - assert!(output.contains("telemt_user_msgs_from_client{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_msgs_to_client{user=\"alice\"} 2")); - assert!(output.contains("telemt_user_unique_ips_current{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_unique_ips_recent_window{user=\"alice\"} 1")); - assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 4")); - assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.250000")); - assert!(output.contains("telemt_ip_tracker_users{scope=\"active\"} 1")); - assert!(output.contains("telemt_ip_tracker_entries{scope=\"active\"} 1")); - assert!(output.contains("telemt_ip_tracker_cleanup_queue_len 0")); - } - - #[tokio::test] - async fn test_render_tls_front_profile_health() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - config.censorship.tls_domain = "primary.example".to_string(); - config.censorship.tls_domains = vec!["fallback.example".to_string()]; - - let cache = TlsFrontCache::new( - &[ - "primary.example".to_string(), - "fallback.example".to_string(), - ], - 1024, - "tlsfront-profile-health-test", - ); - cache - .set( - "primary.example", - CachedTlsData { - server_hello_template: ParsedServerHello { - version: [0x03, 0x03], - random: [0u8; 32], - session_id: Vec::new(), - cipher_suite: [0x13, 0x01], - compression: 0, - extensions: { - let mut key_share = vec![0x00, 0x1d, 0x00, 0x20]; - key_share.resize(36, 0x42); - vec![ - crate::tls_front::types::TlsExtension { - ext_type: 0x002b, - data: vec![0x03, 0x04], - }, - crate::tls_front::types::TlsExtension { - ext_type: 0x0033, - data: key_share, - }, - ] - }, - }, - cert_info: None, - cert_payload: Some(TlsCertPayload { - cert_chain_der: vec![vec![0x30, 0x01]], - certificate_message: vec![0x0b, 0x00, 0x00, 0x00], - }), - app_data_records_sizes: vec![1024, 512], - total_app_data_len: 1536, - behavior_profile: TlsBehaviorProfile { - change_cipher_spec_count: 1, - app_data_record_sizes: vec![1024, 512], - ticket_record_sizes: vec![69], - source: TlsProfileSource::Merged, - ..TlsBehaviorProfile::default() - }, - fetched_at: SystemTime::now(), - domain: "primary.example".to_string(), - }, - ) - .await; - - let output = render_metrics(&stats, &shared_state, &config, &tracker, Some(&cache)).await; - - assert!(output.contains("telemt_tls_front_profile_domains{status=\"configured\"} 2")); - assert!(output.contains("telemt_tls_front_profile_domains{status=\"emitted\"} 2")); - assert!(output.contains("telemt_tls_front_profile_domains{status=\"suppressed\"} 0")); - assert!( - output.contains("telemt_tls_front_profile_info{domain=\"primary.example\",source=\"merged\",is_default=\"false\",has_cert_info=\"false\",has_cert_payload=\"true\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_info{domain=\"fallback.example\",source=\"default\",is_default=\"true\",has_cert_info=\"false\",has_cert_payload=\"false\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_quality_info{domain=\"primary.example\",quality=\"raw_strict\",key_share_group=\"x25519\"} 1") - ); - assert!( - output.contains("telemt_tls_front_profile_quality_info{domain=\"fallback.example\",quality=\"fallback\",key_share_group=\"none\"} 1") - ); - assert!(output.contains( - "telemt_tls_front_profile_server_hello_bytes{domain=\"primary.example\"} 90" - )); - assert!(output.contains( - "telemt_tls_front_profile_server_hello_extensions{domain=\"primary.example\"} 2" - )); - assert!( - output.contains( - "telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2" - ) - ); - assert!( - output - .contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1") - ); - assert!(output.contains( - "telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1" - )); - assert!( - output.contains( - "telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536" - ) - ); - } - - #[tokio::test] - async fn test_render_empty_stats() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - assert!(output.contains("telemt_connections_total 0")); - assert!(output.contains("telemt_connections_bad_total 0")); - assert!(output.contains("telemt_handshake_timeouts_total 0")); - assert!(output.contains("telemt_auth_expensive_checks_total 0")); - assert!(output.contains("telemt_auth_budget_exhausted_total 0")); - assert!(output.contains("telemt_user_unique_ips_current{user=")); - assert!(output.contains("telemt_user_unique_ips_recent_window{user=")); - } - - #[tokio::test] - async fn test_render_uses_global_each_unique_ip_limit() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - stats.increment_user_connects("alice"); - stats.increment_user_curr_connects("alice"); - let tracker = UserIpTracker::new(); - tracker - .check_and_add("alice", "203.0.113.10".parse().unwrap()) - .await - .unwrap(); - let mut config = ProxyConfig::default(); - config.access.user_max_unique_ips_global_each = 2; - - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - - assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 2")); - assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.500000")); - } - - #[tokio::test] - async fn test_render_has_type_annotations() { - let stats = Stats::new(); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let config = ProxyConfig::default(); - let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; - assert!(output.contains("# TYPE telemt_uptime_seconds gauge")); - assert!(output.contains("# TYPE telemt_connections_total counter")); - assert!(output.contains("# TYPE telemt_connections_bad_total counter")); - assert!(output.contains("# TYPE telemt_connections_bad_by_class_total counter")); - assert!(output.contains("# TYPE telemt_handshake_timeouts_total counter")); - assert!(output.contains("# TYPE telemt_handshake_failures_by_class_total counter")); - assert!(output.contains("# TYPE telemt_auth_expensive_checks_total counter")); - assert!(output.contains("# TYPE telemt_auth_budget_exhausted_total counter")); - assert!(output.contains("# TYPE telemt_upstream_connect_attempt_total counter")); - assert!(output.contains("# TYPE telemt_me_rpc_proxy_req_signal_sent_total counter")); - assert!(output.contains("# TYPE telemt_me_idle_close_by_peer_total counter")); - assert!(output.contains("# TYPE telemt_relay_idle_soft_mark_total counter")); - assert!(output.contains("# TYPE telemt_relay_idle_hard_close_total counter")); - assert!(output.contains("# TYPE telemt_relay_pressure_evict_total counter")); - assert!(output.contains("# TYPE telemt_relay_protocol_desync_close_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_batches_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_flush_reason_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_write_mode_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_batch_frames_bucket_total counter")); - assert!(output.contains("# TYPE telemt_me_d2c_flush_duration_us_bucket_total counter")); - assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_total counter")); - assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_unexpected_total counter")); - assert!( - output - .contains("# TYPE telemt_me_endpoint_quarantine_draining_suppressed_total counter") - ); - assert!(output.contains("# TYPE telemt_me_writer_removed_total counter")); - assert!( - output - .contains("# TYPE telemt_me_writer_removed_unexpected_minus_restored_total gauge") - ); - assert!(output.contains("# TYPE telemt_user_unique_ips_current gauge")); - assert!(output.contains("# TYPE telemt_user_unique_ips_recent_window gauge")); - assert!(output.contains("# TYPE telemt_user_unique_ips_limit gauge")); - assert!(output.contains("# TYPE telemt_user_unique_ips_utilization gauge")); - assert!(output.contains("# TYPE telemt_stats_user_entries gauge")); - assert!(output.contains("# TYPE telemt_telemetry_user_series_users gauge")); - assert!(output.contains("# TYPE telemt_ip_tracker_users gauge")); - assert!(output.contains("# TYPE telemt_ip_tracker_entries gauge")); - assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_queue_len gauge")); - assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_total counter")); - assert!(output.contains("# TYPE telemt_ip_tracker_cap_rejects_total counter")); - assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_entries gauge")); - assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_cap_drops_total counter")); - assert!(output.contains("# TYPE telemt_tls_front_full_cert_budget_ips gauge")); - assert!( - output.contains("# TYPE telemt_tls_front_full_cert_budget_cap_drops_total counter") - ); - assert!(output.contains("# TYPE telemt_tls_front_profile_domains gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_info gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_quality_info gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_age_seconds gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_bytes gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_extensions gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_records gauge")); - assert!(output.contains("# TYPE telemt_tls_front_profile_ticket_records gauge")); - assert!( - output.contains("# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge") - ); - assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_bytes gauge")); - } - - #[tokio::test] - async fn test_endpoint_integration() { - let stats = Arc::new(Stats::new()); - let beobachten = Arc::new(BeobachtenStore::new()); - let shared_state = ProxySharedState::new(); - let tracker = UserIpTracker::new(); - let mut config = ProxyConfig::default(); - stats.increment_connects_all(); - stats.increment_connects_all(); - stats.increment_connects_all(); - - let req = Request::builder().uri("/metrics").body(()).unwrap(); - let resp = handle( - req, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = resp.into_body().collect().await.unwrap().to_bytes(); - assert!( - std::str::from_utf8(body.as_ref()) - .unwrap() - .contains("telemt_connections_total 3") - ); - assert!( - std::str::from_utf8(body.as_ref()) - .unwrap() - .contains(&format!( - "telemt_build_info{{version=\"{}\"}} 1", - env!("CARGO_PKG_VERSION") - )) - ); - - config.general.beobachten = true; - config.general.beobachten_minutes = 10; - beobachten.record( - "TLS-scanner", - "203.0.113.10".parse::().unwrap(), - Duration::from_secs(600), - ); - let req_beob = Request::builder().uri("/beobachten").body(()).unwrap(); - let resp_beob = handle( - req_beob, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp_beob.status(), StatusCode::OK); - let body_beob = resp_beob.into_body().collect().await.unwrap().to_bytes(); - let beob_text = std::str::from_utf8(body_beob.as_ref()).unwrap(); - assert!(beob_text.contains("[TLS-scanner]")); - assert!(beob_text.contains("203.0.113.10-1")); - - let req404 = Request::builder().uri("/other").body(()).unwrap(); - let resp404 = handle( - req404, - &stats, - &beobachten, - shared_state.as_ref(), - &tracker, - None, - &config, - ) - .await - .unwrap(); - assert_eq!(resp404.status(), StatusCode::NOT_FOUND); - } -} diff --git a/src/metrics/tests.rs b/src/metrics/tests.rs new file mode 100644 index 0000000..8db5fdf --- /dev/null +++ b/src/metrics/tests.rs @@ -0,0 +1,417 @@ +use super::render::render_metrics; +use super::*; +use http_body_util::BodyExt; +use std::net::IpAddr; +use std::time::SystemTime; + +use crate::tls_front::types::{ + CachedTlsData, ParsedServerHello, TlsBehaviorProfile, TlsCertPayload, TlsProfileSource, +}; + +#[tokio::test] +async fn test_render_metrics_format() { + let stats = Arc::new(Stats::new()); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + config + .access + .user_max_unique_ips + .insert("alice".to_string(), 4); + + stats.increment_connects_all(); + stats.increment_connects_all(); + stats.increment_connects_bad_with_class("tls_handshake_bad_client"); + stats.increment_handshake_timeouts(); + stats.increment_handshake_failure_class("timeout"); + shared_state + .handshake + .auth_expensive_checks_total + .fetch_add(9, std::sync::atomic::Ordering::Relaxed); + shared_state + .handshake + .auth_budget_exhausted_total + .fetch_add(2, std::sync::atomic::Ordering::Relaxed); + stats.increment_upstream_connect_attempt_total(); + stats.increment_upstream_connect_attempt_total(); + stats.increment_upstream_connect_success_total(); + stats.increment_upstream_connect_fail_total(); + stats.increment_upstream_connect_failfast_hard_error_total(); + stats.observe_upstream_connect_attempts_per_request(2); + stats.observe_upstream_connect_duration_ms(220, true); + stats.observe_upstream_connect_duration_ms(1500, false); + stats.increment_me_rpc_proxy_req_signal_sent_total(); + stats.increment_me_rpc_proxy_req_signal_failed_total(); + stats.increment_me_rpc_proxy_req_signal_skipped_no_meta_total(); + stats.increment_me_rpc_proxy_req_signal_response_total(); + stats.increment_me_rpc_proxy_req_signal_close_sent_total(); + stats.increment_me_idle_close_by_peer_total(); + stats.increment_relay_idle_soft_mark_total(); + stats.increment_relay_idle_hard_close_total(); + stats.increment_relay_pressure_evict_total(); + stats.increment_relay_protocol_desync_close_total(); + stats.increment_me_d2c_batches_total(); + stats.add_me_d2c_batch_frames_total(3); + stats.add_me_d2c_batch_bytes_total(2048); + stats.increment_me_d2c_flush_reason(crate::stats::MeD2cFlushReason::AckImmediate); + stats.increment_me_d2c_data_frames_total(); + stats.increment_me_d2c_ack_frames_total(); + stats.add_me_d2c_payload_bytes_total(1800); + stats.increment_me_d2c_write_mode(crate::stats::MeD2cWriteMode::Coalesced); + stats.increment_me_d2c_quota_reject_total(crate::stats::MeD2cQuotaRejectStage::PostWrite); + stats.observe_me_d2c_frame_buf_shrink(4096); + stats.increment_me_endpoint_quarantine_total(); + stats.increment_me_endpoint_quarantine_unexpected_total(); + stats.increment_me_endpoint_quarantine_draining_suppressed_total(); + stats.increment_user_connects("alice"); + stats.increment_user_curr_connects("alice"); + stats.add_user_octets_from("alice", 1024); + stats.add_user_octets_to("alice", 2048); + stats.increment_user_msgs_from("alice"); + stats.increment_user_msgs_to("alice"); + stats.increment_user_msgs_to("alice"); + tracker + .check_and_add("alice", "203.0.113.10".parse().unwrap()) + .await + .unwrap(); + + let output = render_metrics(&stats, shared_state.as_ref(), &config, &tracker, None).await; + + assert!(output.contains(&format!( + "telemt_build_info{{version=\"{}\"}} 1", + env!("CARGO_PKG_VERSION") + ))); + assert!(output.contains("telemt_connections_total 2")); + assert!(output.contains("telemt_connections_bad_total 1")); + assert!( + output.contains("telemt_connections_bad_by_class_total{class=\"tls_handshake_bad_client\"} 1") + ); + assert!(output.contains("telemt_handshake_timeouts_total 1")); + assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); + assert!(output.contains("telemt_auth_expensive_checks_total 9")); + assert!(output.contains("telemt_auth_budget_exhausted_total 2")); + assert!(output.contains("telemt_upstream_connect_attempt_total 2")); + assert!(output.contains("telemt_upstream_connect_success_total 1")); + assert!(output.contains("telemt_upstream_connect_fail_total 1")); + assert!(output.contains("telemt_upstream_connect_failfast_hard_error_total 1")); + assert!(output.contains("telemt_upstream_connect_attempts_per_request{bucket=\"2\"} 1")); + assert!( + output.contains("telemt_upstream_connect_duration_success_total{bucket=\"101_500ms\"} 1") + ); + assert!(output.contains("telemt_upstream_connect_duration_fail_total{bucket=\"gt_1000ms\"} 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_sent_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_failed_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_skipped_no_meta_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_response_total 1")); + assert!(output.contains("telemt_me_rpc_proxy_req_signal_close_sent_total 1")); + assert!(output.contains("telemt_me_idle_close_by_peer_total 1")); + assert!(output.contains("telemt_relay_idle_soft_mark_total 1")); + assert!(output.contains("telemt_relay_idle_hard_close_total 1")); + assert!(output.contains("telemt_relay_pressure_evict_total 1")); + assert!(output.contains("telemt_relay_protocol_desync_close_total 1")); + assert!(output.contains("telemt_me_d2c_batches_total 1")); + assert!(output.contains("telemt_me_d2c_batch_frames_total 3")); + assert!(output.contains("telemt_me_d2c_batch_bytes_total 2048")); + assert!(output.contains("telemt_me_d2c_flush_reason_total{reason=\"ack_immediate\"} 1")); + assert!(output.contains("telemt_me_d2c_data_frames_total 1")); + assert!(output.contains("telemt_me_d2c_ack_frames_total 1")); + assert!(output.contains("telemt_me_d2c_payload_bytes_total 1800")); + assert!(output.contains("telemt_me_d2c_write_mode_total{mode=\"coalesced\"} 1")); + assert!(output.contains("telemt_me_d2c_quota_reject_total{stage=\"post_write\"} 1")); + assert!(output.contains("telemt_me_d2c_frame_buf_shrink_total 1")); + assert!(output.contains("telemt_me_d2c_frame_buf_shrink_bytes_total 4096")); + assert!(output.contains("telemt_me_endpoint_quarantine_total 1")); + assert!(output.contains("telemt_me_endpoint_quarantine_unexpected_total 1")); + assert!(output.contains("telemt_me_endpoint_quarantine_draining_suppressed_total 1")); + assert!(output.contains("telemt_user_connections_total{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_connections_current{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_octets_from_client{user=\"alice\"} 1024")); + assert!(output.contains("telemt_user_octets_to_client{user=\"alice\"} 2048")); + assert!(output.contains("telemt_user_msgs_from_client{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_msgs_to_client{user=\"alice\"} 2")); + assert!(output.contains("telemt_user_unique_ips_current{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_unique_ips_recent_window{user=\"alice\"} 1")); + assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 4")); + assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.250000")); + assert!(output.contains("telemt_ip_tracker_users{scope=\"active\"} 1")); + assert!(output.contains("telemt_ip_tracker_entries{scope=\"active\"} 1")); + assert!(output.contains("telemt_ip_tracker_cleanup_queue_len 0")); +} + +#[tokio::test] +async fn test_render_tls_front_profile_health() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + config.censorship.tls_domain = "primary.example".to_string(); + config.censorship.tls_domains = vec!["fallback.example".to_string()]; + + let cache = TlsFrontCache::new( + &[ + "primary.example".to_string(), + "fallback.example".to_string(), + ], + 1024, + "tlsfront-profile-health-test", + ); + cache + .set( + "primary.example", + CachedTlsData { + server_hello_template: ParsedServerHello { + version: [0x03, 0x03], + random: [0u8; 32], + session_id: Vec::new(), + cipher_suite: [0x13, 0x01], + compression: 0, + extensions: { + let mut key_share = vec![0x00, 0x1d, 0x00, 0x20]; + key_share.resize(36, 0x42); + vec![ + crate::tls_front::types::TlsExtension { + ext_type: 0x002b, + data: vec![0x03, 0x04], + }, + crate::tls_front::types::TlsExtension { + ext_type: 0x0033, + data: key_share, + }, + ] + }, + }, + cert_info: None, + cert_payload: Some(TlsCertPayload { + cert_chain_der: vec![vec![0x30, 0x01]], + certificate_message: vec![0x0b, 0x00, 0x00, 0x00], + }), + app_data_records_sizes: vec![1024, 512], + total_app_data_len: 1536, + behavior_profile: TlsBehaviorProfile { + change_cipher_spec_count: 1, + app_data_record_sizes: vec![1024, 512], + ticket_record_sizes: vec![69], + source: TlsProfileSource::Merged, + ..TlsBehaviorProfile::default() + }, + fetched_at: SystemTime::now(), + domain: "primary.example".to_string(), + }, + ) + .await; + + let output = render_metrics(&stats, &shared_state, &config, &tracker, Some(&cache)).await; + + assert!(output.contains("telemt_tls_front_profile_domains{status=\"configured\"} 2")); + assert!(output.contains("telemt_tls_front_profile_domains{status=\"emitted\"} 2")); + assert!(output.contains("telemt_tls_front_profile_domains{status=\"suppressed\"} 0")); + assert!( + output.contains("telemt_tls_front_profile_info{domain=\"primary.example\",source=\"merged\",is_default=\"false\",has_cert_info=\"false\",has_cert_payload=\"true\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_info{domain=\"fallback.example\",source=\"default\",is_default=\"true\",has_cert_info=\"false\",has_cert_payload=\"false\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_quality_info{domain=\"primary.example\",quality=\"raw_strict\",key_share_group=\"x25519\"} 1") + ); + assert!( + output.contains("telemt_tls_front_profile_quality_info{domain=\"fallback.example\",quality=\"fallback\",key_share_group=\"none\"} 1") + ); + assert!( + output + .contains("telemt_tls_front_profile_server_hello_bytes{domain=\"primary.example\"} 90") + ); + assert!(output.contains( + "telemt_tls_front_profile_server_hello_extensions{domain=\"primary.example\"} 2" + )); + assert!( + output.contains("telemt_tls_front_profile_app_data_records{domain=\"primary.example\"} 2") + ); + assert!( + output.contains("telemt_tls_front_profile_ticket_records{domain=\"primary.example\"} 1") + ); + assert!(output.contains( + "telemt_tls_front_profile_change_cipher_spec_records{domain=\"primary.example\"} 1" + )); + assert!( + output.contains("telemt_tls_front_profile_app_data_bytes{domain=\"primary.example\"} 1536") + ); +} + +#[tokio::test] +async fn test_render_empty_stats() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let config = ProxyConfig::default(); + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + assert!(output.contains("telemt_connections_total 0")); + assert!(output.contains("telemt_connections_bad_total 0")); + assert!(output.contains("telemt_handshake_timeouts_total 0")); + assert!(output.contains("telemt_auth_expensive_checks_total 0")); + assert!(output.contains("telemt_auth_budget_exhausted_total 0")); + assert!(output.contains("telemt_user_unique_ips_current{user=")); + assert!(output.contains("telemt_user_unique_ips_recent_window{user=")); +} + +#[tokio::test] +async fn test_render_uses_global_each_unique_ip_limit() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + stats.increment_user_connects("alice"); + stats.increment_user_curr_connects("alice"); + let tracker = UserIpTracker::new(); + tracker + .check_and_add("alice", "203.0.113.10".parse().unwrap()) + .await + .unwrap(); + let mut config = ProxyConfig::default(); + config.access.user_max_unique_ips_global_each = 2; + + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + + assert!(output.contains("telemt_user_unique_ips_limit{user=\"alice\"} 2")); + assert!(output.contains("telemt_user_unique_ips_utilization{user=\"alice\"} 0.500000")); +} + +#[tokio::test] +async fn test_render_has_type_annotations() { + let stats = Stats::new(); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let config = ProxyConfig::default(); + let output = render_metrics(&stats, &shared_state, &config, &tracker, None).await; + assert!(output.contains("# TYPE telemt_uptime_seconds gauge")); + assert!(output.contains("# TYPE telemt_connections_total counter")); + assert!(output.contains("# TYPE telemt_connections_bad_total counter")); + assert!(output.contains("# TYPE telemt_connections_bad_by_class_total counter")); + assert!(output.contains("# TYPE telemt_handshake_timeouts_total counter")); + assert!(output.contains("# TYPE telemt_handshake_failures_by_class_total counter")); + assert!(output.contains("# TYPE telemt_auth_expensive_checks_total counter")); + assert!(output.contains("# TYPE telemt_auth_budget_exhausted_total counter")); + assert!(output.contains("# TYPE telemt_upstream_connect_attempt_total counter")); + assert!(output.contains("# TYPE telemt_me_rpc_proxy_req_signal_sent_total counter")); + assert!(output.contains("# TYPE telemt_me_idle_close_by_peer_total counter")); + assert!(output.contains("# TYPE telemt_relay_idle_soft_mark_total counter")); + assert!(output.contains("# TYPE telemt_relay_idle_hard_close_total counter")); + assert!(output.contains("# TYPE telemt_relay_pressure_evict_total counter")); + assert!(output.contains("# TYPE telemt_relay_protocol_desync_close_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_batches_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_flush_reason_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_write_mode_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_batch_frames_bucket_total counter")); + assert!(output.contains("# TYPE telemt_me_d2c_flush_duration_us_bucket_total counter")); + assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_total counter")); + assert!(output.contains("# TYPE telemt_me_endpoint_quarantine_unexpected_total counter")); + assert!( + output.contains("# TYPE telemt_me_endpoint_quarantine_draining_suppressed_total counter") + ); + assert!(output.contains("# TYPE telemt_me_writer_removed_total counter")); + assert!( + output.contains("# TYPE telemt_me_writer_removed_unexpected_minus_restored_total gauge") + ); + assert!(output.contains("# TYPE telemt_user_unique_ips_current gauge")); + assert!(output.contains("# TYPE telemt_user_unique_ips_recent_window gauge")); + assert!(output.contains("# TYPE telemt_user_unique_ips_limit gauge")); + assert!(output.contains("# TYPE telemt_user_unique_ips_utilization gauge")); + assert!(output.contains("# TYPE telemt_stats_user_entries gauge")); + assert!(output.contains("# TYPE telemt_telemetry_user_series_users gauge")); + assert!(output.contains("# TYPE telemt_ip_tracker_users gauge")); + assert!(output.contains("# TYPE telemt_ip_tracker_entries gauge")); + assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_queue_len gauge")); + assert!(output.contains("# TYPE telemt_ip_tracker_cleanup_total counter")); + assert!(output.contains("# TYPE telemt_ip_tracker_cap_rejects_total counter")); + assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_entries gauge")); + assert!(output.contains("# TYPE telemt_tls_fetch_profile_cache_cap_drops_total counter")); + assert!(output.contains("# TYPE telemt_tls_front_full_cert_budget_ips gauge")); + assert!(output.contains("# TYPE telemt_tls_front_full_cert_budget_cap_drops_total counter")); + assert!(output.contains("# TYPE telemt_tls_front_profile_domains gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_info gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_quality_info gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_age_seconds gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_bytes gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_server_hello_extensions gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_ticket_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge")); + assert!(output.contains("# TYPE telemt_tls_front_profile_app_data_bytes gauge")); +} + +#[tokio::test] +async fn test_endpoint_integration() { + let stats = Arc::new(Stats::new()); + let beobachten = Arc::new(BeobachtenStore::new()); + let shared_state = ProxySharedState::new(); + let tracker = UserIpTracker::new(); + let mut config = ProxyConfig::default(); + stats.increment_connects_all(); + stats.increment_connects_all(); + stats.increment_connects_all(); + + let req = Request::builder().uri("/metrics").body(()).unwrap(); + let resp = handle( + req, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let body = resp.into_body().collect().await.unwrap().to_bytes(); + assert!( + std::str::from_utf8(body.as_ref()) + .unwrap() + .contains("telemt_connections_total 3") + ); + assert!( + std::str::from_utf8(body.as_ref()) + .unwrap() + .contains(&format!( + "telemt_build_info{{version=\"{}\"}} 1", + env!("CARGO_PKG_VERSION") + )) + ); + + config.general.beobachten = true; + config.general.beobachten_minutes = 10; + beobachten.record( + "TLS-scanner", + "203.0.113.10".parse::().unwrap(), + Duration::from_secs(600), + ); + let req_beob = Request::builder().uri("/beobachten").body(()).unwrap(); + let resp_beob = handle( + req_beob, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp_beob.status(), StatusCode::OK); + let body_beob = resp_beob.into_body().collect().await.unwrap().to_bytes(); + let beob_text = std::str::from_utf8(body_beob.as_ref()).unwrap(); + assert!(beob_text.contains("[TLS-scanner]")); + assert!(beob_text.contains("203.0.113.10-1")); + + let req404 = Request::builder().uri("/other").body(()).unwrap(); + let resp404 = handle( + req404, + &stats, + &beobachten, + shared_state.as_ref(), + &tracker, + None, + &config, + ) + .await + .unwrap(); + assert_eq!(resp404.status(), StatusCode::NOT_FOUND); +} diff --git a/src/metrics/tls_front.rs b/src/metrics/tls_front.rs new file mode 100644 index 0000000..1a4db5c --- /dev/null +++ b/src/metrics/tls_front.rs @@ -0,0 +1,167 @@ +use crate::config::ProxyConfig; +use crate::tls_front::TlsFrontCache; + +// Keeps TLS-front per-domain health series bounded for large generated configs. +const TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS: usize = 256; + +fn tls_front_domains(config: &ProxyConfig) -> Vec { + let mut domains = Vec::with_capacity(1 + config.censorship.tls_domains.len()); + if !config.censorship.tls_domain.is_empty() { + domains.push(config.censorship.tls_domain.clone()); + } + for domain in &config.censorship.tls_domains { + if !domain.is_empty() && !domains.contains(domain) { + domains.push(domain.clone()); + } + } + domains +} + +fn prometheus_label_value(value: &str) -> String { + value.replace('\\', "\\\\").replace('"', "\\\"") +} + +pub(super) async fn render_tls_front_profile_health( + out: &mut String, + config: &ProxyConfig, + tls_cache: Option<&TlsFrontCache>, +) { + use std::fmt::Write; + + let domains = tls_front_domains(config); + let (health, suppressed) = match (config.censorship.tls_emulation, tls_cache) { + (true, Some(cache)) => { + cache + .profile_health_snapshot(&domains, TLS_FRONT_PROFILE_HEALTH_MAX_DOMAINS) + .await + } + _ => (Vec::new(), domains.len()), + }; + + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_domains TLS front configured profile domains by export status" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_domains gauge"); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"configured\"}} {}", + domains.len() + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"emitted\"}} {}", + health.len() + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_domains{{status=\"suppressed\"}} {}", + suppressed + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_info TLS front profile source and feature flags per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_info gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_quality_info TLS front profile quality and key-share group per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_quality_info gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_age_seconds Age of cached TLS front profile data per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_age_seconds gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_server_hello_bytes TLS front cached ServerHello record body bytes per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_server_hello_bytes gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_server_hello_extensions TLS front cached visible ServerHello extension count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_server_hello_extensions gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_app_data_records TLS front cached app-data record count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_app_data_records gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_ticket_records TLS front cached ticket-like tail record count per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_ticket_records gauge"); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_change_cipher_spec_records TLS front cached ChangeCipherSpec record count per configured domain" + ); + let _ = writeln!( + out, + "# TYPE telemt_tls_front_profile_change_cipher_spec_records gauge" + ); + let _ = writeln!( + out, + "# HELP telemt_tls_front_profile_app_data_bytes TLS front cached total app-data bytes per configured domain" + ); + let _ = writeln!(out, "# TYPE telemt_tls_front_profile_app_data_bytes gauge"); + + for item in health { + let domain = prometheus_label_value(&item.domain); + let _ = writeln!( + out, + "telemt_tls_front_profile_info{{domain=\"{}\",source=\"{}\",is_default=\"{}\",has_cert_info=\"{}\",has_cert_payload=\"{}\"}} 1", + domain, item.source, item.is_default, item.has_cert_info, item.has_cert_payload + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_quality_info{{domain=\"{}\",quality=\"{}\",key_share_group=\"{}\"}} 1", + domain, item.quality, item.key_share_group + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_age_seconds{{domain=\"{}\"}} {}", + domain, item.age_seconds + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_server_hello_bytes{{domain=\"{}\"}} {}", + domain, item.server_hello_record_len + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_server_hello_extensions{{domain=\"{}\"}} {}", + domain, item.server_hello_extensions + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_app_data_records{{domain=\"{}\"}} {}", + domain, item.app_data_records + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_ticket_records{{domain=\"{}\"}} {}", + domain, item.ticket_records + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_change_cipher_spec_records{{domain=\"{}\"}} {}", + domain, item.change_cipher_spec_count + ); + let _ = writeln!( + out, + "telemt_tls_front_profile_app_data_bytes{{domain=\"{}\"}} {}", + domain, item.total_app_data_len + ); + } +} diff --git a/src/proxy/middle_relay/session.rs b/src/proxy/middle_relay/session.rs index 371a5c5..22477ee 100644 --- a/src/proxy/middle_relay/session.rs +++ b/src/proxy/middle_relay/session.rs @@ -78,25 +78,11 @@ where return Err(ProxyError::RouteSwitched); } - // Per-user ad_tag from access.user_ad_tags; fallback to general.ad_tag (hot-reloadable) - let user_tag: Option> = config - .access - .user_ad_tags - .get(&user) - .and_then(|s| hex::decode(s).ok()) - .filter(|v| v.len() == 16); - let global_tag: Option> = config - .general - .ad_tag - .as_ref() - .and_then(|s| hex::decode(s).ok()) - .filter(|v| v.len() == 16); - let effective_tag = user_tag.or(global_tag); - - let proto_flags = proto_flags_for_tag(proto_tag, effective_tag.is_some()); - let effective_tag_array = effective_tag - .as_deref() - .and_then(|tag| <[u8; 16]>::try_from(tag).ok()); + // Per-user ad_tag from access.user_ad_tags with fallback to general.ad_tag. + // Decoded once at config load / hot-reload (see ProxyConfig::effective_ad_tag), + // so this is a cache lookup rather than a per-session hex decode + allocation. + let effective_tag_array = config.effective_ad_tag(&user); + let proto_flags = proto_flags_for_tag(proto_tag, effective_tag_array.is_some()); debug!( trace_id = format_args!("0x{:016x}", trace_id), user = %user,