Graceful shutdown (SIGTERM) + HTTP /health endpoint
Раньше 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<AtomicU64>,
|
||||
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: ошибка записи ответа");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+59
-3
@@ -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<String>,
|
||||
|
||||
/// Порт для внутреннего HTTP `/health` (биндится только на 127.0.0.1 —
|
||||
/// не для публичного доступа, только supervisor/docker healthcheck на
|
||||
/// этой же машине). Не задан — health-эндпоинт выключен.
|
||||
#[arg(long)]
|
||||
health_port: Option<u16>,
|
||||
}
|
||||
|
||||
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, "Задача сервера завершилась с паникой при остановке");
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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<Arc<dyn AuthValidator>>,
|
||||
/// `None` — health-эндпоинт выключен (по умолчанию для обратной
|
||||
/// совместимости с уже развёрнутыми нодами без этого флага).
|
||||
health_port: Option<u16>,
|
||||
}
|
||||
|
||||
impl Network {
|
||||
@@ -35,12 +46,14 @@ impl Network {
|
||||
port: u16,
|
||||
decoy_host: impl Into<Arc<str>>,
|
||||
auth: Option<Arc<dyn AuthValidator>>,
|
||||
health_port: Option<u16>,
|
||||
) -> 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));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user