feat: implement traffic insight marker backend (ubridge mark filter)

Add compute-side marker subsystem that receives ubridge UDP MARK signals
and forwards them as project-scoped notifications to the web UI for
real-time traffic coloring. Matched packets are always saved to per-link
pcaps for future replay.

Key components:
- gns3server/compute/marker/: MarkerManager (singleton, UDP listener +
  O(1) registry keyed by (node_id, filter_name)) + MarkerListener
  (DatagramProtocol parsing MARK lines per ubridge integration contract)
- gns3server/compute/base_node.py: marker sink/node config at ubridge
  startup; shared _ubridge_add_marker_filter / _ubridge_delete_marker_filter
- Per-node start_marker/stop_marker: VPCS (VPCS-{id}), QEMU
  (QEMU-{id}-{adapter}), Docker (bridge{adapter})
- Compute REST /markers/start + /markers/stop (vpcs/qemu/docker route files)
- Controller Link._markers state + UDPLink.start_marker/stop_marker/
  update_marker (mirror capture pattern: BPF validation, _choose_capture_side,
  node.post forwarding, topology persistence)
- Controller REST GET/POST/DELETE/PUT /v3/projects/{p}/links/{l}/markers
- Config: marker_listen_host / marker_listen_port in ServerSettings
- Signal routing: creation-time registry O(1) lookup, no node-table scan;
  project-scoped WS stream (not global); event payload always carries
  project_id for frontend scoping

Tests: 14 unit tests (registry, listener parsing, UDP round-trip);
562 existing tests pass with zero regressions.
This commit is contained in:
YueGuobin committed 2026-07-16 00:39:51 +08:00
1 parent 01d72aed9a
commit d2e9823e6c
21 files changed
+1135 -1

No files matched your search

