From 10d357b8ecb2319c1203ac49dccc88d3f6132839 Mon Sep 17 00:00:00 2001 From: nineap Date: Thu, 9 Jul 2026 16:37:25 +0700 Subject: [PATCH] Graceful shutdown (SIGTERM) + HTTP /health endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Раньше shutdown_signal вообще не существовал — весь механизм graceful shutdown в network.rs был мёртвым кодом, ничего его не вызывало. main.rs теперь spawn'ит net.run() и последовательно ждёт сигнал, затем drain (НЕ tokio::select! — select! убил бы run()-future на середине drain'а в момент, когда shutdown-future резолвится). /health на 127.0.0.1 отдаёт активные соединения — для systemd/LB health-check. Co-Authored-By: Claude Sonnet 5 --- server/netrunner-server.dev.service | 5 ++- server/netrunner-server.service | 6 ++- server/src/health.rs | 62 +++++++++++++++++++++++++++++ server/src/main.rs | 62 +++++++++++++++++++++++++++-- server/src/network.rs | 50 +++++++++++++++++++++++ 5 files changed, 180 insertions(+), 5 deletions(-) create mode 100644 server/src/health.rs diff --git a/server/netrunner-server.dev.service b/server/netrunner-server.dev.service index 4b6281e..e5cfe2b 100644 --- a/server/netrunner-server.dev.service +++ b/server/netrunner-server.dev.service @@ -5,10 +5,13 @@ After=network.target [Service] User=root WorkingDirectory=/root/netr-core -ExecStart=/root/netr-core/netrunner-server --host=0.0.0.0 --port=8443 +ExecStart=/root/netr-core/netrunner-server --host=0.0.0.0 --port=8443 --health-port=9092 Restart=always RestartSec=3 LimitNOFILE=65535 +# systemctl stop/restart шлёт SIGTERM — теперь сервер ловит его сам и +# дожидается отключения активных клиентов вместо мгновенного обрыва. +TimeoutStopSec=35 [Install] WantedBy=multi-user.target diff --git a/server/netrunner-server.service b/server/netrunner-server.service index a3f2d1d..c3d7d24 100644 --- a/server/netrunner-server.service +++ b/server/netrunner-server.service @@ -5,10 +5,14 @@ After=network.target [Service] User=root WorkingDirectory=/root/netr-core -ExecStart=/root/netr-core/netrunner-server --host=0.0.0.0 --port=443 +ExecStart=/root/netr-core/netrunner-server --host=0.0.0.0 --port=443 --health-port=9091 Restart=always RestartSec=3 LimitNOFILE=65535 +# systemctl stop/restart шлёт SIGTERM — теперь сервер ловит его сам и +# дожидается отключения активных клиентов (см. server/src/main.rs, +# server/src/network.rs) вместо мгновенного обрыва по TimeoutStopSec. +TimeoutStopSec=35 [Install] WantedBy=multi-user.target diff --git a/server/src/health.rs b/server/src/health.rs new file mode 100644 index 0000000..1cd9b4b --- /dev/null +++ b/server/src/health.rs @@ -0,0 +1,62 @@ +//! Минимальный HTTP `/health` для supervisor'а (systemd/docker healthcheck) — +//! отдельный порт, не тот, где слушается замаскированный туннельный протокол +//! (смешивать их нельзя: health-эндпоинт — обычный читаемый HTTP, это выдало +//! бы DPI ровно то, что декой должен скрывать). По умолчанию биндится только +//! на 127.0.0.1 — наружу торчать не должен, это внутренняя проверка для +//! supervisor'а/docker healthcheck на этой же машине. +//! +//! Не использует HTTP-фреймворк (лишняя зависимость ради одного эндпоинта) — +//! отвечает одним и тем же 200+JSON на любое подключение, не разбирая запрос. + +use netrunner_logger::{error, info, warn}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpListener; +use tokio_util::sync::CancellationToken; + +/// Отдаёт реальное состояние — не просто "процесс жив", а сколько сейчас +/// физических соединений реально обслуживается (см. `Network::run`, +/// `active_connections`). +pub async fn run( + host: String, + port: u16, + active_connections: Arc, + token: CancellationToken, +) { + let addr = format!("{host}:{port}"); + let listener = match TcpListener::bind(&addr).await { + Ok(l) => l, + Err(e) => { + error!(error = %e, addr = %addr, "Не удалось поднять health-listener"); + return; + } + }; + info!("🩺 Health-эндпоинт слушает на {}", addr); + + loop { + tokio::select! { + _ = token.cancelled() => { + info!("Health-listener остановлен по сигналу отмены."); + break; + } + res = listener.accept() => { + let Ok((mut stream, _)) = res else { continue }; + let active = active_connections.load(Ordering::Relaxed); + tokio::spawn(async move { + let body = format!( + r#"{{"status":"ok","active_connections":{active}}}"# + ); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ); + if let Err(e) = stream.write_all(response.as_bytes()).await { + warn!(error = %e, "Health-listener: ошибка записи ответа"); + } + }); + } + } + } +} diff --git a/server/src/main.rs b/server/src/main.rs index 3a5e90c..94a1847 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -12,16 +12,46 @@ mod backend_client; mod diagnostics; +mod health; mod network; use clap::Parser; use netrunner_core::net::AuthValidator; -use netrunner_logger::Logger; +use netrunner_logger::{error, info, Logger}; use std::sync::Arc; use tokio_util::sync::CancellationToken; use crate::backend_client::BackendClient; use crate::network::Network; +/// Ждёт SIGTERM (docker stop/systemctl stop) или SIGINT (Ctrl+C) и отменяет +/// токен — раньше этой функции не было вообще: сервер получал сигнал прямо +/// от ОС мимо CancellationToken'а, и вся drain-логика в `Network::run` была +/// мертва, ни разу не срабатывая на реальном шатдауне. +async fn shutdown_signal(token: CancellationToken) { + let ctrl_c = async { + tokio::signal::ctrl_c() + .await + .expect("Не удалось установить обработчик Ctrl+C"); + }; + + #[cfg(unix)] + let terminate = async { + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("Не удалось установить обработчик SIGTERM") + .recv() + .await; + }; + + #[cfg(not(unix))] + let terminate = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => info!("🛑 Получен SIGINT (Ctrl+C). Останавливаемся..."), + _ = terminate => info!("🛑 Получен SIGTERM. Останавливаемся..."), + } + token.cancel(); +} + /// Аргументы командной строки сервера. #[derive(Parser, Debug)] #[command(author, version, about = "Netrunner Proxy Server")] @@ -52,6 +82,12 @@ struct Args { /// Обязателен, только если передан `--require-auth`. #[arg(long)] backend_url: Option, + + /// Порт для внутреннего HTTP `/health` (биндится только на 127.0.0.1 — + /// не для публичного доступа, только supervisor/docker healthcheck на + /// этой же машине). Не задан — health-эндпоинт выключен. + #[arg(long)] + health_port: Option, } fn main() { @@ -70,11 +106,31 @@ fn main() { None }; - let net = Network::new(args.host, args.port, args.decoy_host, auth); + let net = Network::new( + args.host, + args.port, + args.decoy_host, + auth, + args.health_port, + ); let rt = tokio::runtime::Runtime::new().expect("Failed to create Tokio runtime"); rt.block_on(async { - net.run(CancellationToken::new()).await; + let token = CancellationToken::new(); + let run_token = token.clone(); + // net.run() спавним отдельной задачей и дожидаемся её ПОСЛЕ сигнала — + // tokio::select! между сигналом и run() тут не подходит: select + // дропает недовершившуюся ветку целиком, оборвав drain-фазу в + // Network::run в момент получения самого сигнала, вместо того чтобы + // дать ей отработать. + let run_handle = tokio::spawn(async move { + net.run(run_token).await; + }); + + shutdown_signal(token).await; + if let Err(e) = run_handle.await { + error!(error = ?e, "Задача сервера завершилась с паникой при остановке"); + } }); } diff --git a/server/src/network.rs b/server/src/network.rs index 3b839ea..0e88020 100644 --- a/server/src/network.rs +++ b/server/src/network.rs @@ -11,11 +11,19 @@ use netrunner_core::net::{ TOPOLOGY_PRINT_INTERVAL, }; use netrunner_logger::{error, info, warn}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::time::Duration; use tokio::net::TcpListener; use tokio_util::sync::CancellationToken; use crate::diagnostics::{ClientDiagnosticsLogger, ServerDiagnosticsLogger}; +use crate::health; + +/// Сколько ждём при остановке, пока уже принятые соединения сами закроются, +/// прежде чем отпустить рантайм (который при Drop абортит все задачи разом, +/// без предупреждения клиентам). +const SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(30); /// Параметры прослушивания сервера. pub struct Network { @@ -27,6 +35,9 @@ pub struct Network { /// `None` — `--require-auth` не передан, авторизация и лимиты трафика /// выключены на этом инстансе целиком (поведение как до этой фичи). auth: Option>, + /// `None` — health-эндпоинт выключен (по умолчанию для обратной + /// совместимости с уже развёрнутыми нодами без этого флага). + health_port: Option, } impl Network { @@ -35,12 +46,14 @@ impl Network { port: u16, decoy_host: impl Into>, auth: Option>, + health_port: Option, ) -> Self { Self { host, port, decoy_host: decoy_host.into(), auth, + health_port, } } @@ -125,6 +138,23 @@ impl Network { info!("🌐 Netrunner Server: Listening on {}", addr); let listener = TcpListener::bind(&addr).await.expect("Server bind failed"); + // Число реально обслуживаемых физических соединений прямо сейчас — не + // "процесс жив", а "сколько клиентов на нём висит". Отдаётся в /health + // и используется ниже, чтобы дождаться отключения клиентов при + // остановке вместо мгновенного разрыва при drop рантайма. + let active_connections = Arc::new(AtomicU64::new(0)); + + if let Some(health_port) = self.health_port { + let health_token = token.clone(); + let health_connections = active_connections.clone(); + tokio::spawn(health::run( + "127.0.0.1".to_string(), + health_port, + health_connections, + health_token, + )); + } + loop { tokio::select! { _ = token.cancelled() => { @@ -145,6 +175,8 @@ impl Network { self.auth.clone(), ); + active_connections.fetch_add(1, Ordering::Relaxed); + let conn_counter = active_connections.clone(); tokio::spawn(async move { // "Входим" в этот Span. Все логи внутри handler.run() привяжутся к этому IP. let _enter = span.enter(); @@ -153,10 +185,28 @@ impl Network { if let Err(e) = handler.run().await { error!(error = %e, "⚠️ Server handler terminated with error"); } + conn_counter.fetch_sub(1, Ordering::Relaxed); }); } } } } + + // Graceful drain: приём новых соединений уже остановлен (цикл выше + // прерван), но уже принятые клиенты продолжают жить как detached-задачи + // рантайма. Без этого ожидания следующий за `run()` выход из + // `rt.block_on` уронит рантайм и оборвёт их все разом без предупреждения. + let drain_start = tokio::time::Instant::now(); + while active_connections.load(Ordering::Relaxed) > 0 { + if drain_start.elapsed() > SHUTDOWN_DRAIN_TIMEOUT { + warn!( + remaining = active_connections.load(Ordering::Relaxed), + "Drain timeout истёк, принудительно завершаем оставшиеся соединения" + ); + break; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + info!("✅ Drain завершён, соединений осталось: {}", active_connections.load(Ordering::Relaxed)); } }