mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-02 09:20:22 +08:00
fix(proxy): 抑制 tunnel/hub 高频重复日志,增加快速断开退避机制
- proxy_tunnel: idle timeout 日志首次 warning,后续降级为 debug/info 汇总 - hub_transport: 连续断开日志降级,追踪快速断开并调整重连初始 delay - tunnel_manager: connect/disconnect 日志按 reconnect 计数分级输出
This commit is contained in:
@@ -26,6 +26,9 @@ router = APIRouter()
|
|||||||
# Per-node 锁: 防止并发的 connect/disconnect 写入 DB 时出现竞态(后断连覆盖先连接)
|
# Per-node 锁: 防止并发的 connect/disconnect 写入 DB 时出现竞态(后断连覆盖先连接)
|
||||||
_node_status_locks: dict[str, asyncio.Lock] = {}
|
_node_status_locks: dict[str, asyncio.Lock] = {}
|
||||||
|
|
||||||
|
# Per-node idle timeout 连续计数: 用于抑制重复日志
|
||||||
|
_idle_timeout_counts: dict[str, int] = {}
|
||||||
|
|
||||||
|
|
||||||
def _get_node_lock(node_id: str) -> asyncio.Lock:
|
def _get_node_lock(node_id: str) -> asyncio.Lock:
|
||||||
lock = _node_status_locks.get(node_id)
|
lock = _node_status_locks.get(node_id)
|
||||||
@@ -195,7 +198,19 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
|
|||||||
try:
|
try:
|
||||||
data = await asyncio.wait_for(ws.receive_bytes(), timeout=_IDLE_TIMEOUT)
|
data = await asyncio.wait_for(ws.receive_bytes(), timeout=_IDLE_TIMEOUT)
|
||||||
except asyncio.TimeoutError:
|
except asyncio.TimeoutError:
|
||||||
logger.warning("tunnel idle timeout for node_id={}", node_id)
|
count = _idle_timeout_counts.get(node_id, 0) + 1
|
||||||
|
_idle_timeout_counts[node_id] = count
|
||||||
|
# 首次 warning,后续每 10 次打印一条 info,其余 debug
|
||||||
|
if count == 1:
|
||||||
|
logger.warning("tunnel idle timeout for node_id={}", node_id)
|
||||||
|
elif count % 10 == 0:
|
||||||
|
logger.info(
|
||||||
|
"tunnel idle timeout for node_id={} (repeated {} times)",
|
||||||
|
node_id,
|
||||||
|
count,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug("tunnel idle timeout for node_id={}", node_id)
|
||||||
disconnect_reason = "idle timeout"
|
disconnect_reason = "idle timeout"
|
||||||
await ws.close(code=4004, reason="idle timeout")
|
await ws.close(code=4004, reason="idle timeout")
|
||||||
break
|
break
|
||||||
@@ -209,6 +224,8 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
|
|||||||
break
|
break
|
||||||
continue
|
continue
|
||||||
oversized_count = 0 # 正常帧重置计数
|
oversized_count = 0 # 正常帧重置计数
|
||||||
|
_idle_timeout_counts.pop(node_id, None) # 收到正常帧,重置 idle timeout 计数
|
||||||
|
manager.reset_reconnect_count(node_id) # 连接正常活跃,重置 reconnect 计数
|
||||||
try:
|
try:
|
||||||
frame = Frame.decode(data)
|
frame = Frame.decode(data)
|
||||||
except ValueError as e:
|
except ValueError as e:
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import gzip
|
import gzip
|
||||||
import json
|
import json
|
||||||
|
import time as _time
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
@@ -68,6 +69,11 @@ class HubConnectionManager:
|
|||||||
|
|
||||||
self._closing = False
|
self._closing = False
|
||||||
|
|
||||||
|
self._disconnect_count = 0 # 连续断开计数,用于抑制重复日志
|
||||||
|
# 连续快速断开退避:防止 Hub 端持续发送 GOAWAY 时产生重连风暴
|
||||||
|
self._last_disconnect_ts: float = 0.0
|
||||||
|
self._rapid_disconnect_count: int = 0
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_connected(self) -> bool:
|
def is_connected(self) -> bool:
|
||||||
ws = self._ws
|
ws = self._ws
|
||||||
@@ -126,7 +132,15 @@ class HubConnectionManager:
|
|||||||
|
|
||||||
self._reader_task = asyncio.create_task(self._reader_loop(ws))
|
self._reader_task = asyncio.create_task(self._reader_loop(ws))
|
||||||
self._ping_task = asyncio.create_task(self._ping_loop(ws))
|
self._ping_task = asyncio.create_task(self._ping_loop(ws))
|
||||||
logger.info("Hub worker channel connected: {}", self._config.worker_ws_url)
|
if self._disconnect_count == 0:
|
||||||
|
logger.info("Hub worker channel connected: {}", self._config.worker_ws_url)
|
||||||
|
else:
|
||||||
|
logger.debug(
|
||||||
|
"Hub worker channel connected: {} (after {} disconnects)",
|
||||||
|
self._config.worker_ws_url,
|
||||||
|
self._disconnect_count,
|
||||||
|
)
|
||||||
|
self._disconnect_count = 0
|
||||||
|
|
||||||
def _start_reconnect_loop(self) -> None:
|
def _start_reconnect_loop(self) -> None:
|
||||||
if self._closing:
|
if self._closing:
|
||||||
@@ -138,7 +152,9 @@ class HubConnectionManager:
|
|||||||
self._reconnect_task = asyncio.create_task(self._reconnect_loop())
|
self._reconnect_task = asyncio.create_task(self._reconnect_loop())
|
||||||
|
|
||||||
async def _reconnect_loop(self) -> None:
|
async def _reconnect_loop(self) -> None:
|
||||||
attempt = 0
|
# 如果连续快速断开(GOAWAY 风暴),初始 attempt 跳过 0-delay 阶段
|
||||||
|
rapid = self._rapid_disconnect_count
|
||||||
|
attempt = min(rapid, len(_RECONNECT_DELAYS_SECONDS) - 1)
|
||||||
while not self._closing and not self.is_connected:
|
while not self._closing and not self.is_connected:
|
||||||
delay = _RECONNECT_DELAYS_SECONDS[min(attempt, len(_RECONNECT_DELAYS_SECONDS) - 1)]
|
delay = _RECONNECT_DELAYS_SECONDS[min(attempt, len(_RECONNECT_DELAYS_SECONDS) - 1)]
|
||||||
if delay > 0:
|
if delay > 0:
|
||||||
@@ -149,7 +165,7 @@ class HubConnectionManager:
|
|||||||
break
|
break
|
||||||
await self._connect_once()
|
await self._connect_once()
|
||||||
if self.is_connected:
|
if self.is_connected:
|
||||||
logger.info("Hub worker channel reconnected")
|
logger.debug("Hub worker channel reconnected (attempt {})", attempt + 1)
|
||||||
break
|
break
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
attempt += 1
|
attempt += 1
|
||||||
@@ -192,7 +208,26 @@ class HubConnectionManager:
|
|||||||
self._pending_streams.clear()
|
self._pending_streams.clear()
|
||||||
|
|
||||||
if not self._closing:
|
if not self._closing:
|
||||||
logger.warning("Hub worker channel disconnected: {}", reason)
|
self._disconnect_count += 1
|
||||||
|
|
||||||
|
# 追踪连续快速断开:如果距上次断开不足 2 秒,累加计数;否则重置
|
||||||
|
now_mono = _time.monotonic()
|
||||||
|
if now_mono - self._last_disconnect_ts < 2.0:
|
||||||
|
self._rapid_disconnect_count += 1
|
||||||
|
else:
|
||||||
|
self._rapid_disconnect_count = 0
|
||||||
|
self._last_disconnect_ts = now_mono
|
||||||
|
|
||||||
|
if self._disconnect_count <= 1:
|
||||||
|
logger.warning("Hub worker channel disconnected: {}", reason)
|
||||||
|
elif self._disconnect_count % 10 == 0:
|
||||||
|
logger.info(
|
||||||
|
"Hub worker channel disconnected: {} (repeated {} times)",
|
||||||
|
reason,
|
||||||
|
self._disconnect_count,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug("Hub worker channel disconnected: {}", reason)
|
||||||
self._start_reconnect_loop()
|
self._start_reconnect_loop()
|
||||||
|
|
||||||
async def _send_frame(self, frame: Frame) -> None:
|
async def _send_frame(self, frame: Frame) -> None:
|
||||||
|
|||||||
@@ -229,6 +229,8 @@ class TunnelManager:
|
|||||||
self._connections: dict[str, list[TunnelConnection]] = {} # node_id -> [conn, ...]
|
self._connections: dict[str, list[TunnelConnection]] = {} # node_id -> [conn, ...]
|
||||||
self._background_tasks: set[asyncio.Task[None]] = set()
|
self._background_tasks: set[asyncio.Task[None]] = set()
|
||||||
self._draining: bool = False
|
self._draining: bool = False
|
||||||
|
# 每个 node 的连续 reconnect 计数,用于抑制高频 connect/disconnect 日志
|
||||||
|
self._reconnect_counts: dict[str, int] = {}
|
||||||
|
|
||||||
def _background(self, coro: Any) -> None: # noqa: ANN401
|
def _background(self, coro: Any) -> None: # noqa: ANN401
|
||||||
"""启动 fire-and-forget task,通过 set 持有引用防止 GC 回收,完成后自动清理"""
|
"""启动 fire-and-forget task,通过 set 持有引用防止 GC 回收,完成后自动清理"""
|
||||||
@@ -269,12 +271,22 @@ class TunnelManager:
|
|||||||
conns = []
|
conns = []
|
||||||
self._connections[conn.node_id] = conns
|
self._connections[conn.node_id] = conns
|
||||||
conns.append(conn)
|
conns.append(conn)
|
||||||
logger.info(
|
reconn = self._reconnect_counts.get(conn.node_id, 0)
|
||||||
"tunnel connected: node_id={}, name={}, pool_size={}",
|
if reconn == 0:
|
||||||
conn.node_id,
|
logger.info(
|
||||||
conn.node_name,
|
"tunnel connected: node_id={}, name={}, pool_size={}",
|
||||||
len(conns),
|
conn.node_id,
|
||||||
)
|
conn.node_name,
|
||||||
|
len(conns),
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug(
|
||||||
|
"tunnel reconnected: node_id={}, name={}, pool_size={}, reconnect_count={}",
|
||||||
|
conn.node_id,
|
||||||
|
conn.node_name,
|
||||||
|
len(conns),
|
||||||
|
reconn,
|
||||||
|
)
|
||||||
|
|
||||||
def unregister(self, conn: TunnelConnection) -> bool:
|
def unregister(self, conn: TunnelConnection) -> bool:
|
||||||
"""
|
"""
|
||||||
@@ -297,14 +309,37 @@ class TunnelManager:
|
|||||||
self._connections.pop(conn.node_id, None)
|
self._connections.pop(conn.node_id, None)
|
||||||
|
|
||||||
remaining = len(conns) if conns else 0
|
remaining = len(conns) if conns else 0
|
||||||
logger.info(
|
reconn = self._reconnect_counts.get(conn.node_id, 0) + 1
|
||||||
"tunnel disconnected: node_id={}, name={}, remaining={}",
|
self._reconnect_counts[conn.node_id] = reconn
|
||||||
conn.node_id,
|
# 首次断开 info,后续每 10 次 info 汇总,其余 debug
|
||||||
conn.node_name,
|
if reconn == 1:
|
||||||
remaining,
|
logger.info(
|
||||||
)
|
"tunnel disconnected: node_id={}, name={}, remaining={}",
|
||||||
|
conn.node_id,
|
||||||
|
conn.node_name,
|
||||||
|
remaining,
|
||||||
|
)
|
||||||
|
elif reconn % 10 == 0:
|
||||||
|
logger.info(
|
||||||
|
"tunnel disconnected: node_id={}, name={}, remaining={} (repeated {} times)",
|
||||||
|
conn.node_id,
|
||||||
|
conn.node_name,
|
||||||
|
remaining,
|
||||||
|
reconn,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.debug(
|
||||||
|
"tunnel disconnected: node_id={}, name={}, remaining={}",
|
||||||
|
conn.node_id,
|
||||||
|
conn.node_name,
|
||||||
|
remaining,
|
||||||
|
)
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
def reset_reconnect_count(self, node_id: str) -> None:
|
||||||
|
"""重置 reconnect 计数(当连接恢复正常活动时调用)"""
|
||||||
|
self._reconnect_counts.pop(node_id, None)
|
||||||
|
|
||||||
async def shutdown_all(self, drain_timeout: float = 60.0) -> None:
|
async def shutdown_all(self, drain_timeout: float = 60.0) -> None:
|
||||||
"""优雅关闭所有 tunnel 连接:drain 飞行中请求 -> GoAway -> 关闭 WebSocket。
|
"""优雅关闭所有 tunnel 连接:drain 飞行中请求 -> GoAway -> 关闭 WebSocket。
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user