mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-12 21:45:41 +03:00
90bcb32473
Switch-to-switch links were the last shape left on the relay: the switch fast path works by absorbing the *peer's* anchor, and a switch has none — its ports are its own bridge, and a kernel interface belongs to exactly one bridge, so two bridges cannot be joined by enslaving anything. Join them with a veth pair instead: one end per bridge, each side enslaving its own end and applying its own port mode to it — an access port on one side and a trunk on the other behaves like the real cable it models. uBridge's docker create_veth/delete_veth (the primitive every Docker adapter is already born from) creates and destroys the pair; the names derive from the link id (utils.kernel_anchor.kernel_cascade_names), so each side computes its own end and the names survive a restart. The pair belongs to the link, and the two ends are wired by two independent uBridge processes — the project-open batch drives different nodes concurrently — so the protocol is order-free: both sides create the pair when their own end is missing (one create wins; the loser's failure is re-checked against the kernel and "the pair is there" is not an error), and either side's teardown destroys it (deleting one veth end takes the other with it, so both converge on the same gone pair and the second delif answers ENOENT). A pre-existing end — this switch restarted, or a leftover from a previous life of the link id — is reused rather than recreated, because the other end may still be a live port of the peer switch; a stale tc qdisc is all a reused end can carry. No new wire format was needed: a cascade is an anchor NIO on both ends, the absorbed-anchor case plus the peer end's name. That also corrects the filter routing — the single-sided rule exists because an absorbed anchor is one interface, while a cascade has one per switch, so impairments go to both ends like any other kernel link. Tests: the wire format, the eligibility flip and the both-ends re-push on node start (controller); the naming contract (utils); the switch-side lifecycle — create, reuse, lost create race, delete destroys the pair (compute routes). Drive-by: the relay filter test's compute mock carried no awaitable `put`, so tests/controller/test_kernel_datapath_link.py failed when run alone while an argument-less full run passed; callers now get one.
1008 lines
46 KiB
Python
1008 lines
46 KiB
Python
#!/usr/bin/env python
|
|
#
|
|
# Copyright (C) 2016 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.config import Config
|
|
from gns3server.utils.application_id import is_iol_runner_environment
|
|
from gns3server.utils.kernel_anchor import kernel_anchor_name, kernel_cascade_names
|
|
from gns3server.utils.packet_filter_validation import (
|
|
KERNEL_UNSUPPORTED_FILTERS,
|
|
split_kernel_only_features,
|
|
validate_bpf_syntax,
|
|
)
|
|
|
|
from .compute import ComputeError
|
|
from .controller_error import ControllerError, ControllerNotFoundError
|
|
from .link import _UNSET, Link
|
|
from .node_types import BUILTIN_NODE_TYPES
|
|
|
|
# Node types that can host a marker (have a uBridge bridge to attach the
|
|
# `mark` filter to). Mirrors _get_filter_node in link.py, minus "nat"
|
|
# (which has no uBridge) and "ethernet_hub" (still Dynamips-hosted, no
|
|
# uBridge of its own). "ethernet_switch" hosts markers on the per-port
|
|
# uBridge relays of its brctl kernel-bridge backend.
|
|
_MARKER_CAPABLE_TYPES = frozenset(
|
|
{
|
|
"vpcs",
|
|
"qemu",
|
|
"docker",
|
|
"iou",
|
|
"dynamips",
|
|
"cloud",
|
|
"ethernet_switch",
|
|
}
|
|
)
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def _is_unix_socket_docker(node):
|
|
"""
|
|
Best-effort detection of vendor containers bridged through AF_UNIX
|
|
socket pairs: they have no host-side interface to enslave into a kernel
|
|
bridge. Two ways in — the generic GNS3_UNIX_SOCKET_NIO knob, or the IOL
|
|
runner marker, whose class (IOLDockerVM, selected on GNS3_IOL_RUNNER)
|
|
forces the unix-socket wiring whatever the knob says. The compute side
|
|
independently rejects kernel-datapath NIOs on these containers, so a
|
|
missed detection surfaces as a clearer-late error — except on the switch
|
|
fast path, where an absent anchor is deferred forever by design (a link
|
|
to a stopped node must not fail), making the miss a silently dead cable.
|
|
"""
|
|
|
|
environment = (node.properties or {}).get("environment") or ""
|
|
return "GNS3_UNIX_SOCKET_NIO" in environment or is_iol_runner_environment(environment)
|
|
|
|
|
|
class UDPLink(Link):
|
|
def __init__(self, project, link_id=None):
|
|
super().__init__(project, link_id=link_id)
|
|
self._created = False
|
|
self._link_data = []
|
|
# True when the link could not be created on the computes because one
|
|
# of its endpoints has a missing image. The NIO creation is retried
|
|
# once the image is resolved.
|
|
self._deferred = False
|
|
|
|
@property
|
|
def debug_link_data(self):
|
|
"""
|
|
Use for the debug exports
|
|
"""
|
|
return self._link_data
|
|
|
|
@property
|
|
def deferred(self):
|
|
"""Whether NIO creation is waiting for missing node images."""
|
|
return self._deferred
|
|
|
|
def _get_node_filters(self, node1, node2, kernel=None):
|
|
"""
|
|
Determine which node gets the active filters applied.
|
|
|
|
:param kernel: explicit datapath decision for the caller's context.
|
|
``_prepare`` must pass it: it clears ``_link_data`` (the source
|
|
of the ``kernel_datapath`` property) before rebuilding the NIO
|
|
specs, so deriving the datapath there would read the previous
|
|
run's. Callers operating on established link data (``update``)
|
|
derive it from the property.
|
|
|
|
:returns: Tuple of (node1_filters, node2_filters)
|
|
"""
|
|
# Kernel-datapath links implement suspend natively (interface carrier
|
|
# driven by the NIO suspend flag). The synthetic frequency_drop filter
|
|
# that get_active_filters() injects for suspended links is a
|
|
# relay-datapath emulation mechanism, so kernel links push the *real*
|
|
# filters straight from storage. Both endpoints receive them: each
|
|
# end's compute attaches one tc netem qdisc to its veth host end
|
|
# (egress = traffic entering that container), so every direction of
|
|
# the link is impaired exactly once — the same net effect as the
|
|
# relay, where both directions cross the single filtered bridge.
|
|
if kernel is None:
|
|
kernel = self.kernel_datapath
|
|
if kernel:
|
|
# Ethernet switch fast path: the link's tc impairments apply on
|
|
# the absorbed anchor, which is a single interface — exactly one
|
|
# end may own them, and that end is the switch (it owns the
|
|
# anchor's bridge membership and port VLANs). A cascade link is
|
|
# two interfaces (one veth end per switch), so it takes the
|
|
# normal two-sided treatment like any other kernel link.
|
|
switch = self._switch_endpoint()
|
|
both_switches = (
|
|
self._nodes[0]["node"].node_type == "ethernet_switch"
|
|
and self._nodes[1]["node"].node_type == "ethernet_switch"
|
|
)
|
|
if switch is not None and not both_switches:
|
|
return (self._filters, {}) if switch[0] == 0 else ({}, self._filters)
|
|
return self._filters, self._filters
|
|
# Relay datapath: the netem-extension filters (rate, reorder, gemodel…)
|
|
# have no relay equivalent. update_filters rejects them on created
|
|
# relay links, but a project loaded with such filters on a link that
|
|
# cannot be kernel-wired (cross-compute, non-docker node, relay-only
|
|
# filter mix…) reaches this point — drop what cannot run with a
|
|
# warning, exactly like invalid filters are dropped at load time.
|
|
relay_filters, dropped = split_kernel_only_features(self.get_active_filters())
|
|
if dropped:
|
|
log.warning(
|
|
"Link %s: dropping packet filter(s) %s — no kernel datapath available on this link (uBridge relay wiring)",
|
|
self._id,
|
|
", ".join(sorted(dropped)),
|
|
)
|
|
filter_node = self._get_filter_node()
|
|
return (
|
|
relay_filters if filter_node == node1 else {},
|
|
relay_filters if filter_node == node2 else {},
|
|
)
|
|
|
|
def _markers_for_node(self, node):
|
|
"""
|
|
Marker specs (name -> {bpf, tag, link_id, direction, data_link_type,
|
|
enabled}) for the markers whose capture side is ``node``. Routed by
|
|
capture_node_id so a marker only rides the NIO of the node whose uBridge
|
|
will host it. A disabled marker is included (installed then turned
|
|
``off`` at uBridge, not dropped) so the UI can toggle it instantly
|
|
without an NIO rebuild.
|
|
"""
|
|
return {
|
|
name: {
|
|
"bpf": m["bpf"],
|
|
"tag": m.get("tag"),
|
|
"link_id": self._id,
|
|
"direction": m.get("direction"),
|
|
"data_link_type": m.get("data_link_type", "DLT_EN10MB"),
|
|
"enabled": m.get("enabled", True),
|
|
}
|
|
for name, m in self._markers.items()
|
|
if m.get("capture_node_id") == node.id
|
|
}
|
|
|
|
def _get_node_markers(self, node1, node2):
|
|
"""
|
|
Determine which node gets which markers applied.
|
|
|
|
:returns: Tuple of (node1_markers, node2_markers)
|
|
"""
|
|
return self._markers_for_node(node1), self._markers_for_node(node2)
|
|
|
|
def _kernel_datapath_eligible(self, node1, node2):
|
|
"""
|
|
Whether this link can be wired on the kernel datapath (the adapters'
|
|
host-side anchors enslaved into a per-link Linux bridge) instead of
|
|
the uBridge UDP relay. Every GNS3 filter type has a kernel equivalent
|
|
— netem for delay/packet_loss/corrupt and the netem extensions,
|
|
cls_bpf for bpf, the eBPF stateful classifier for frequency_drop — so
|
|
filters no longer disqualify the kernel path. Capture and markers are
|
|
served by uBridge's AF_PACKET modules on the anchor.
|
|
|
|
Every adapter owns an anchor, so the datapath is a runtime decision —
|
|
links attach to running nodes too: the compute side enslaves the
|
|
anchor (brctl) or attaches the relay to it (add_nio_ethernet) without
|
|
touching the interfaces the node itself uses. Both endpoints must be
|
|
on the same compute (there is no kernel link across hosts) and both
|
|
must be able to anchor: Docker adapters are born as veth pairs,
|
|
QEMU adapters as persistent TAPs, which need uBridge's tap module —
|
|
asked of that compute, since an old uBridge leaves QEMU on the relay
|
|
datapath and cannot carry a kernel link at all. IOU's Ethernet
|
|
bays anchor on persistent TAPs too, bound to its IOL fabric
|
|
(``iol_bridge add_nio_tap``) — a separate capability, since it can
|
|
be present or absent independently of the tap module. Dynamips
|
|
Ethernet slot ports anchor on persistent TAPs the hypervisor opens
|
|
(the tap module again — the Dynamips binary itself creates nothing),
|
|
so it gates like QEMU. A kernel link is an Ethernet segment, so
|
|
non-Ethernet ports (IOU serial, Dynamips serial/ATM/POS adapters)
|
|
stay on the relay whatever the node's capabilities.
|
|
"""
|
|
|
|
if not Config.instance().settings.Server.enable_kernel_datapath:
|
|
return False
|
|
if node1.compute.id != node2.compute.id:
|
|
return False
|
|
if _is_unix_socket_docker(node1) or _is_unix_socket_docker(node2):
|
|
return False
|
|
for side in self._nodes:
|
|
port = side.get("port")
|
|
if port is not None and port.link_type != "ethernet":
|
|
return False
|
|
# Ethernet switch fast path: the switch absorbs the peer's anchor
|
|
# into its own kernel bridge, so the switch side needs no anchoring
|
|
# capability of its own (its bridge and brctl already exist) — only
|
|
# the peer must be able to anchor. Two switches cascade instead: a
|
|
# veth pair joins the two kernel bridges, each side enslaving its
|
|
# own end (a kernel interface belongs to exactly one bridge, so the
|
|
# pair is the only shape a bridge-to-bridge cable can take).
|
|
switch = self._switch_endpoint()
|
|
if switch is not None:
|
|
if node1.node_type == "ethernet_switch" and node2.node_type == "ethernet_switch":
|
|
return True
|
|
peer = node2 if switch[0] == 0 else node1
|
|
return self._kernel_endpoint_ready(peer)
|
|
return self._kernel_endpoint_ready(node1) and self._kernel_endpoint_ready(node2)
|
|
|
|
def _switch_endpoint(self):
|
|
"""
|
|
The Ethernet switch endpoint of this link as ``(index, node)``, or
|
|
None — the fast path where the switch owns the kernel bridge.
|
|
"""
|
|
|
|
for index, side in enumerate(self._nodes):
|
|
if side["node"].node_type == "ethernet_switch":
|
|
return index, side["node"]
|
|
return None
|
|
|
|
@staticmethod
|
|
def _kernel_endpoint_ready(node):
|
|
"""
|
|
Whether a node can anchor a kernel link on its compute.
|
|
"""
|
|
|
|
if node.node_type == "docker":
|
|
return True
|
|
capabilities = node.compute.capabilities or {}
|
|
if node.node_type == "qemu":
|
|
# Strict True: an unreported capability (old uBridge, failed
|
|
# probe) keeps the link on the relay, where it always works.
|
|
return capabilities.get("ubridge_tap") is True
|
|
if node.node_type == "iou":
|
|
return capabilities.get("ubridge_iol_tap") is True
|
|
if node.node_type == "dynamips":
|
|
# The anchors are uBridge-created TAPs the Dynamips hypervisor
|
|
# merely opens (nio create_tap) — the same tap-module capability
|
|
# QEMU needs, not a Dynamips-specific one.
|
|
return capabilities.get("ubridge_tap") is True
|
|
return False
|
|
|
|
async def _prepare(self):
|
|
"""
|
|
Local-only link setup: resolve peer addresses, reserve UDP ports and
|
|
build the two NIO tunnel specs (``self._link_data``). No NIO is sent to
|
|
the computes — the caller decides how to dispatch them (one-by-one via
|
|
:meth:`create`, or batched via the project-open bulk path).
|
|
|
|
:returns: list of two ``(node, adapter_number, port_number, nio_data)``
|
|
tuples, ready to be POSTed to each node's compute.
|
|
"""
|
|
|
|
# Start from a clean slate: reset() re-creates the link on the same
|
|
# object (delete() + create()), and _commit_nios()/update() always
|
|
# address indices 0/1 — appending onto the previous run's entries
|
|
# would re-commit the stale (already released) port pair and orphan
|
|
# the freshly allocated ports.
|
|
self._link_data = []
|
|
|
|
node1 = self._nodes[0]["node"]
|
|
adapter_number1 = self._nodes[0]["adapter_number"]
|
|
port_number1 = self._nodes[0]["port_number"]
|
|
node2 = self._nodes[1]["node"]
|
|
adapter_number2 = self._nodes[1]["adapter_number"]
|
|
port_number2 = self._nodes[1]["port_number"]
|
|
|
|
if self._kernel_datapath_eligible(node1, node2):
|
|
node1_filters, node2_filters = self._get_node_filters(node1, node2, kernel=True)
|
|
node1_markers, node2_markers = self._get_node_markers(node1, node2)
|
|
switch = self._switch_endpoint()
|
|
if switch is not None:
|
|
if node1.node_type == "ethernet_switch" and node2.node_type == "ethernet_switch":
|
|
# Switch-to-switch cascade: one veth pair joins the two
|
|
# kernel bridges, each side enslaving its own end into
|
|
# its own bridge and applying its own port mode to it —
|
|
# the kernel-native shape of a cable between two
|
|
# bridges (an interface belongs to exactly one bridge,
|
|
# so the pair is the only way to express it). The names
|
|
# derive from the link id (utils.kernel_anchor) and both
|
|
# sides carry the other end's name: each creates the
|
|
# pair when its end is missing (the two uBridge
|
|
# processes may race — one create wins, the other's
|
|
# failure is re-checked against the kernel), and either
|
|
# side's teardown may destroy it (deleting one veth end
|
|
# takes the whole pair), so no ordering between the two
|
|
# NIO posts is needed. Filters are two-sided (each end
|
|
# is its own interface, so each direction is impaired
|
|
# exactly once, like any other kernel link — not the
|
|
# single-sided absorbed-anchor case).
|
|
end0, end1 = kernel_cascade_names(self._id)
|
|
self._link_data = [
|
|
{
|
|
"type": "nio_anchor",
|
|
"anchor": end0,
|
|
"peer": end1,
|
|
"filters": node1_filters,
|
|
"markers": node1_markers,
|
|
"suspend": self._suspended,
|
|
},
|
|
{
|
|
"type": "nio_anchor",
|
|
"anchor": end1,
|
|
"peer": end0,
|
|
"filters": node2_filters,
|
|
"markers": node2_markers,
|
|
"suspend": self._suspended,
|
|
},
|
|
]
|
|
return [
|
|
(node1, adapter_number1, port_number1, self._link_data[0]),
|
|
(node2, adapter_number2, port_number2, self._link_data[1]),
|
|
]
|
|
# Ethernet switch fast path: the switch absorbs the peer's
|
|
# anchor into its own kernel bridge — no per-link bridge at
|
|
# all, so the peer end carries no bridge name (its anchor is
|
|
# bridged by the switch; it only keeps its own interface,
|
|
# markers it hosts and its carrier). The switch end gets the
|
|
# anchor's name to join: the name is the shared naming
|
|
# contract (utils.kernel_anchor), a pure function of the
|
|
# peer's node type/id/adapter/port — valid whether or not
|
|
# the peer is currently running, which is what makes links
|
|
# to stopped nodes wireable (the switch joins once the
|
|
# anchor exists, re-pushed by node_started).
|
|
index = switch[0]
|
|
peer_index = 1 - index
|
|
peer = self._nodes[peer_index]["node"]
|
|
peer_anchor = kernel_anchor_name(
|
|
peer.node_type,
|
|
peer.id,
|
|
self._nodes[peer_index]["adapter_number"],
|
|
self._nodes[peer_index]["port_number"],
|
|
)
|
|
peer_data = {
|
|
"type": "nio_bridge",
|
|
"bridge": None,
|
|
"filters": node1_filters if peer_index == 0 else node2_filters,
|
|
"markers": node1_markers if peer_index == 0 else node2_markers,
|
|
"suspend": self._suspended,
|
|
}
|
|
switch_data = {
|
|
"type": "nio_anchor",
|
|
"anchor": peer_anchor,
|
|
"filters": node1_filters if index == 0 else node2_filters,
|
|
"markers": node1_markers if index == 0 else node2_markers,
|
|
"suspend": self._suspended,
|
|
}
|
|
self._link_data = [switch_data, peer_data] if index == 0 else [peer_data, switch_data]
|
|
return [
|
|
(node1, adapter_number1, port_number1, self._link_data[0]),
|
|
(node2, adapter_number2, port_number2, self._link_data[1]),
|
|
]
|
|
# Kernel datapath: each endpoint enslaves its veth host end into
|
|
# a per-link Linux bridge. The name derives from the link id so
|
|
# both ends compute it independently (no coordinator) and crash
|
|
# leftovers sweep deterministically; 11 hex chars fill IFNAMSIZ
|
|
# (15), pushing name-collision probability to irrelevance.
|
|
# Markers ride the NIO like on the relay datapath, routed by
|
|
# capture node; they attach to the veth host end via uBridge's
|
|
# AF_PACKET marker module instead of a relay `mark` filter.
|
|
# Filters become one tc netem qdisc per veth host end (the netem
|
|
# surface and its extensions), plus cls_bpf match-drop (bpf) and
|
|
# the eBPF stateful classifier (frequency_drop, quota) — every
|
|
# GNS3 filter type has a kernel equivalent now.
|
|
bridge_name = "gns3" + self._id.replace("-", "")[:11]
|
|
self._link_data = [
|
|
{
|
|
"type": "nio_bridge",
|
|
"bridge": bridge_name,
|
|
"filters": node1_filters,
|
|
"markers": node1_markers,
|
|
"suspend": self._suspended,
|
|
},
|
|
{
|
|
"type": "nio_bridge",
|
|
"bridge": bridge_name,
|
|
"filters": node2_filters,
|
|
"markers": node2_markers,
|
|
"suspend": self._suspended,
|
|
},
|
|
]
|
|
return [
|
|
(node1, adapter_number1, port_number1, self._link_data[0]),
|
|
(node2, adapter_number2, port_number2, self._link_data[1]),
|
|
]
|
|
|
|
# Get an IP allowing communication between both host
|
|
try:
|
|
(node1_host, node2_host) = await node1.compute.get_ip_on_same_subnet(node2.compute)
|
|
except ValueError as e:
|
|
raise ControllerError(f"Cannot get an IP address on same subnet: {e}")
|
|
|
|
# Reserve a UDP port on both sides in parallel. Pre-allocated ports
|
|
# (used during batch project loading) are popped from memory; otherwise
|
|
# each side falls back to a single HTTP round-trip to its compute.
|
|
async def _allocate_port(compute):
|
|
port = self._project.pop_preallocated_udp_port(compute.id)
|
|
if port is not None:
|
|
return port
|
|
response = await compute.post(f"/projects/{self._project.id}/ports/udp")
|
|
return response.json["udp_port"]
|
|
|
|
self._node1_port, self._node2_port = await asyncio.gather(
|
|
_allocate_port(node1.compute), _allocate_port(node2.compute)
|
|
)
|
|
|
|
node1_filters, node2_filters = self._get_node_filters(node1, node2, kernel=False)
|
|
node1_markers, node2_markers = self._get_node_markers(node1, node2)
|
|
|
|
# Build the tunnel specs for both sides. Index 0 is always node1 so
|
|
# that update()/delete() keep addressing self._link_data[0]/[1].
|
|
self._link_data.append(
|
|
{
|
|
"lport": self._node1_port,
|
|
"rhost": node2_host,
|
|
"rport": self._node2_port,
|
|
"type": "nio_udp",
|
|
"filters": node1_filters,
|
|
"markers": node1_markers,
|
|
"suspend": self._suspended,
|
|
}
|
|
)
|
|
self._link_data.append(
|
|
{
|
|
"lport": self._node2_port,
|
|
"rhost": node1_host,
|
|
"rport": self._node1_port,
|
|
"type": "nio_udp",
|
|
"filters": node2_filters,
|
|
"markers": node2_markers,
|
|
"suspend": self._suspended,
|
|
}
|
|
)
|
|
|
|
return [
|
|
(node1, adapter_number1, port_number1, self._link_data[0]),
|
|
(node2, adapter_number2, port_number2, self._link_data[1]),
|
|
]
|
|
|
|
async def _commit_nios(self, entries):
|
|
"""
|
|
Send the two NIO tunnel POSTs in parallel and roll back on failure.
|
|
|
|
:param entries: the two ``(node, adapter_number, port_number, nio_data)``
|
|
tuples returned by :meth:`_prepare`.
|
|
"""
|
|
|
|
(node1, adapter_number1, port_number1, nio_data1), (node2, adapter_number2, port_number2, nio_data2) = entries
|
|
|
|
# The two ends are independent once the ports and peer addresses are
|
|
# known — each node talks to its own compute/uBridge with no shared
|
|
# lock between them — so the two POSTs overlap. That includes a
|
|
# cascade link: both sides create the veth pair when their end is
|
|
# missing (a race one side is designated to lose benignly), so
|
|
# neither post depends on the other's completion. If either fails,
|
|
# roll back whichever side succeeded before re-raising the first
|
|
# error.
|
|
results = await asyncio.gather(
|
|
node1.post(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=nio_data1, timeout=120),
|
|
node2.post(f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=nio_data2, timeout=120),
|
|
return_exceptions=True,
|
|
)
|
|
errors = [result for result in results if isinstance(result, Exception)]
|
|
if errors:
|
|
cleanup = []
|
|
if not isinstance(results[0], Exception):
|
|
cleanup.append(node1.delete(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", timeout=120))
|
|
if not isinstance(results[1], Exception):
|
|
cleanup.append(node2.delete(f"/adapters/{adapter_number2}/ports/{port_number2}/nio", timeout=120))
|
|
if cleanup:
|
|
await asyncio.gather(*cleanup, return_exceptions=True)
|
|
raise errors[0]
|
|
self._created = True
|
|
|
|
async def create(self):
|
|
"""
|
|
Create the link on the nodes (interactive path: prepare + commit).
|
|
"""
|
|
|
|
entries = await self._prepare()
|
|
await self._commit_nios(entries)
|
|
# New links automatically inherit every active project-level marker
|
|
# definition so the user doesn't have to reconfigure.
|
|
await self._project.apply_defs_to_new_link(self)
|
|
|
|
async def update(self):
|
|
"""
|
|
Update the link on the nodes
|
|
"""
|
|
|
|
if len(self._link_data) == 0:
|
|
return
|
|
node1 = self._nodes[0]["node"]
|
|
node2 = self._nodes[1]["node"]
|
|
|
|
node1_filters, node2_filters = self._get_node_filters(node1, node2)
|
|
node1_markers, node2_markers = self._get_node_markers(node1, node2)
|
|
|
|
adapter_number1 = self._nodes[0]["adapter_number"]
|
|
port_number1 = self._nodes[0]["port_number"]
|
|
self._link_data[0]["filters"] = node1_filters
|
|
self._link_data[0]["markers"] = node1_markers
|
|
self._link_data[0]["suspend"] = self._suspended
|
|
# The Ethernet hub is still Dynamips-hosted (no uBridge of its own and
|
|
# no PUT NIO route) — keep skipping its side. Every other node type,
|
|
# including the brctl Ethernet switch, re-applies via the NIO update.
|
|
if node1.node_type != "ethernet_hub":
|
|
await node1.put(
|
|
f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=self._link_data[0], timeout=120
|
|
)
|
|
|
|
adapter_number2 = self._nodes[1]["adapter_number"]
|
|
port_number2 = self._nodes[1]["port_number"]
|
|
self._link_data[1]["filters"] = node2_filters
|
|
self._link_data[1]["markers"] = node2_markers
|
|
self._link_data[1]["suspend"] = self._suspended
|
|
if node2.node_type != "ethernet_hub":
|
|
await node2.put(
|
|
f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=self._link_data[1], timeout=120
|
|
)
|
|
|
|
async def delete(self):
|
|
"""
|
|
Delete the link and free the resources
|
|
"""
|
|
if not self._created:
|
|
if self._deferred:
|
|
# There is no NIO on the computes to delete, but the local
|
|
# back-references created while loading must be cleared.
|
|
await super().delete()
|
|
return
|
|
try:
|
|
node1 = self._nodes[0]["node"]
|
|
adapter_number1 = self._nodes[0]["adapter_number"]
|
|
port_number1 = self._nodes[0]["port_number"]
|
|
except IndexError:
|
|
return
|
|
try:
|
|
await node1.delete(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", timeout=120)
|
|
# If the node is already deleted (user selected multiple element and delete all in the same time)
|
|
except ControllerNotFoundError:
|
|
pass
|
|
|
|
try:
|
|
node2 = self._nodes[1]["node"]
|
|
adapter_number2 = self._nodes[1]["adapter_number"]
|
|
port_number2 = self._nodes[1]["port_number"]
|
|
except IndexError:
|
|
return
|
|
try:
|
|
await node2.delete(f"/adapters/{adapter_number2}/ports/{port_number2}/nio", timeout=120)
|
|
# If the node is already deleted (user selected multiple element and delete all in the same time)
|
|
except ControllerNotFoundError:
|
|
pass
|
|
await super().delete()
|
|
|
|
async def reset(self):
|
|
"""
|
|
Reset the link.
|
|
"""
|
|
|
|
# recreate the link on the compute
|
|
await self.delete()
|
|
await self.create()
|
|
|
|
async def start_capture(self, data_link_type="DLT_EN10MB", capture_file_name=None, wireshark=False, jwt_token=None):
|
|
"""
|
|
Start capture on a link
|
|
"""
|
|
if not capture_file_name:
|
|
capture_file_name = self.default_capture_file_name()
|
|
self._capture_node = self._choose_capture_side()
|
|
data = {"capture_file_name": capture_file_name, "data_link_type": data_link_type}
|
|
await self._capture_node["node"].post(
|
|
"/adapters/{adapter_number}/ports/{port_number}/capture/start".format(
|
|
adapter_number=self._capture_node["adapter_number"], port_number=self._capture_node["port_number"]
|
|
),
|
|
data=data,
|
|
)
|
|
await super().start_capture(
|
|
data_link_type=data_link_type, capture_file_name=capture_file_name, wireshark=wireshark, jwt_token=jwt_token
|
|
)
|
|
|
|
async def stop_capture(self):
|
|
"""
|
|
Stop capture on a link
|
|
"""
|
|
if self._capture_node:
|
|
await self._capture_node["node"].post(
|
|
"/adapters/{adapter_number}/ports/{port_number}/capture/stop".format(
|
|
adapter_number=self._capture_node["adapter_number"], port_number=self._capture_node["port_number"]
|
|
)
|
|
)
|
|
self._capture_node = None
|
|
await super().stop_capture()
|
|
|
|
def _choose_capture_side(self):
|
|
"""
|
|
Run capture on the best candidate.
|
|
|
|
The ideal candidate is a node who on controller server and always
|
|
running (capture will not be cut off)
|
|
|
|
:returns: Node where the capture should run
|
|
"""
|
|
|
|
for node in self._nodes:
|
|
if (
|
|
node["node"].compute.id == "local"
|
|
and node["node"].node_type in BUILTIN_NODE_TYPES
|
|
and node["node"].status == "started"
|
|
):
|
|
return node
|
|
|
|
for node in self._nodes:
|
|
if node["node"].node_type in BUILTIN_NODE_TYPES and node["node"].status == "started":
|
|
return node
|
|
|
|
for node in self._nodes:
|
|
if node["node"].compute.id == "local" and node["node"].status == "started":
|
|
return node
|
|
|
|
for node in self._nodes:
|
|
if node["node"].node_type and node["node"].status == "started":
|
|
return node
|
|
|
|
raise ControllerError("Cannot capture because there is no running device on this link")
|
|
|
|
def _choose_marker_side(self):
|
|
"""
|
|
Pick the node that will host the marker, mirroring ``_get_filter_node``
|
|
in link.py. Only types with a uBridge bridge (``_MARKER_CAPABLE_TYPES``)
|
|
are eligible. A running node is preferred, but a stopped one is
|
|
accepted — like packet filters, the marker is stored on the NIO and
|
|
applied when the node starts.
|
|
"""
|
|
|
|
# Prefer started.
|
|
for node in self._nodes:
|
|
if node["node"].node_type in _MARKER_CAPABLE_TYPES and node["node"].status == "started":
|
|
return node
|
|
|
|
# Accept stopped but capable (marker rides NIO, applied at start).
|
|
for node in self._nodes:
|
|
if node["node"].node_type in _MARKER_CAPABLE_TYPES:
|
|
return node
|
|
|
|
raise ControllerError("Cannot add marker because no device on this link supports traffic insight")
|
|
|
|
def _node_by_id(self, node_id):
|
|
"""
|
|
Resolve a caller-chosen capture node by id, validating it is an
|
|
endpoint of this link and marker-capable. Used when the caller
|
|
(REST/MCP) explicitly pins the observer side instead of letting
|
|
``_choose_marker_side`` auto-pick.
|
|
|
|
:param node_id: node id (UUID or str) the caller requested
|
|
:returns: a ``self._nodes`` entry (node/adapter_number/port_number)
|
|
"""
|
|
|
|
target = str(node_id)
|
|
for node in self._nodes:
|
|
if str(node["node"].id) != target:
|
|
continue
|
|
if node["node"].node_type not in _MARKER_CAPABLE_TYPES:
|
|
raise ControllerError(
|
|
f"Node {node_id} ({node['node'].node_type}) cannot host a "
|
|
f"marker — no uBridge bridge to attach the filter to"
|
|
)
|
|
return node
|
|
raise ControllerNotFoundError(f"Node {node_id} is not an endpoint of link {self._id}")
|
|
|
|
async def node_updated(self, node):
|
|
"""
|
|
Called when a node member of the link is updated
|
|
"""
|
|
if self._capture_node and node == self._capture_node["node"] and node.status != "started":
|
|
await self.stop_capture()
|
|
# Marker clean-up is *not* done on node stop — markers are a persistent
|
|
# link-scoped feature that recovers via NIO on restart (see
|
|
# _ubridge_apply_markers in add_ubridge_udp_connection). The user
|
|
# explicitly deletes a marker via the REST API, and a marker is torn
|
|
# down automatically only when its link is deleted.
|
|
|
|
async def node_started(self, node):
|
|
"""
|
|
Called after a node of this link reached the started state.
|
|
|
|
The Ethernet switch fast path wires itself on the switch side, but a
|
|
peer's anchor only comes into existence when the peer starts (and a
|
|
peer restart can replace the interface under the same name — e.g.
|
|
QEMU recreates its TAPs at every start). The switch cannot observe
|
|
that, so the controller re-pushes its NIO: the switch completes a
|
|
join it had to defer (or re-joins an anchor that disappeared),
|
|
then re-applies the link state it owns.
|
|
"""
|
|
|
|
if not self._created:
|
|
return
|
|
switch = self._switch_endpoint()
|
|
if switch is None or not self.kernel_datapath:
|
|
return
|
|
both_switches = (
|
|
self._nodes[0]["node"].node_type == "ethernet_switch"
|
|
and self._nodes[1]["node"].node_type == "ethernet_switch"
|
|
)
|
|
# A cascade link has a switch on both ends and each end is its own
|
|
# veth half: either switch starting (recreating its bridge) can
|
|
# leave its own half unjoined, so both ends are re-pushed.
|
|
indices = [0, 1] if both_switches else [switch[0]]
|
|
for index in indices:
|
|
switch_node = self._nodes[index]["node"]
|
|
adapter_number = self._nodes[index]["adapter_number"]
|
|
port_number = self._nodes[index]["port_number"]
|
|
try:
|
|
await switch_node.put(
|
|
f"/adapters/{adapter_number}/ports/{port_number}/nio",
|
|
data=self._link_data[index],
|
|
timeout=120,
|
|
)
|
|
except (ComputeError, ControllerError):
|
|
# The link keeps its stored NIO; the next node start or link
|
|
# update retries. A failure here must not fail the node start.
|
|
log.warning("Could not re-push switch NIO for link %s after node start", self.id)
|
|
|
|
async def start_marker(
|
|
self,
|
|
name,
|
|
bpf,
|
|
tag=None,
|
|
direction=None,
|
|
data_link_type="DLT_EN10MB",
|
|
capture_node_id=None,
|
|
color=None,
|
|
highlight_duration=None,
|
|
enabled=True,
|
|
inherited_from=None,
|
|
dump=True,
|
|
memory_only=False,
|
|
):
|
|
"""
|
|
Attach a traffic-insight marker to this link.
|
|
|
|
State-only model (mirrors ``update_filters``): record the marker in
|
|
``_markers`` (with its capture-side node id for NIO routing), then push
|
|
via ``self.update()`` so it rides the NIO and is applied by
|
|
``_ubridge_apply_markers``. No dedicated uBridge round-trip — exactly
|
|
how packet filters are applied.
|
|
|
|
:param name: stable filter name — echoed in MARK signals + pcap identity
|
|
:param bpf: libpcap BPF expression
|
|
:param tag: optional correlation id
|
|
:param capture_node_id: optional explicit observer node. When set the
|
|
marker is pinned to that endpoint's uBridge (and ``direction`` is
|
|
interpreted from its perspective); validated by ``_node_by_id``.
|
|
Omitted = auto-pick via ``_choose_marker_side``. Ignored for
|
|
inherited markers (project defs are link-agnostic → always auto).
|
|
:param color: optional hex color for the Web UI (e.g. '#ff5722'); stored
|
|
with the link and persisted in the topology, never sent to uBridge
|
|
:param highlight_duration: optional UI-only hint (milliseconds) for how
|
|
long a match keeps the marker highlighted; stored, never sent to uBridge
|
|
:param inherited_from: def name when this marker is a project-level
|
|
inheritance copy; set automatically, never exposed to REST callers
|
|
"""
|
|
|
|
if name in self._markers:
|
|
raise ControllerError(f"Marker '{name}' already exists on link {self._id}")
|
|
|
|
# Validate the BPF only for private per-link markers. An inherited copy
|
|
# (``inherited_from`` set) fans out from a definition whose BPF was
|
|
# already validated once at create/update (and on project load), so
|
|
# re-validating per link would spawn one ``tcpdump -d`` per link for the
|
|
# same expression.
|
|
if not inherited_from:
|
|
result = validate_bpf_syntax(bpf)
|
|
if not result.get("valid"):
|
|
raise ControllerError(f"Invalid BPF expression: {result.get('error', 'unknown error')}")
|
|
|
|
if capture_node_id and not inherited_from:
|
|
marker_side = self._node_by_id(capture_node_id)
|
|
else:
|
|
marker_side = self._choose_marker_side()
|
|
marker_entry = {
|
|
"bpf": bpf,
|
|
"tag": tag,
|
|
"enabled": enabled,
|
|
"color": color,
|
|
"highlight_duration": highlight_duration,
|
|
"capture_node_id": marker_side["node"].id,
|
|
"direction": direction,
|
|
"data_link_type": data_link_type,
|
|
}
|
|
if inherited_from:
|
|
marker_entry["inherited_from"] = inherited_from
|
|
self._markers[name] = marker_entry
|
|
if memory_only:
|
|
# Project-open prepare / marker-def fan-out: only refresh the
|
|
# in-memory NIO specs so a later batch dispatch carries the new
|
|
# markers — no per-link update HTTP, notification or dump.
|
|
self._refresh_link_data()
|
|
return
|
|
if self._created:
|
|
await self.update()
|
|
self._project.emit_notification("link.updated", self.asdict())
|
|
# Bulk fan-out passes dump=False: N per-link topology writes on a
|
|
# 500-link project are the dominant cost — the caller dumps once after.
|
|
if dump:
|
|
self._project.dump()
|
|
|
|
def _refresh_link_data(self):
|
|
"""
|
|
Recompute the filters / markers / suspend fields of ``_link_data`` from
|
|
the current link state without pushing to computes. Used by the
|
|
memory-only marker path so a batch dispatch picks up the new markers.
|
|
"""
|
|
|
|
if len(self._link_data) < 2:
|
|
return
|
|
node1 = self._nodes[0]["node"]
|
|
node2 = self._nodes[1]["node"]
|
|
node1_filters, node2_filters = self._get_node_filters(node1, node2)
|
|
node1_markers, node2_markers = self._get_node_markers(node1, node2)
|
|
self._link_data[0]["filters"] = node1_filters
|
|
self._link_data[0]["markers"] = node1_markers
|
|
self._link_data[0]["suspend"] = self._suspended
|
|
self._link_data[1]["filters"] = node2_filters
|
|
self._link_data[1]["markers"] = node2_markers
|
|
self._link_data[1]["suspend"] = self._suspended
|
|
|
|
async def stop_marker(self, name, inherited=False, dump=True, memory_only=False):
|
|
"""
|
|
Remove a traffic-insight marker from this link.
|
|
|
|
Drop it from ``_markers`` and push via ``self.update()``: the NIO
|
|
reset+reapply in ``_ubridge_apply_filters``/``_ubridge_apply_markers``
|
|
drops it from uBridge. Mirrors how deleting a packet filter works.
|
|
|
|
:param name: filter name to remove
|
|
:param inherited: set by project-level def-delete to bypass the
|
|
inheritance guard (the project layer is the legitimate remover)
|
|
"""
|
|
|
|
if name not in self._markers:
|
|
raise ControllerNotFoundError(f"Marker '{name}' not found on link {self._id}")
|
|
|
|
if self._markers[name].get("inherited_from") and not inherited:
|
|
raise ControllerError(
|
|
f"Marker '{name}' is inherited from the project-level "
|
|
f"definition '{self._markers[name]['inherited_from']}'. "
|
|
"Delete or update it via the marker-definitions API instead."
|
|
)
|
|
|
|
capture_node_id = self._markers[name].get("capture_node_id")
|
|
del self._markers[name]
|
|
if memory_only:
|
|
# Project-level def-delete fan-out: marker is gone from _markers;
|
|
# refresh _link_data so the batch dispatch drops it from uBridge
|
|
# via full reapply. No per-link delete round-trip, notification or dump.
|
|
self._refresh_link_data()
|
|
return
|
|
# Remove the marker filter + its pcap on the capture node directly — NOT a
|
|
# full NIO reapply (which would reset_packet_filters and close/reopen every
|
|
# sibling marker's pcap). delete_packet_filter removes just this filter;
|
|
# the marker is already gone from _markers, so any later reapply (filter
|
|
# change, node restart) won't re-add it either.
|
|
if capture_node_id is not None:
|
|
side = next((s for s in self._nodes if str(s["node"].id) == str(capture_node_id)), None)
|
|
if side is not None:
|
|
try:
|
|
await side["node"].delete(
|
|
f"/adapters/{side['adapter_number']}/ports/{side['port_number']}/markers/{name}",
|
|
params={"link_id": self._id},
|
|
)
|
|
except Exception:
|
|
pass # best-effort: old compute without the route leaves the file
|
|
self._project.emit_notification("link.updated", self.asdict())
|
|
if dump:
|
|
self._project.dump()
|
|
|
|
async def update_marker(
|
|
self,
|
|
name,
|
|
bpf=None,
|
|
tag=None,
|
|
enabled=None,
|
|
direction=_UNSET,
|
|
color=None,
|
|
highlight_duration=None,
|
|
inherited=False,
|
|
dump=True,
|
|
memory_only=False,
|
|
):
|
|
"""
|
|
Update an existing marker's fields and push to uBridge fine-grained — no
|
|
full NIO reapply, so sibling markers' pcaps stay open. bpf/tag/direction
|
|
rebuild just this filter (delete + add); enabled is an instant toggle;
|
|
color/highlight_duration are UI-only (stored, never pushed).
|
|
|
|
:param name: filter name to update
|
|
:param bpf: new BPF expression (None = keep existing)
|
|
:param tag: new tag id (None = keep existing)
|
|
:param enabled: toggle (None = keep existing)
|
|
:param color: new hex color (None = keep existing)
|
|
:param highlight_duration: new UI highlight duration in ms (None = keep existing)
|
|
:param inherited: set by project-level sync to bypass the inheritance
|
|
guard (the project layer is the legitimate editor)
|
|
"""
|
|
|
|
marker_info = self._markers.get(name)
|
|
if not marker_info:
|
|
raise ControllerNotFoundError(f"Marker '{name}' not found on link {self._id}")
|
|
|
|
if marker_info.get("inherited_from") and not inherited:
|
|
raise ControllerError(
|
|
f"Marker '{name}' is inherited from the project-level "
|
|
f"definition '{marker_info['inherited_from']}'. "
|
|
"Update it via the marker-definitions API instead."
|
|
)
|
|
|
|
# Merge every changed field into the marker state first.
|
|
if bpf is not None and bpf != marker_info["bpf"]:
|
|
# An inherited marker is synced from a definition whose BPF was
|
|
# already validated at create/update (or load); re-validating per
|
|
# link is redundant. Private markers validate here as before.
|
|
if not inherited:
|
|
result = validate_bpf_syntax(bpf)
|
|
if not result.get("valid"):
|
|
raise ControllerError(f"Invalid BPF expression: {result.get('error', 'unknown error')}")
|
|
marker_info["bpf"] = bpf
|
|
if tag is not None:
|
|
marker_info["tag"] = tag
|
|
if enabled is not None:
|
|
marker_info["enabled"] = enabled
|
|
if color is not None:
|
|
marker_info["color"] = color
|
|
if highlight_duration is not None:
|
|
marker_info["highlight_duration"] = highlight_duration
|
|
if direction is not _UNSET:
|
|
marker_info["direction"] = direction # None = clear back to both directions
|
|
|
|
if memory_only:
|
|
# Project-level def sync fan-out: state is already merged into
|
|
# _markers; just refresh _link_data so the batch dispatch carries
|
|
# it. No per-link uBridge rebuild, notification or dump.
|
|
self._refresh_link_data()
|
|
return
|
|
|
|
# Push to uBridge fine-grained — NO full NIO reapply (which would
|
|
# reset_packet_filters and close/reopen every sibling marker's pcap):
|
|
# * bpf/tag/direction changed → rebuild just this filter (delete + add),
|
|
# reopening only this marker's pcap (expected, new BPF)
|
|
# * only enabled changed → instant toggle (enable_packet_filter)
|
|
# * only UI fields changed → nothing to push to uBridge
|
|
if self._created:
|
|
ubridge_rebuild = (bpf is not None) or (tag is not None) or (direction is not _UNSET)
|
|
capture_node_id = marker_info.get("capture_node_id")
|
|
side = next((s for s in self._nodes if str(s["node"].id) == str(capture_node_id)), None)
|
|
if side is not None:
|
|
try:
|
|
if ubridge_rebuild:
|
|
await side["node"].put(
|
|
f"/markers/{name}/rebuild",
|
|
data={
|
|
"bpf": marker_info["bpf"],
|
|
"tag": marker_info.get("tag"),
|
|
"direction": marker_info.get("direction"),
|
|
"enabled": marker_info.get("enabled", True),
|
|
"link_id": self._id,
|
|
},
|
|
)
|
|
elif enabled is not None:
|
|
await side["node"].put(f"/markers/{name}", data={"enabled": enabled})
|
|
except Exception:
|
|
# Old compute without the route / node down: state is already
|
|
# correct in _markers; the next NIO reapply converges uBridge.
|
|
pass
|
|
self._project.emit_notification("link.updated", self.asdict())
|
|
if dump:
|
|
self._project.dump()
|