From 4a5dc0b21b8aee724fd1a315374330d37cc55621 Mon Sep 17 00:00:00 2001 From: astronaut808 <38975427+astronaut808@users.noreply.github.com> Date: Sat, 18 Jul 2026 20:46:00 +0500 Subject: [PATCH 1/3] add handshake failure stage accounting --- docs/Architecture/API/API.md | 8 +++ src/api/mod.rs | 11 +++- src/api/model.rs | 8 +++ src/api/runtime_stats.rs | 12 +++-- src/metrics.rs | 23 +++++++++ src/proxy/client.rs | 98 ++++++++++++++++++++++++++++++++++-- src/stats/core_counters.rs | 12 +++++ src/stats/core_getters.rs | 15 ++++++ src/stats/mod.rs | 1 + 9 files changed, 180 insertions(+), 8 deletions(-) diff --git a/docs/Architecture/API/API.md b/docs/Architecture/API/API.md index 18f9927..78b34e8 100644 --- a/docs/Architecture/API/API.md +++ b/docs/Architecture/API/API.md @@ -334,6 +334,7 @@ Returned by `PATCH /v1/config` on success (`200`, or `202` when a reload was acc | `connections_bad_total` | `u64` | Failed/invalid client connections. | | `connections_bad_by_class` | `ClassCount[]` | Failed/invalid connections grouped by class. | | `handshake_failures_by_class` | `ClassCount[]` | Handshake failures grouped by class. | +| `handshake_failures_by_stage` | `StageCount[]` | Handshake failures grouped by state-machine stage. | | `handshake_timeouts_total` | `u64` | Handshake timeout count. | | `configured_users` | `usize` | Number of configured users in config. | @@ -343,6 +344,12 @@ Returned by `PATCH /v1/config` on success (`200`, or `202` when a reload was acc | `class` | `string` | Failure class label. | | `total` | `u64` | Counter value for this class. | +#### `StageCount` +| Field | Type | Description | +| --- | --- | --- | +| `stage` | `string` | State-machine stage label. | +| `total` | `u64` | Counter value for this stage. | + ### `SystemInfoData` | Field | Type | Description | | --- | --- | --- | @@ -927,6 +934,7 @@ JA3 follows the Salesforce ClientHello field order. JA4 follows the FoxIO TLS-cl | `connections_bad_total` | `u64` | Failed/invalid connections. | | `connections_bad_by_class` | `ClassCount[]` | Failed/invalid connections grouped by class. | | `handshake_failures_by_class` | `ClassCount[]` | Handshake failures grouped by class. | +| `handshake_failures_by_stage` | `StageCount[]` | Handshake failures grouped by state-machine stage. | | `handshake_timeouts_total` | `u64` | Handshake timeouts. | | `accept_permit_timeout_total` | `u64` | Listener admission permit acquisition timeouts. | | `configured_users` | `usize` | Configured user count. | diff --git a/src/api/mod.rs b/src/api/mod.rs index 968ea2d..6936b7d 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -54,8 +54,8 @@ use events::ApiEventStore; use http_utils::{error_response, read_json, read_optional_json, success_response}; use model::{ ApiFailure, ClassCount, CreateUserRequest, DeleteUserResponse, HealthData, HealthReadyData, - PatchUserRequest, ResetUserQuotaResponse, RotateSecretRequest, SummaryData, UserActiveIps, - is_valid_username, + PatchUserRequest, ResetUserQuotaResponse, RotateSecretRequest, StageCount, SummaryData, + UserActiveIps, is_valid_username, }; use patch::Patch; use runtime_edge::{ @@ -540,12 +540,19 @@ async fn handle( .into_iter() .map(|(class, total)| ClassCount { class, total }) .collect(); + let handshake_failures_by_stage = shared + .stats + .get_handshake_failure_stage_counts() + .into_iter() + .map(|(stage, total)| StageCount { stage, total }) + .collect(); let data = SummaryData { uptime_seconds: shared.stats.uptime_secs(), connections_total: shared.stats.get_connects_all(), connections_bad_total: shared.stats.get_connects_bad(), connections_bad_by_class, handshake_failures_by_class, + handshake_failures_by_stage, handshake_timeouts_total: shared.stats.get_handshake_timeouts(), configured_users: cfg.access.users.len(), }; diff --git a/src/api/model.rs b/src/api/model.rs index 5a183d5..eac7b8f 100644 --- a/src/api/model.rs +++ b/src/api/model.rs @@ -89,6 +89,12 @@ pub(super) struct ClassCount { pub(super) total: u64, } +#[derive(Serialize, Clone)] +pub(super) struct StageCount { + pub(super) stage: String, + pub(super) total: u64, +} + #[derive(Serialize)] pub(super) struct SummaryData { pub(super) uptime_seconds: f64, @@ -96,6 +102,7 @@ pub(super) struct SummaryData { pub(super) connections_bad_total: u64, pub(super) connections_bad_by_class: Vec, pub(super) handshake_failures_by_class: Vec, + pub(super) handshake_failures_by_stage: Vec, pub(super) handshake_timeouts_total: u64, pub(super) configured_users: usize, } @@ -113,6 +120,7 @@ pub(super) struct ZeroCoreData { pub(super) connections_bad_total: u64, pub(super) connections_bad_by_class: Vec, pub(super) handshake_failures_by_class: Vec, + pub(super) handshake_failures_by_stage: Vec, pub(super) handshake_timeouts_total: u64, pub(super) accept_permit_timeout_total: u64, pub(super) configured_users: usize, diff --git a/src/api/runtime_stats.rs b/src/api/runtime_stats.rs index 6fd9bb6..693891c 100644 --- a/src/api/runtime_stats.rs +++ b/src/api/runtime_stats.rs @@ -9,9 +9,9 @@ use super::ApiShared; use super::model::{ ClassCount, DcEndpointWriters, DcStatus, DcStatusData, MeWriterStatus, MeWritersData, MeWritersSummary, MinimalAllData, MinimalAllPayload, MinimalDcPathData, MinimalMeRuntimeData, - MinimalQuarantineData, UpstreamDcStatus, UpstreamStatus, UpstreamSummaryData, UpstreamsData, - ZeroAllData, ZeroCodeCount, ZeroCoreData, ZeroDesyncData, ZeroMiddleProxyData, ZeroPoolData, - ZeroUpstreamData, + MinimalQuarantineData, StageCount, UpstreamDcStatus, UpstreamStatus, UpstreamSummaryData, + UpstreamsData, ZeroAllData, ZeroCodeCount, ZeroCoreData, ZeroDesyncData, ZeroMiddleProxyData, + ZeroPoolData, ZeroUpstreamData, }; const FEATURE_DISABLED_REASON: &str = "feature_disabled"; @@ -36,6 +36,11 @@ pub(super) fn build_zero_all_data(stats: &Stats, configured_users: usize) -> Zer .into_iter() .map(|(class, total)| ClassCount { class, total }) .collect(); + let handshake_failure_stages = stats + .get_handshake_failure_stage_counts() + .into_iter() + .map(|(stage, total)| StageCount { stage, total }) + .collect(); let handshake_error_codes = stats .get_me_handshake_error_code_counts() .into_iter() @@ -50,6 +55,7 @@ pub(super) fn build_zero_all_data(stats: &Stats, configured_users: usize) -> Zer connections_bad_total: stats.get_connects_bad(), connections_bad_by_class: bad_connection_classes, handshake_failures_by_class: handshake_failure_classes, + handshake_failures_by_stage: handshake_failure_stages, handshake_timeouts_total: stats.get_handshake_timeouts(), accept_permit_timeout_total: stats.get_accept_permit_timeout_total(), configured_users, diff --git a/src/metrics.rs b/src/metrics.rs index fb35b84..9b2452e 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -739,6 +739,24 @@ async fn render_metrics( } } + let _ = writeln!( + out, + "# HELP telemt_handshake_failures_by_stage_total Handshake failures by state-machine stage" + ); + let _ = writeln!( + out, + "# TYPE telemt_handshake_failures_by_stage_total counter" + ); + if core_enabled { + for (stage, total) in stats.get_handshake_failure_stage_counts() { + let _ = writeln!( + out, + "telemt_handshake_failures_by_stage_total{{stage=\"{}\"}} {}", + stage, total + ); + } + } + let _ = writeln!( out, "# HELP telemt_auth_expensive_checks_total Expensive authentication candidate checks executed during handshake validation" @@ -3886,6 +3904,7 @@ mod tests { stats.increment_connects_bad_with_class("tls_handshake_bad_client"); stats.increment_handshake_timeouts(); stats.increment_handshake_failure_class("timeout"); + stats.increment_handshake_failure_stage("tls_post_serverhello_mtproto"); shared_state .handshake .auth_expensive_checks_total @@ -3950,6 +3969,9 @@ mod tests { )); assert!(output.contains("telemt_handshake_timeouts_total 1")); assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); + assert!(output.contains( + "telemt_handshake_failures_by_stage_total{stage=\"tls_post_serverhello_mtproto\"} 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")); @@ -4156,6 +4178,7 @@ mod tests { 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_handshake_failures_by_stage_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")); diff --git a/src/proxy/client.rs b/src/proxy/client.rs index bd30e5b..c2ed13b 100644 --- a/src/proxy/client.rs +++ b/src/proxy/client.rs @@ -7,7 +7,7 @@ use std::net::{IpAddr, SocketAddr}; use std::pin::Pin; use std::sync::Arc; use std::sync::OnceLock; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite}; use tokio::net::TcpStream; @@ -26,6 +26,57 @@ enum HandshakeOutcome { NeedsMasking(PostHandshakeFuture), } +#[derive(Clone, Copy)] +enum HandshakeFailureStage { + FirstPacketPrelude, + TlsClientHelloBody, + TlsCore, + TlsPostServerHelloMtproto, + DirectMtproto, +} + +impl HandshakeFailureStage { + fn as_label(self) -> &'static str { + match self { + Self::FirstPacketPrelude => "first_packet_prelude", + Self::TlsClientHelloBody => "tls_clienthello_body", + Self::TlsCore => "tls_core", + Self::TlsPostServerHelloMtproto => "tls_post_serverhello_mtproto", + Self::DirectMtproto => "direct_mtproto", + } + } + + fn as_code(self) -> u8 { + match self { + Self::FirstPacketPrelude => 0, + Self::TlsClientHelloBody => 1, + Self::TlsCore => 2, + Self::TlsPostServerHelloMtproto => 3, + Self::DirectMtproto => 4, + } + } + + fn from_code(code: u8) -> Self { + match code { + 1 => Self::TlsClientHelloBody, + 2 => Self::TlsCore, + 3 => Self::TlsPostServerHelloMtproto, + 4 => Self::DirectMtproto, + _ => Self::FirstPacketPrelude, + } + } +} + +fn set_handshake_failure_stage(stage: &AtomicU8, value: HandshakeFailureStage) { + stage.store(value.as_code(), Ordering::Relaxed); +} + +fn increment_handshake_failure_stage(stats: &Stats, stage: &AtomicU8) { + stats.increment_handshake_failure_stage( + HandshakeFailureStage::from_code(stage.load(Ordering::Relaxed)).as_label(), + ); +} + #[must_use = "UserConnectionReservation must be kept alive to retain user/IP reservation until release or drop"] struct UserConnectionReservation { stats: Arc, @@ -682,10 +733,15 @@ where let config_for_timeout = config.clone(); let beobachten_for_timeout = beobachten.clone(); let peer_for_timeout = real_peer.ip(); + let handshake_stage = AtomicU8::new(HandshakeFailureStage::FirstPacketPrelude.as_code()); // Phase 2: active handshake (with timeout after the first client byte) let outcome = match timeout(handshake_timeout, async { let mut first_bytes = [0u8; 5]; + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::FirstPacketPrelude, + ); if let Some(first_byte) = first_byte { first_bytes[0] = first_byte; stream.read_exact(&mut first_bytes[1..]).await?; @@ -726,6 +782,10 @@ where let mut handshake = vec![0u8; 5 + tls_len]; handshake[..5].copy_from_slice(&first_bytes); + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::TlsClientHelloBody, + ); let body_read = match read_with_progress(&mut stream, &mut handshake[5..]).await { Ok(n) => n, Err(e) => { @@ -770,6 +830,10 @@ where let (read_half, write_half) = tokio::io::split(stream); + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::TlsCore, + ); let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared( &handshake, read_half, write_half, real_peer, &config, &replay_checker, &rng, tls_cache.clone(), @@ -815,6 +879,10 @@ where ); debug!(peer = %peer, "Reading MTProto handshake through TLS"); + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::TlsPostServerHelloMtproto, + ); let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await?; let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..].try_into() .map_err(|_| ProxyError::InvalidHandshake("Short MTProto handshake".into()))?; @@ -886,6 +954,10 @@ where let mut handshake = [0u8; HANDSHAKE_LEN]; handshake[..5].copy_from_slice(&first_bytes); + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::DirectMtproto, + ); stream.read_exact(&mut handshake[5..]).await?; let (read_half, write_half) = tokio::io::split(stream); @@ -937,6 +1009,7 @@ where Ok(Err(e)) => { debug!(peer = %peer, error = %e, "Handshake failed"); stats_for_timeout.increment_handshake_failure_class(classify_handshake_failure_class(&e)); + increment_handshake_failure_stage(stats_for_timeout.as_ref(), &handshake_stage); record_handshake_failure_class( &beobachten_for_timeout, &config_for_timeout, @@ -948,6 +1021,7 @@ where Err(_) => { stats_for_timeout.increment_handshake_timeouts(); stats_for_timeout.increment_handshake_failure_class("timeout"); + increment_handshake_failure_stage(stats_for_timeout.as_ref(), &handshake_stage); debug!(peer = %peer, "Handshake timeout"); record_beobachten_class( &beobachten_for_timeout, @@ -1267,9 +1341,14 @@ impl RunningClientHandler { let beobachten_for_timeout = self.beobachten.clone(); let peer_for_timeout = self.peer.ip(); let peer_for_log = self.peer; + let handshake_stage = AtomicU8::new(HandshakeFailureStage::FirstPacketPrelude.as_code()); let outcome = match timeout(handshake_timeout, async { let mut first_bytes = [0u8; 5]; + set_handshake_failure_stage( + &handshake_stage, + HandshakeFailureStage::FirstPacketPrelude, + ); if let Some(first_byte) = first_byte { first_bytes[0] = first_byte; self.stream.read_exact(&mut first_bytes[1..]).await?; @@ -1283,9 +1362,11 @@ impl RunningClientHandler { debug!(peer = %peer, is_tls = is_tls, "Handshake type detected"); if is_tls { - self.handle_tls_client(first_bytes, local_addr).await + self.handle_tls_client(first_bytes, local_addr, &handshake_stage) + .await } else { - self.handle_direct_client(first_bytes, local_addr).await + self.handle_direct_client(first_bytes, local_addr, &handshake_stage) + .await } }) .await @@ -1294,6 +1375,7 @@ impl RunningClientHandler { Ok(Err(e)) => { debug!(peer = %peer_for_log, error = %e, "Handshake failed"); stats.increment_handshake_failure_class(classify_handshake_failure_class(&e)); + increment_handshake_failure_stage(stats.as_ref(), &handshake_stage); record_handshake_failure_class( &beobachten_for_timeout, &config_for_timeout, @@ -1305,6 +1387,7 @@ impl RunningClientHandler { Err(_) => { stats.increment_handshake_timeouts(); stats.increment_handshake_failure_class("timeout"); + increment_handshake_failure_stage(stats.as_ref(), &handshake_stage); debug!(peer = %peer_for_log, "Handshake timeout"); record_beobachten_class( &beobachten_for_timeout, @@ -1323,6 +1406,7 @@ impl RunningClientHandler { mut self, first_bytes: [u8; 5], local_addr: SocketAddr, + handshake_stage: &AtomicU8, ) -> Result { let peer = self.peer; @@ -1358,6 +1442,7 @@ impl RunningClientHandler { let mut handshake = vec![0u8; 5 + tls_len]; handshake[..5].copy_from_slice(&first_bytes); + set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::TlsClientHelloBody); let body_read = match read_with_progress(&mut self.stream, &mut handshake[5..]).await { Ok(n) => n, Err(e) => { @@ -1412,6 +1497,7 @@ impl RunningClientHandler { let (read_half, write_half) = self.stream.into_split(); + set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::TlsCore); let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared( &handshake, read_half, @@ -1465,6 +1551,10 @@ impl RunningClientHandler { ); debug!(peer = %peer, "Reading MTProto handshake through TLS"); + set_handshake_failure_stage( + handshake_stage, + HandshakeFailureStage::TlsPostServerHelloMtproto, + ); let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await?; let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..] .try_into() @@ -1541,6 +1631,7 @@ impl RunningClientHandler { mut self, first_bytes: [u8; 5], local_addr: SocketAddr, + handshake_stage: &AtomicU8, ) -> Result { let peer = self.peer; @@ -1564,6 +1655,7 @@ impl RunningClientHandler { let mut handshake = [0u8; HANDSHAKE_LEN]; handshake[..5].copy_from_slice(&first_bytes); + set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::DirectMtproto); self.stream.read_exact(&mut handshake[5..]).await?; let config = self.config.clone(); diff --git a/src/stats/core_counters.rs b/src/stats/core_counters.rs index 9017270..48cf368 100644 --- a/src/stats/core_counters.rs +++ b/src/stats/core_counters.rs @@ -50,6 +50,18 @@ impl Stats { .or_insert_with(|| AtomicU64::new(0)); entry.fetch_add(1, Ordering::Relaxed); } + + pub fn increment_handshake_failure_stage(&self, stage: &'static str) { + if !self.telemetry_core_enabled() { + return; + } + let entry = self + .handshake_failure_stages + .entry(stage) + .or_insert_with(|| AtomicU64::new(0)); + entry.fetch_add(1, Ordering::Relaxed); + } + pub fn increment_current_connections_direct(&self) { self.current_connections_direct .fetch_add(1, Ordering::Relaxed); diff --git a/src/stats/core_getters.rs b/src/stats/core_getters.rs index c1730c1..5aa9be8 100644 --- a/src/stats/core_getters.rs +++ b/src/stats/core_getters.rs @@ -38,6 +38,21 @@ impl Stats { out } + pub fn get_handshake_failure_stage_counts(&self) -> Vec<(String, u64)> { + let mut out: Vec<(String, u64)> = self + .handshake_failure_stages + .iter() + .map(|entry| { + ( + entry.key().to_string(), + entry.value().load(Ordering::Relaxed), + ) + }) + .collect(); + out.sort_by(|a, b| a.0.cmp(&b.0)); + out + } + pub fn get_accept_permit_timeout_total(&self) -> u64 { self.accept_permit_timeout_total.load(Ordering::Relaxed) } diff --git a/src/stats/mod.rs b/src/stats/mod.rs index a16f8eb..afffd10 100644 --- a/src/stats/mod.rs +++ b/src/stats/mod.rs @@ -133,6 +133,7 @@ pub struct Stats { connects_bad: AtomicU64, connects_bad_classes: DashMap<&'static str, AtomicU64>, handshake_failure_classes: DashMap<&'static str, AtomicU64>, + handshake_failure_stages: DashMap<&'static str, AtomicU64>, current_connections_direct: AtomicU64, current_connections_me: AtomicU64, route_cutover_parked_direct_current: AtomicU64, From 7291c3192c23ea8175d422f2572a62c888313052 Mon Sep 17 00:00:00 2001 From: astronaut808 <38975427+astronaut808@users.noreply.github.com> Date: Sat, 18 Jul 2026 21:25:40 +0500 Subject: [PATCH 2/3] docs handshake failure stage diagnostics --- docs/Architecture/API/API.md | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/docs/Architecture/API/API.md b/docs/Architecture/API/API.md index 78b34e8..9685b29 100644 --- a/docs/Architecture/API/API.md +++ b/docs/Architecture/API/API.md @@ -350,6 +350,32 @@ Returned by `PATCH /v1/config` on success (`200`, or `202` when a reload was acc | `stage` | `string` | State-machine stage label. | | `total` | `u64` | Counter value for this stage. | +#### Handshake failure stage diagnostics + +`handshake_failures_by_class` and `telemt_handshake_failures_by_class_total` describe the error kind. `handshake_failures_by_stage` and `telemt_handshake_failures_by_stage_total` describe where the same failure happened in the handshake state machine. + +This does not add a DPI verdict or any protocol decision. The stage is derived from the existing Telemt handshake control flow and is counted only when the existing handshake failure or timeout accounting path is reached. + +Fixed stage labels: + +| Stage | Meaning | +| --- | --- | +| `first_packet_prelude` | Reading the first 5 bytes before selecting the TLS or direct branch. | +| `tls_clienthello_body` | Reading the TLS ClientHello body after the TLS record header. | +| `tls_core` | Running the TLS-F handshake/auth flow. | +| `tls_post_serverhello_mtproto` | Waiting for the 64-byte MTProto handshake after TLS ServerHello. | +| `direct_mtproto` | Reading the direct classic/secure 64-byte MTProto handshake. | + +Example: + +```text +telemt_handshake_failures_by_class_total{class="expected_64_got_0_unexpected_eof"} 3 +telemt_handshake_failures_by_stage_total{stage="direct_mtproto"} 1 +telemt_handshake_failures_by_stage_total{stage="tls_post_serverhello_mtproto"} 2 +``` + +This means the same EOF-while-reading-64-bytes failure happened once in the direct MTProto path and twice after TLS ServerHello. + ### `SystemInfoData` | Field | Type | Description | | --- | --- | --- | From d862deebb203a9e7a4ae86d684d2da213f985ab7 Mon Sep 17 00:00:00 2001 From: Alexey <247128645+axkurcom@users.noreply.github.com> Date: Sun, 19 Jul 2026 17:20:18 +0300 Subject: [PATCH 3/3] Fix handshake stage attribution and bound counters Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com> --- src/api/mod.rs | 11 +---- src/api/model.rs | 8 ---- src/api/runtime_stats.rs | 12 ++--- src/metrics.rs | 23 --------- src/proxy/client.rs | 98 ++------------------------------------ src/stats/core_counters.rs | 12 ----- src/stats/core_getters.rs | 15 ------ src/stats/mod.rs | 1 - 8 files changed, 8 insertions(+), 172 deletions(-) diff --git a/src/api/mod.rs b/src/api/mod.rs index 72bc8d7..40bfdc7 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -56,8 +56,8 @@ use events::ApiEventStore; use http_utils::{error_response, read_json, read_optional_json, success_response}; use model::{ ApiFailure, ClassCount, CreateUserRequest, DeleteUserResponse, HealthData, HealthReadyData, - PatchUserRequest, ResetUserQuotaResponse, RotateSecretRequest, StageCount, SummaryData, - UserActiveIps, is_valid_username, + PatchUserRequest, ResetUserQuotaResponse, RotateSecretRequest, SummaryData, UserActiveIps, + is_valid_username, }; use patch::Patch; use runtime_edge::{ @@ -571,19 +571,12 @@ async fn handle( .into_iter() .map(|(class, total)| ClassCount { class, total }) .collect(); - let handshake_failures_by_stage = shared - .stats - .get_handshake_failure_stage_counts() - .into_iter() - .map(|(stage, total)| StageCount { stage, total }) - .collect(); let data = SummaryData { uptime_seconds: shared.stats.uptime_secs(), connections_total: shared.stats.get_connects_all(), connections_bad_total: shared.stats.get_connects_bad(), connections_bad_by_class, handshake_failures_by_class, - handshake_failures_by_stage, handshake_timeouts_total: shared.stats.get_handshake_timeouts(), configured_users: cfg.access.users.len(), }; diff --git a/src/api/model.rs b/src/api/model.rs index eac7b8f..5a183d5 100644 --- a/src/api/model.rs +++ b/src/api/model.rs @@ -89,12 +89,6 @@ pub(super) struct ClassCount { pub(super) total: u64, } -#[derive(Serialize, Clone)] -pub(super) struct StageCount { - pub(super) stage: String, - pub(super) total: u64, -} - #[derive(Serialize)] pub(super) struct SummaryData { pub(super) uptime_seconds: f64, @@ -102,7 +96,6 @@ pub(super) struct SummaryData { pub(super) connections_bad_total: u64, pub(super) connections_bad_by_class: Vec, pub(super) handshake_failures_by_class: Vec, - pub(super) handshake_failures_by_stage: Vec, pub(super) handshake_timeouts_total: u64, pub(super) configured_users: usize, } @@ -120,7 +113,6 @@ pub(super) struct ZeroCoreData { pub(super) connections_bad_total: u64, pub(super) connections_bad_by_class: Vec, pub(super) handshake_failures_by_class: Vec, - pub(super) handshake_failures_by_stage: Vec, pub(super) handshake_timeouts_total: u64, pub(super) accept_permit_timeout_total: u64, pub(super) configured_users: usize, diff --git a/src/api/runtime_stats.rs b/src/api/runtime_stats.rs index 693891c..6fd9bb6 100644 --- a/src/api/runtime_stats.rs +++ b/src/api/runtime_stats.rs @@ -9,9 +9,9 @@ use super::ApiShared; use super::model::{ ClassCount, DcEndpointWriters, DcStatus, DcStatusData, MeWriterStatus, MeWritersData, MeWritersSummary, MinimalAllData, MinimalAllPayload, MinimalDcPathData, MinimalMeRuntimeData, - MinimalQuarantineData, StageCount, UpstreamDcStatus, UpstreamStatus, UpstreamSummaryData, - UpstreamsData, ZeroAllData, ZeroCodeCount, ZeroCoreData, ZeroDesyncData, ZeroMiddleProxyData, - ZeroPoolData, ZeroUpstreamData, + MinimalQuarantineData, UpstreamDcStatus, UpstreamStatus, UpstreamSummaryData, UpstreamsData, + ZeroAllData, ZeroCodeCount, ZeroCoreData, ZeroDesyncData, ZeroMiddleProxyData, ZeroPoolData, + ZeroUpstreamData, }; const FEATURE_DISABLED_REASON: &str = "feature_disabled"; @@ -36,11 +36,6 @@ pub(super) fn build_zero_all_data(stats: &Stats, configured_users: usize) -> Zer .into_iter() .map(|(class, total)| ClassCount { class, total }) .collect(); - let handshake_failure_stages = stats - .get_handshake_failure_stage_counts() - .into_iter() - .map(|(stage, total)| StageCount { stage, total }) - .collect(); let handshake_error_codes = stats .get_me_handshake_error_code_counts() .into_iter() @@ -55,7 +50,6 @@ pub(super) fn build_zero_all_data(stats: &Stats, configured_users: usize) -> Zer connections_bad_total: stats.get_connects_bad(), connections_bad_by_class: bad_connection_classes, handshake_failures_by_class: handshake_failure_classes, - handshake_failures_by_stage: handshake_failure_stages, handshake_timeouts_total: stats.get_handshake_timeouts(), accept_permit_timeout_total: stats.get_accept_permit_timeout_total(), configured_users, diff --git a/src/metrics.rs b/src/metrics.rs index 9b2452e..fb35b84 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -739,24 +739,6 @@ async fn render_metrics( } } - let _ = writeln!( - out, - "# HELP telemt_handshake_failures_by_stage_total Handshake failures by state-machine stage" - ); - let _ = writeln!( - out, - "# TYPE telemt_handshake_failures_by_stage_total counter" - ); - if core_enabled { - for (stage, total) in stats.get_handshake_failure_stage_counts() { - let _ = writeln!( - out, - "telemt_handshake_failures_by_stage_total{{stage=\"{}\"}} {}", - stage, total - ); - } - } - let _ = writeln!( out, "# HELP telemt_auth_expensive_checks_total Expensive authentication candidate checks executed during handshake validation" @@ -3904,7 +3886,6 @@ mod tests { stats.increment_connects_bad_with_class("tls_handshake_bad_client"); stats.increment_handshake_timeouts(); stats.increment_handshake_failure_class("timeout"); - stats.increment_handshake_failure_stage("tls_post_serverhello_mtproto"); shared_state .handshake .auth_expensive_checks_total @@ -3969,9 +3950,6 @@ mod tests { )); assert!(output.contains("telemt_handshake_timeouts_total 1")); assert!(output.contains("telemt_handshake_failures_by_class_total{class=\"timeout\"} 1")); - assert!(output.contains( - "telemt_handshake_failures_by_stage_total{stage=\"tls_post_serverhello_mtproto\"} 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")); @@ -4178,7 +4156,6 @@ mod tests { 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_handshake_failures_by_stage_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")); diff --git a/src/proxy/client.rs b/src/proxy/client.rs index c2ed13b..bd30e5b 100644 --- a/src/proxy/client.rs +++ b/src/proxy/client.rs @@ -7,7 +7,7 @@ use std::net::{IpAddr, SocketAddr}; use std::pin::Pin; use std::sync::Arc; use std::sync::OnceLock; -use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite}; use tokio::net::TcpStream; @@ -26,57 +26,6 @@ enum HandshakeOutcome { NeedsMasking(PostHandshakeFuture), } -#[derive(Clone, Copy)] -enum HandshakeFailureStage { - FirstPacketPrelude, - TlsClientHelloBody, - TlsCore, - TlsPostServerHelloMtproto, - DirectMtproto, -} - -impl HandshakeFailureStage { - fn as_label(self) -> &'static str { - match self { - Self::FirstPacketPrelude => "first_packet_prelude", - Self::TlsClientHelloBody => "tls_clienthello_body", - Self::TlsCore => "tls_core", - Self::TlsPostServerHelloMtproto => "tls_post_serverhello_mtproto", - Self::DirectMtproto => "direct_mtproto", - } - } - - fn as_code(self) -> u8 { - match self { - Self::FirstPacketPrelude => 0, - Self::TlsClientHelloBody => 1, - Self::TlsCore => 2, - Self::TlsPostServerHelloMtproto => 3, - Self::DirectMtproto => 4, - } - } - - fn from_code(code: u8) -> Self { - match code { - 1 => Self::TlsClientHelloBody, - 2 => Self::TlsCore, - 3 => Self::TlsPostServerHelloMtproto, - 4 => Self::DirectMtproto, - _ => Self::FirstPacketPrelude, - } - } -} - -fn set_handshake_failure_stage(stage: &AtomicU8, value: HandshakeFailureStage) { - stage.store(value.as_code(), Ordering::Relaxed); -} - -fn increment_handshake_failure_stage(stats: &Stats, stage: &AtomicU8) { - stats.increment_handshake_failure_stage( - HandshakeFailureStage::from_code(stage.load(Ordering::Relaxed)).as_label(), - ); -} - #[must_use = "UserConnectionReservation must be kept alive to retain user/IP reservation until release or drop"] struct UserConnectionReservation { stats: Arc, @@ -733,15 +682,10 @@ where let config_for_timeout = config.clone(); let beobachten_for_timeout = beobachten.clone(); let peer_for_timeout = real_peer.ip(); - let handshake_stage = AtomicU8::new(HandshakeFailureStage::FirstPacketPrelude.as_code()); // Phase 2: active handshake (with timeout after the first client byte) let outcome = match timeout(handshake_timeout, async { let mut first_bytes = [0u8; 5]; - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::FirstPacketPrelude, - ); if let Some(first_byte) = first_byte { first_bytes[0] = first_byte; stream.read_exact(&mut first_bytes[1..]).await?; @@ -782,10 +726,6 @@ where let mut handshake = vec![0u8; 5 + tls_len]; handshake[..5].copy_from_slice(&first_bytes); - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::TlsClientHelloBody, - ); let body_read = match read_with_progress(&mut stream, &mut handshake[5..]).await { Ok(n) => n, Err(e) => { @@ -830,10 +770,6 @@ where let (read_half, write_half) = tokio::io::split(stream); - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::TlsCore, - ); let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared( &handshake, read_half, write_half, real_peer, &config, &replay_checker, &rng, tls_cache.clone(), @@ -879,10 +815,6 @@ where ); debug!(peer = %peer, "Reading MTProto handshake through TLS"); - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::TlsPostServerHelloMtproto, - ); let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await?; let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..].try_into() .map_err(|_| ProxyError::InvalidHandshake("Short MTProto handshake".into()))?; @@ -954,10 +886,6 @@ where let mut handshake = [0u8; HANDSHAKE_LEN]; handshake[..5].copy_from_slice(&first_bytes); - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::DirectMtproto, - ); stream.read_exact(&mut handshake[5..]).await?; let (read_half, write_half) = tokio::io::split(stream); @@ -1009,7 +937,6 @@ where Ok(Err(e)) => { debug!(peer = %peer, error = %e, "Handshake failed"); stats_for_timeout.increment_handshake_failure_class(classify_handshake_failure_class(&e)); - increment_handshake_failure_stage(stats_for_timeout.as_ref(), &handshake_stage); record_handshake_failure_class( &beobachten_for_timeout, &config_for_timeout, @@ -1021,7 +948,6 @@ where Err(_) => { stats_for_timeout.increment_handshake_timeouts(); stats_for_timeout.increment_handshake_failure_class("timeout"); - increment_handshake_failure_stage(stats_for_timeout.as_ref(), &handshake_stage); debug!(peer = %peer, "Handshake timeout"); record_beobachten_class( &beobachten_for_timeout, @@ -1341,14 +1267,9 @@ impl RunningClientHandler { let beobachten_for_timeout = self.beobachten.clone(); let peer_for_timeout = self.peer.ip(); let peer_for_log = self.peer; - let handshake_stage = AtomicU8::new(HandshakeFailureStage::FirstPacketPrelude.as_code()); let outcome = match timeout(handshake_timeout, async { let mut first_bytes = [0u8; 5]; - set_handshake_failure_stage( - &handshake_stage, - HandshakeFailureStage::FirstPacketPrelude, - ); if let Some(first_byte) = first_byte { first_bytes[0] = first_byte; self.stream.read_exact(&mut first_bytes[1..]).await?; @@ -1362,11 +1283,9 @@ impl RunningClientHandler { debug!(peer = %peer, is_tls = is_tls, "Handshake type detected"); if is_tls { - self.handle_tls_client(first_bytes, local_addr, &handshake_stage) - .await + self.handle_tls_client(first_bytes, local_addr).await } else { - self.handle_direct_client(first_bytes, local_addr, &handshake_stage) - .await + self.handle_direct_client(first_bytes, local_addr).await } }) .await @@ -1375,7 +1294,6 @@ impl RunningClientHandler { Ok(Err(e)) => { debug!(peer = %peer_for_log, error = %e, "Handshake failed"); stats.increment_handshake_failure_class(classify_handshake_failure_class(&e)); - increment_handshake_failure_stage(stats.as_ref(), &handshake_stage); record_handshake_failure_class( &beobachten_for_timeout, &config_for_timeout, @@ -1387,7 +1305,6 @@ impl RunningClientHandler { Err(_) => { stats.increment_handshake_timeouts(); stats.increment_handshake_failure_class("timeout"); - increment_handshake_failure_stage(stats.as_ref(), &handshake_stage); debug!(peer = %peer_for_log, "Handshake timeout"); record_beobachten_class( &beobachten_for_timeout, @@ -1406,7 +1323,6 @@ impl RunningClientHandler { mut self, first_bytes: [u8; 5], local_addr: SocketAddr, - handshake_stage: &AtomicU8, ) -> Result { let peer = self.peer; @@ -1442,7 +1358,6 @@ impl RunningClientHandler { let mut handshake = vec![0u8; 5 + tls_len]; handshake[..5].copy_from_slice(&first_bytes); - set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::TlsClientHelloBody); let body_read = match read_with_progress(&mut self.stream, &mut handshake[5..]).await { Ok(n) => n, Err(e) => { @@ -1497,7 +1412,6 @@ impl RunningClientHandler { let (read_half, write_half) = self.stream.into_split(); - set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::TlsCore); let (mut tls_reader, tls_writer, tls_user) = match handle_tls_handshake_with_shared( &handshake, read_half, @@ -1551,10 +1465,6 @@ impl RunningClientHandler { ); debug!(peer = %peer, "Reading MTProto handshake through TLS"); - set_handshake_failure_stage( - handshake_stage, - HandshakeFailureStage::TlsPostServerHelloMtproto, - ); let mtproto_data = tls_reader.read_exact(HANDSHAKE_LEN).await?; let mtproto_handshake: [u8; HANDSHAKE_LEN] = mtproto_data[..] .try_into() @@ -1631,7 +1541,6 @@ impl RunningClientHandler { mut self, first_bytes: [u8; 5], local_addr: SocketAddr, - handshake_stage: &AtomicU8, ) -> Result { let peer = self.peer; @@ -1655,7 +1564,6 @@ impl RunningClientHandler { let mut handshake = [0u8; HANDSHAKE_LEN]; handshake[..5].copy_from_slice(&first_bytes); - set_handshake_failure_stage(handshake_stage, HandshakeFailureStage::DirectMtproto); self.stream.read_exact(&mut handshake[5..]).await?; let config = self.config.clone(); diff --git a/src/stats/core_counters.rs b/src/stats/core_counters.rs index 48cf368..9017270 100644 --- a/src/stats/core_counters.rs +++ b/src/stats/core_counters.rs @@ -50,18 +50,6 @@ impl Stats { .or_insert_with(|| AtomicU64::new(0)); entry.fetch_add(1, Ordering::Relaxed); } - - pub fn increment_handshake_failure_stage(&self, stage: &'static str) { - if !self.telemetry_core_enabled() { - return; - } - let entry = self - .handshake_failure_stages - .entry(stage) - .or_insert_with(|| AtomicU64::new(0)); - entry.fetch_add(1, Ordering::Relaxed); - } - pub fn increment_current_connections_direct(&self) { self.current_connections_direct .fetch_add(1, Ordering::Relaxed); diff --git a/src/stats/core_getters.rs b/src/stats/core_getters.rs index 5aa9be8..c1730c1 100644 --- a/src/stats/core_getters.rs +++ b/src/stats/core_getters.rs @@ -38,21 +38,6 @@ impl Stats { out } - pub fn get_handshake_failure_stage_counts(&self) -> Vec<(String, u64)> { - let mut out: Vec<(String, u64)> = self - .handshake_failure_stages - .iter() - .map(|entry| { - ( - entry.key().to_string(), - entry.value().load(Ordering::Relaxed), - ) - }) - .collect(); - out.sort_by(|a, b| a.0.cmp(&b.0)); - out - } - pub fn get_accept_permit_timeout_total(&self) -> u64 { self.accept_permit_timeout_total.load(Ordering::Relaxed) } diff --git a/src/stats/mod.rs b/src/stats/mod.rs index afffd10..a16f8eb 100644 --- a/src/stats/mod.rs +++ b/src/stats/mod.rs @@ -133,7 +133,6 @@ pub struct Stats { connects_bad: AtomicU64, connects_bad_classes: DashMap<&'static str, AtomicU64>, handshake_failure_classes: DashMap<&'static str, AtomicU64>, - handshake_failure_stages: DashMap<&'static str, AtomicU64>, current_connections_direct: AtomicU64, current_connections_me: AtomicU64, route_cutover_parked_direct_current: AtomicU64,