+24
View File
@@ -0,0 +1,24 @@
#!/usr/bin/env python
#
# Copyright (C) 2024 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/>.
#
#
# Traffic-insight marker subsystem (compute side).
#
# ubridge's ``marker`` module is a passive tap: on a BPF match it emits a UDP
# ``MARK`` signal to a configured sink and/or appends the packet to a pcap.
# This package owns the compute-side UDP sink: one listener per compute process
# serves every ubridge on that host, disambiguated by ``node=<id>``.
@@ -0,0 +1,102 @@
#!/usr/bin/env python
#
# Copyright (C) 2024 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 asyncio
import logging
log = logging.getLogger(__name__)
class MarkerListener(asyncio.DatagramProtocol):
"""
Receives ubridge ``MARK`` signal datagrams and turns each into a
``marker.match`` notification.
Signal format (one datagram per match, ASCII)::
MARK <sec.usec> node=<id> filter=<name> tag=<tag> len=<n>\\n
The signal carries metadata only (no packet bytes). The compute-side
:class:`~gns3server.compute.marker.marker_manager.MarkerManager` registry
resolves ``(node_id, filter_name)`` to ``(project_id, link_id, tag)`` so the
event can be emitted on the right project-scoped notification stream.
"""
def __init__(self, manager):
# MarkerManager owns this listener and the registry.
self._manager = manager
self.transport = None
def connection_made(self, transport):
self.transport = transport
def datagram_received(self, data, addr):
try:
self._handle(data)
except Exception:
# Never let a malformed datagram kill the listener.
log.exception("Failed to process MARK datagram from %s: %r", addr, data)
def _handle(self, data):
line = data.decode("utf-8", errors="replace").strip()
if not line.startswith("MARK"):
return
parts = line.split()
# parts[0] == "MARK"; parts[1] == "<sec.usec>"
if len(parts) < 2:
return
try:
ts = float(parts[1])
except ValueError:
log.warning("Ignoring MARK signal with bad timestamp: %r", line)
return
kv = {}
for token in parts[2:]:
if "=" in token:
key, value = token.split("=", 1)
kv[key] = value
node_id = kv.get("node")
filter_name = kv.get("filter")
if not node_id or not filter_name:
return
# "-" means the field was unset on the ubridge side (see contract §3.3).
tag = kv.get("tag")
length = kv.get("len")
project_id, link_id, registered_tag = self._manager.lookup(node_id, filter_name)
if project_id is None:
log.warning(
"MARK signal for unregistered node=%s filter=%s, dropping", node_id, filter_name
)
return
event = {
"project_id": project_id,
"node_id": node_id,
"link_id": link_id,
"filter": filter_name,
# Prefer the value carried in the signal; fall back to the one we registered.
"tag": tag if tag and tag != "-" else registered_tag,
"ts": ts,
"len": int(length) if length and length.isdigit() else 0,
}
self._manager.emit_match(project_id, event)
+163
View File
@@ -0,0 +1,163 @@
#!/usr/bin/env python
#
# Copyright (C) 2024 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 asyncio
import logging
from gns3server.compute.marker.marker_listener import MarkerListener
from gns3server.compute.notification_manager import NotificationManager
log = logging.getLogger(__name__)
class MarkerManager:
"""
Singleton owning the compute-side UDP sink for ubridge ``MARK`` signals and
the registry that maps each ``(node_id, filter_name)`` back to its
``(project_id, link_id, tag)``.
The registry is populated when a marker is created on a link (the compute
endpoint has project_id + node_id from its route path and link_id/name/tag
from the request body) and cleared when the marker is deleted or the project
closed. At signal time it is an O(1) lookup — no node-table scan, and the
signal payload is untouched.
One listener per compute process serves every ubridge on that host; source
ubridges are disambiguated by ``node=<id>`` (UUID, globally unique).
"""
def __init__(self):
self._listener = None
self._transport = None
self._host = None
self._port = None
# Flat lookup: (node_id, filter_name) -> {"project_id", "link_id", "tag"}
self._entries = {}
# Reverse index for O(1) per-project teardown: project_id -> set of keys
self._by_project = {}
@property
def host(self):
"""The host the UDP sink is reachable on (for ``marker sink``)."""
return self._host
@property
def port(self):
"""The UDP port the sink is bound on (for ``marker sink``)."""
return self._port
@property
def running(self):
return self._transport is not None
async def start(self, host="127.0.0.1", port=0):
"""
Bind the UDP sink. ``port=0`` lets the OS choose a free port, which is
then read back and exposed via :attr:`port` for ``marker sink`` commands.
"""
if self.running:
return
loop = asyncio.get_running_loop()
self._listener = MarkerListener(self)
self._transport, _ = await loop.create_datagram_endpoint(
lambda: self._listener, local_addr=(host, port)
)
sock = self._transport.get_extra_info("socket")
self._host = host
self._port = sock.getsockname()[1] if sock else port
log.info("Marker signal sink listening on %s:%s", self._host, self._port)
async def stop(self):
"""Close the UDP sink and drop the whole registry."""
if self._transport:
self._transport.close()
self._transport = None
self._listener = None
self._entries.clear()
self._by_project.clear()
self._host = None
self._port = None
def register(self, project_id, node_id, filter_name, link_id, tag=None):
"""
Record that ``filter_name`` on ``node_id`` belongs to ``project_id`` /
``link_id``. Called from the compute marker-start endpoint.
Re-registering the same key updates the stored tag (e.g. on re-add).
"""
key = (node_id, filter_name)
self._entries[key] = {"project_id": project_id, "link_id": link_id, "tag": tag}
self._by_project.setdefault(project_id, set()).add(key)
def unregister(self, node_id, filter_name):
"""Forget a single marker. Returns True if something was removed."""
key = (node_id, filter_name)
entry = self._entries.pop(key, None)
if entry is None:
return False
project_entries = self._by_project.get(entry["project_id"])
if project_entries is not None:
project_entries.discard(key)
if not project_entries:
self._by_project.pop(entry["project_id"], None)
return True
def unregister_project(self, project_id):
"""Drop every marker belonging to ``project_id`` (project close)."""
keys = self._by_project.pop(project_id, None)
if not keys:
return
for key in keys:
self._entries.pop(key, None)
def lookup(self, node_id, filter_name):
"""
O(1) resolution of an incoming signal to its project/link/tag.
:returns: (project_id, link_id, tag) or (None, None, None) on miss.
"""
entry = self._entries.get((node_id, filter_name))
if entry is None:
return None, None, None
return entry["project_id"], entry["link_id"], entry["tag"]
def emit_match(self, project_id, event):
"""
Forward a parsed match as a project-scoped ``marker.match`` notification.
Flows compute -> controller dispatch -> project_emit -> web UI WS.
"""
NotificationManager.instance().emit("marker.match", event, project_id=project_id)
_instance = None
@staticmethod
def instance():
if MarkerManager._instance is None:
MarkerManager._instance = MarkerManager()
return MarkerManager._instance
@staticmethod
def reset():
MarkerManager._instance = None