feat(api): add Notification schema and document notification streams

This commit is contained in:
Mark Paronyan 2026-10-07 10:39:36 +00:00
parent 2fb7056cb9
commit 86a279330d
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
7 changed files with 327 additions and 11 deletions

View File

@ -0,0 +1,41 @@
#
# Copyright (C) 2026 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
from fastapi.responses import JSONResponse
from gns3server import schemas
NDJSON_MEDIA_TYPE = "application/x-ndjson"
class NDJSONResponse(JSONResponse):
"""
Only used as the response_class of streaming routes to publish the application/x-ndjson media type
and the schema of the streamed objects in the OpenAPI document.
"""
media_type = NDJSON_MEDIA_TYPE
NOTIFICATION_STREAM_RESPONSES: dict[int | str, dict[str, Any]] = {
200: {
"model": schemas.Notification,
"description": "Stream of notifications, one JSON object per line. "
"Each line is a Notification, the first one is a ping and then a ping is sent at least every 5 seconds.",
}
}

View File

@ -30,6 +30,7 @@ from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.agent.web_wireshark.stats import collect_webwireshark_stats
from gns3server.api.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.config import Config
from gns3server.controller import Controller
from gns3server.controller.controller_error import ControllerError, ControllerForbiddenError
@ -254,10 +255,19 @@ async def statistics() -> dict:
}
@router.get("/notifications", dependencies=[Depends(get_current_active_user)])
@router.get(
"/notifications",
dependencies=[Depends(get_current_active_user)],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def controller_http_notifications(request: Request) -> StreamingResponse:
"""
Receive controller notifications about the controller from HTTP stream.
Receive controller notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/notifications/ws`.
Controller actions: `compute.*`, `template.*`, `project.created`, `project.updated`, `project.opened`,
`project.closed`, `project.deleted`, `settings.updated`, `log.*` and `ping`.
"""
from gns3server.api.server import app
@ -274,7 +284,7 @@ async def controller_http_notifications(request: Request) -> StreamingResponse:
finally:
log.info(f"Client {client} has disconnected from controller HTTP notification stream")
return StreamingResponse(event_stream(), media_type="application/json")
return StreamingResponse(event_stream(), media_type=NDJSON_MEDIA_TYPE)
@router.websocket("/notifications/ws")
@ -282,7 +292,8 @@ async def controller_ws_notifications(
websocket: WebSocket, current_user: schemas.User = Depends(get_current_active_user_from_websocket)
) -> None:
"""
Receive project notifications about the controller from WebSocket.
Receive controller notifications from a WebSocket, one Notification JSON object per text frame.
Same messages as the HTTP stream `/v3/notifications`.
"""
if current_user is None:

View File

@ -40,6 +40,7 @@ from fastapi.responses import FileResponse, StreamingResponse
from websockets.exceptions import ConnectionClosed, WebSocketException
from gns3server import schemas
from gns3server.api.responses import NDJSON_MEDIA_TYPE, NOTIFICATION_STREAM_RESPONSES, NDJSONResponse
from gns3server.controller import Controller, marker_replay
from gns3server.controller.controller_error import ControllerBadRequestError, ControllerError
from gns3server.controller.export_project import export_project as export_controller_project
@ -523,10 +524,20 @@ async def load_project(path: str = Body(..., embed=True)) -> schemas.Project:
return project.asdict()
@router.get("/{project_id}/notifications", dependencies=[Depends(has_privilege("Project.Audit"))])
@router.get(
"/{project_id}/notifications",
dependencies=[Depends(has_privilege("Project.Audit"))],
response_class=NDJSONResponse,
responses=NOTIFICATION_STREAM_RESPONSES,
)
async def project_http_notifications(project_id: UUID) -> StreamingResponse:
"""
Receive project notifications about the controller from HTTP stream.
Receive project notifications from an HTTP stream of newline delimited JSON objects (see the Notification schema).
The same messages are available as WebSocket text frames on `/v3/projects/{project_id}/notifications/ws`.
Marker events (`marker.match`) are not sent here, they are only available on
`/v3/projects/{project_id}/notifications/markers/ws`.
Project actions: `node.*`, `link.*`, `drawing.*`, `project.updated`, `snapshot.restored`, `log.*` and `ping`.
Required privilege: Project.Audit
"""
@ -555,7 +566,7 @@ async def project_http_notifications(project_id: UUID) -> StreamingResponse:
log.info(f"Project '{project.id}' is automatically closing due to no client listening")
await project.close()
return StreamingResponse(event_stream(), media_type="application/json")
return StreamingResponse(event_stream(), media_type=NDJSON_MEDIA_TYPE)
@router.websocket("/{project_id}/notifications/ws")
@ -565,7 +576,8 @@ async def project_ws_notifications(
current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")),
) -> None:
"""
Receive project notifications about the controller from WebSocket.
Receive project notifications from a WebSocket, one Notification JSON object per text frame.
Same messages as the HTTP stream `/v3/projects/{project_id}/notifications`.
Required privilege: Project.Audit
"""
@ -603,8 +615,8 @@ async def project_marker_ws_notifications(
current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")),
) -> None:
"""
Receive marker notifications (e.g. marker.match) for a project on a
dedicated WebSocket, separate from the main project stream so high-frequency
Receive marker notifications (`marker.match`, see the Notification schema) for a project on a
dedicated WebSocket, one JSON object per text frame, separate from the main project stream so high-frequency
marker.matches do not block topology events (node.*/link.*).
Required privilege: Project.Audit

View File

@ -76,7 +76,14 @@ def get_application() -> FastAPI:
application = FastAPI(
lifespan=tasks.lifespan,
title="GNS3 controller API",
description="This page describes the public controller API for GNS3",
description="This page describes the public controller API for GNS3.\n\n"
"## Notification streams\n\n"
"Notifications are available as HTTP streams of newline delimited JSON objects (`GET /v3/notifications` and "
"`GET /v3/projects/{project_id}/notifications`) and as WebSockets sending one JSON object per text frame "
"(`/v3/notifications/ws`, `/v3/projects/{project_id}/notifications/ws` and "
"`/v3/projects/{project_id}/notifications/markers/ws`). "
"OpenAPI cannot describe WebSocket routes, they carry the same `Notification` messages as the HTTP streams. "
"Marker events (`marker.match`) are only sent on the markers WebSocket.",
version="3.0.0",
docs_url=None,
redoc_url=None,

View File

@ -109,6 +109,7 @@ from .compute.vpcs_nodes import VPCS, VPCSCreate, VPCSUpdate
from .controller.capabilities import Capabilities
from .controller.iou_license import IOULicense
from .controller.netmiko import NetmikoDeviceType, NetmikoDeviceTypeList
from .controller.notifications import Notification, NotificationAction
from .controller.pools import Resource, ResourceCreate, ResourcePool, ResourcePoolCreate, ResourcePoolUpdate
from .controller.rbac import ACE, ACECreate, ACEUpdate, Privilege, Role, RoleCreate, RoleUpdate
from .controller.settings import SettingsResponse, SettingsUpdate, SettingsUpdateResponse
@ -261,6 +262,8 @@ __all__ = [
"NodeDuplicate",
"NodeFile",
"NodeUpdate",
"Notification",
"NotificationAction",
"OpenAIMessage",
"OpenAIToolCall",
"Privilege",

View File

@ -0,0 +1,154 @@
#
# Copyright (C) 2026 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from typing import Any, Dict, List, Optional, Union
from uuid import UUID
from pydantic import BaseModel, Field
from .computes import Compute
from .drawings import Drawing
from .links import Link
from .nodes import Node
from .projects import Project
from .snapshots import Snapshot
class NotificationAction(str, Enum):
"""
Values of the "action" field of a notification.
"""
ping = "ping"
log_info = "log.info"
log_warning = "log.warning"
log_error = "log.error"
node_created = "node.created"
node_updated = "node.updated"
node_deleted = "node.deleted"
node_interface_status = "node.interface_status"
link_created = "link.created"
link_updated = "link.updated"
link_deleted = "link.deleted"
link_web_wireshark_started = "link.web_wireshark_started"
drawing_created = "drawing.created"
drawing_updated = "drawing.updated"
drawing_deleted = "drawing.deleted"
marker_match = "marker.match"
snapshot_restored = "snapshot.restored"
project_created = "project.created"
project_updated = "project.updated"
project_opened = "project.opened"
project_closed = "project.closed"
project_deleted = "project.deleted"
compute_created = "compute.created"
compute_updated = "compute.updated"
compute_deleted = "compute.deleted"
template_created = "template.created"
template_updated = "template.updated"
template_deleted = "template.deleted"
settings_updated = "settings.updated"
class PingEvent(BaseModel):
"""
Event of a "ping" notification, sent when the stream is opened and then at least every 5 seconds.
"""
cpu_usage_percent: float = Field(..., description="CPU usage of the server in percent")
memory_usage_percent: float = Field(..., description="Memory usage of the server in percent")
disk_usage_percent: float = Field(..., description="Disk usage of the project directory in percent")
compute_id: Optional[str] = Field(None, description="Compute that sent the ping (project streams only)")
class LogEvent(BaseModel):
"""
Event of a "log.info", "log.warning" or "log.error" notification.
"""
message: str = Field(..., description="Log message")
class NodeInterfaceStatusEvent(BaseModel):
"""
Event of a "node.interface_status" notification.
"""
project_id: UUID
node_id: UUID
adapter_number: int
port_number: int
status: str = Field(..., description="Status of the interface")
class MarkerMatchEvent(BaseModel):
"""
Event of a "marker.match" notification, only sent on the markers WebSocket.
"""
project_id: UUID
node_id: UUID
link_id: Optional[UUID] = None
filter: str = Field(..., description="Name of the packet filter that matched")
tag: Optional[int] = Field(None, description="Tag of the marker")
ts: float = Field(..., description="Timestamp reported by uBridge (seconds since the epoch)")
len: int = Field(0, description="Length of the matching packet")
dir: Optional[str] = Field(None, description="Direction of the packet relative to the capture node")
class LinkWebWiresharkStartedEvent(BaseModel):
"""
Event of a "link.web_wireshark_started" notification.
"""
link_id: UUID
ws_url: Optional[str] = Field(None, description="URL of the Web Wireshark session")
class SettingsUpdatedEvent(BaseModel):
"""
Event of a "settings.updated" notification.
"""
changed: List[str] = Field(..., description="Names of the settings that changed (values are never sent)")
restart_required: List[str] = Field(..., description="Names of the changed settings that need a server restart")
class Notification(BaseModel):
"""
A message of the notification streams. HTTP streams send one JSON object per line (application/x-ndjson),
WebSocket streams send one JSON object per text frame.
"""
action: NotificationAction = Field(..., description="Type of the notification")
event: Union[
PingEvent,
LogEvent,
NodeInterfaceStatusEvent,
MarkerMatchEvent,
LinkWebWiresharkStartedEvent,
SettingsUpdatedEvent,
Node,
Link,
Drawing,
Project,
Compute,
Snapshot,
Dict[str, Any],
] = Field(..., description="Payload of the notification, its content depends on the action")
project_id: Optional[UUID] = Field(None, description="Project the notification belongs to, when set by the sender")
compute_id: Optional[str] = Field(None, description="Compute that emitted the notification, when set by the sender")

View File

@ -0,0 +1,88 @@
#
# Copyright (C) 2026 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# 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 json
import uuid
import pytest
from fastapi import FastAPI
from pydantic import ValidationError
from gns3server import schemas
from gns3server.utils.notification_queue import NotificationQueue
STREAM_PATHS = ["/v3/notifications", "/v3/projects/{project_id}/notifications"]
class TestNotificationOpenAPI:
def test_notification_schema_in_spec(self, app: FastAPI) -> None:
spec = app.openapi()
schema = spec["components"]["schemas"]["Notification"]
assert set(schema["required"]) == {"action", "event"}
assert {"action", "event", "project_id", "compute_id"} <= set(schema["properties"])
action_schema = spec["components"]["schemas"]["NotificationAction"]
assert {"ping", "node.created", "node.updated", "link.created", "marker.match"} <= set(action_schema["enum"])
@pytest.mark.parametrize("path", STREAM_PATHS)
def test_stream_responses_reference_schema(self, app: FastAPI, path: str) -> None:
content = app.openapi()["paths"][path]["get"]["responses"]["200"]["content"]
assert list(content) == ["application/x-ndjson"]
assert content["application/x-ndjson"]["schema"] == {"$ref": "#/components/schemas/Notification"}
class TestNotificationSchema:
def test_ping(self) -> None:
message = {
"action": "ping",
"event": {"cpu_usage_percent": 1.5, "memory_usage_percent": 40.0, "disk_usage_percent": 10.0},
}
notification = schemas.Notification.model_validate(message)
assert notification.action == schemas.NotificationAction.ping
def test_unknown_action(self) -> None:
with pytest.raises(ValidationError):
schemas.Notification.model_validate({"action": "unknown.action", "event": {}})
def test_optional_ids(self) -> None:
project_id = str(uuid.uuid4())
notification = schemas.Notification.model_validate(
{"action": "log.info", "event": {"message": "hello"}, "project_id": project_id, "compute_id": "local"}
)
assert str(notification.project_id) == project_id
assert notification.compute_id == "local"
def test_generic_event(self) -> None:
notification = schemas.Notification.model_validate(
{"action": "template.deleted", "event": {"template_id": str(uuid.uuid4())}}
)
assert notification.action == schemas.NotificationAction.template_deleted
@pytest.mark.asyncio
async def test_queue_message_matches_schema(self) -> None:
queue = NotificationQueue()
queue.put_nowait(("log.warning", {"message": "disk space is low"}, {"compute_id": "local"}))
await queue.get_json(5)
message = json.loads(await queue.get_json(5))
notification = schemas.Notification.model_validate(message)
assert notification.action == schemas.NotificationAction.log_warning
assert notification.compute_id == "local"