mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
fix: compute notification stream silently died on uncaught exceptions
Two exception paths could permanently kill the compute notification chain (no more compute.updated events, no reconnection until a server restart): - connect() only caught ComputeError, but _run_http_query translates HTTP status errors (401/403/404/...) into sibling ControllerError subclasses (and a raw fastapi HTTPException for unexpected statuses). Those escaped the fire-and-forget connect() task started at controller startup and died silently. Now they notify clients, schedule an exponential-backoff retry, and still re-raise for explicit callers. The dead web.HTTP* except branches (never reached since _run_http_query converts HTTP errors itself) are removed. - _connect_notification() only caught aiohttp.ClientError. A malformed frame (e.g. missing 'action') or any error raised while dispatching a compute event (e.g. a pydantic ValidationError in node.parse_node_response) escaped the task, skipped the reconnect scheduling placed after the try block, and killed the stream forever. Now any exception is logged with its traceback (the gather() future holding it was never retrieved, so nothing was ever printed) and the reconnect scheduling + final compute.updated emit live in the finally block so every exit path recovers. Also moves the usage-stats reset before the disconnect log line so the emitted compute.updated snapshot is consistent.
This commit is contained in:
YueGuobin
parent
89d7f866cb
commit
c709d74826
@ -24,7 +24,6 @@ import sys
|
||||
import io
|
||||
|
||||
from fastapi import HTTPException
|
||||
from aiohttp import web
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
from asyncio import timeout as asynctimeout
|
||||
@ -373,6 +372,27 @@ class Compute:
|
||||
except ControllerError:
|
||||
pass
|
||||
|
||||
async def _report_connection_failure(self, error):
|
||||
"""
|
||||
Update the connection state after a failure, notify clients and
|
||||
schedule a reconnection attempt with exponential backoff.
|
||||
"""
|
||||
|
||||
self._connected = False
|
||||
self._last_error = str(error)
|
||||
self._controller.notification.controller_emit("compute.updated", self.asdict())
|
||||
# Try to reconnect if server unavailable only if not during tests (otherwise we create a ressource usage bomb)
|
||||
if hasattr(sys, "_called_from_test") and sys._called_from_test:
|
||||
return
|
||||
self._connection_failure += 1
|
||||
# After 10 failures we close the project using the compute to avoid sync issues
|
||||
if self._connection_failure == 10:
|
||||
log.error(f"Could not connect to compute '{self._id}' after multiple attempts: {error}")
|
||||
await self._controller.close_compute_projects(self)
|
||||
# Exponential backoff: 5s, 10s, 20s, 40s, 80s, then cap at 300s
|
||||
delay = min(5 * (2 ** (self._connection_failure - 1)), 300)
|
||||
asyncio.get_event_loop().call_later(delay, lambda: asyncio.ensure_future(self._try_reconnect()))
|
||||
|
||||
@locking
|
||||
async def connect(self, report_failed_connection=False):
|
||||
"""
|
||||
@ -385,32 +405,20 @@ class Compute:
|
||||
response = await self._run_http_query("GET", "/capabilities")
|
||||
except ComputeError as e:
|
||||
# Update connection status and notify UI
|
||||
self._connected = False
|
||||
self._last_error = str(e)
|
||||
self._controller.notification.controller_emit("compute.updated", self.asdict())
|
||||
|
||||
await self._report_connection_failure(e)
|
||||
if report_failed_connection:
|
||||
raise
|
||||
log.warning(f"Cannot connect to compute '{self._id}': {e}")
|
||||
# Try to reconnect if server unavailable only if not during tests (otherwise we create a ressource usage bomb)
|
||||
if not hasattr(sys, "_called_from_test") or not sys._called_from_test:
|
||||
self._connection_failure += 1
|
||||
# After 10 failures we close the project using the compute to avoid sync issues
|
||||
if self._connection_failure == 10:
|
||||
log.error(f"Could not connect to compute '{self._id}' after multiple attempts: {e}")
|
||||
await self._controller.close_compute_projects(self)
|
||||
# Exponential backoff: 5s, 10s, 20s, 40s, 80s, then cap at 300s
|
||||
delay = min(5 * (2 ** (self._connection_failure - 1)), 300)
|
||||
asyncio.get_event_loop().call_later(delay, lambda: asyncio.ensure_future(self._try_reconnect()))
|
||||
return
|
||||
except web.HTTPNotFound:
|
||||
raise ControllerNotFoundError(f"The server {self._id} is not a GNS3 server or it's a 1.X server")
|
||||
except web.HTTPUnauthorized:
|
||||
raise ControllerUnauthorizedError(f"Invalid auth for server {self._id}")
|
||||
except web.HTTPServiceUnavailable:
|
||||
raise ControllerNotFoundError(f"The server {self._id} is unavailable")
|
||||
except ValueError:
|
||||
raise ComputeError(f"Invalid server url for server {self._id}")
|
||||
except (ControllerError, HTTPException) as e:
|
||||
# _run_http_query translates HTTP status errors into ControllerError
|
||||
# subclasses (or a raw HTTPException for unexpected status codes).
|
||||
# They used to escape this method and silently kill the fire-and-forget
|
||||
# connect() task started at controller startup: no notification, no retry.
|
||||
# Schedule the retry, then re-raise so explicit callers still get the error.
|
||||
await self._report_connection_failure(e)
|
||||
log.warning(f"Cannot connect to compute '{self._id}': {e}")
|
||||
raise
|
||||
|
||||
if "version" not in response.json:
|
||||
msg = f"The server {self._id} is not a GNS3 server"
|
||||
@ -488,22 +496,27 @@ class Compute:
|
||||
elif response.type == aiohttp.WSMsgType.CLOSED:
|
||||
pass
|
||||
break
|
||||
except aiohttp.ClientError as e:
|
||||
log.error(f"Client response error received on compute '{self._id}' WebSocket '{ws_url}': {e}")
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
# A malformed frame or an error raised while dispatching a compute event
|
||||
# used to escape this task (only aiohttp.ClientError was caught) and
|
||||
# permanently killed the notification stream: no more compute.updated
|
||||
# events and no reconnection until the server was restarted. Log the
|
||||
# error with its traceback and reconnect below.
|
||||
log.error(f"Error on compute '{self._id}' notification stream '{ws_url}': {e!r}", exc_info=True)
|
||||
finally:
|
||||
self._connected = False
|
||||
self._cpu_usage_percent = None
|
||||
self._memory_usage_percent = None
|
||||
self._disk_usage_percent = None
|
||||
log.info(f"Connection closed to compute '{self._id}' WebSocket '{ws_url}'")
|
||||
|
||||
# Try to reconnect after 1 second if server unavailable only if not during tests (otherwise we create a resources usage bomb)
|
||||
from gns3server.api.server import app
|
||||
if not app.state.exiting and not hasattr(sys, "_called_from_test"):
|
||||
log.info(f"Reconnecting to compute '{self._id}' WebSocket '{ws_url}'")
|
||||
asyncio.get_event_loop().call_later(1, lambda: asyncio.ensure_future(self.connect()))
|
||||
|
||||
self._cpu_usage_percent = None
|
||||
self._memory_usage_percent = None
|
||||
self._disk_usage_percent = None
|
||||
self._controller.notification.controller_emit("compute.updated", self.asdict())
|
||||
self._controller.notification.controller_emit("compute.updated", self.asdict())
|
||||
# Try to reconnect after 1 second if server unavailable only if not during tests (otherwise we create a resources usage bomb)
|
||||
from gns3server.api.server import app
|
||||
if not app.state.exiting and not hasattr(sys, "_called_from_test"):
|
||||
log.info(f"Reconnecting to compute '{self._id}' WebSocket '{ws_url}'")
|
||||
asyncio.get_event_loop().call_later(1, lambda: asyncio.ensure_future(self.connect()))
|
||||
|
||||
def _getUrl(self, path):
|
||||
host = self._host
|
||||
|
||||
@ -15,13 +15,22 @@
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import sys
|
||||
import json
|
||||
import asyncio
|
||||
import aiohttp
|
||||
import pytest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from gns3server.controller.project import Project
|
||||
from gns3server.controller.compute import Compute
|
||||
from gns3server.controller.controller_error import ControllerError, ControllerNotFoundError, ComputeConflictError
|
||||
from gns3server.controller.controller_error import (
|
||||
ControllerError,
|
||||
ControllerNotFoundError,
|
||||
ControllerUnauthorizedError,
|
||||
ComputeConflictError,
|
||||
)
|
||||
from pydantic import SecretStr
|
||||
from tests.utils import asyncio_patch, AsyncioMagicMock
|
||||
|
||||
@ -524,3 +533,100 @@ async def test_get_ip_on_same_subnet(controller):
|
||||
},
|
||||
]
|
||||
assert await compute1.get_ip_on_same_subnet(compute2) == ('192.168.2.1', '192.168.1.2')
|
||||
|
||||
|
||||
class FakeWebSocket:
|
||||
"""
|
||||
Minimal aiohttp WebSocketResponse stand-in for notification stream tests.
|
||||
"""
|
||||
|
||||
def __init__(self, frames):
|
||||
self._frames = list(frames)
|
||||
|
||||
async def __aenter__(self):
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc_info):
|
||||
return False
|
||||
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._frames:
|
||||
raise StopAsyncIteration
|
||||
return self._frames.pop(0)
|
||||
|
||||
|
||||
def _text_frame(payload):
|
||||
return SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=json.dumps(payload))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_notification_poison_frame_autoreconnects(compute, monkeypatch):
|
||||
"""
|
||||
A malformed frame must not permanently kill the notification stream: the
|
||||
error is logged, clients are notified and a reconnection is scheduled.
|
||||
"""
|
||||
|
||||
emit_mock = MagicMock()
|
||||
monkeypatch.setattr(compute._controller.notification, "controller_emit", emit_mock)
|
||||
frames = [
|
||||
_text_frame({"action": "ping", "event": {"cpu_usage_percent": 10.0, "memory_usage_percent": 20.0, "disk_usage_percent": 30.0}}),
|
||||
_text_frame({"event": {"poison": True}}), # missing "action": raises KeyError in the receive loop
|
||||
]
|
||||
session = MagicMock()
|
||||
session.closed = False
|
||||
session.ws_connect = MagicMock(return_value=FakeWebSocket(frames))
|
||||
compute._http_session = session
|
||||
|
||||
# allow the reconnection to be scheduled during the test
|
||||
monkeypatch.delattr(sys, "_called_from_test", raising=False)
|
||||
from gns3server.api.server import app as gns3_app
|
||||
monkeypatch.setattr(gns3_app.state, "exiting", False)
|
||||
|
||||
async def fake_connect():
|
||||
compute._reconnect_attempted = True
|
||||
monkeypatch.setattr(compute, "connect", fake_connect)
|
||||
|
||||
# must not raise despite the poison frame
|
||||
await compute._connect_notification()
|
||||
|
||||
actions = [c.args[0] for c in emit_mock.call_args_list]
|
||||
assert actions.count("compute.updated") >= 2 # one for the ping, one for the disconnect
|
||||
assert compute._connected is False
|
||||
|
||||
# the reconnection scheduled by the finally block fires after 1 second
|
||||
await asyncio.sleep(1.2)
|
||||
assert compute._reconnect_attempted is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_connect_http_error_notifies_schedules_retry_and_raises(compute, monkeypatch):
|
||||
"""
|
||||
HTTP-level failures (401/403/404...) reach connect() as ControllerError
|
||||
subclasses. They must notify clients, schedule a retry and still raise for
|
||||
explicit callers. They used to silently kill the fire-and-forget connect()
|
||||
task started at controller startup: no notification, no retry.
|
||||
"""
|
||||
|
||||
compute._connected = False
|
||||
emit_mock = MagicMock()
|
||||
monkeypatch.setattr(compute._controller.notification, "controller_emit", emit_mock)
|
||||
|
||||
async def raise_unauthorized(*args, **kwargs):
|
||||
raise ControllerUnauthorizedError("Invalid authentication for compute 'my_compute_id'")
|
||||
|
||||
monkeypatch.setattr(compute, "_run_http_query", raise_unauthorized)
|
||||
monkeypatch.delattr(sys, "_called_from_test", raising=False)
|
||||
scheduled_delays = []
|
||||
monkeypatch.setattr(asyncio.get_event_loop(), "call_later", lambda delay, callback: scheduled_delays.append(delay))
|
||||
|
||||
with pytest.raises(ControllerUnauthorizedError):
|
||||
await compute.connect()
|
||||
|
||||
assert compute._last_error == "Invalid authentication for compute 'my_compute_id'"
|
||||
assert compute.connected is False
|
||||
actions = [c.args[0] for c in emit_mock.call_args_list]
|
||||
assert "compute.updated" in actions
|
||||
assert scheduled_delays == [5] # first exponential backoff delay
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user