Merge pull request #2993 from yueguobin/feat/multi-kernel-capture

Concurrent kernel captures, one per interface
This commit is contained in:
Jeremy Grossmann 2026-10-15 06:02:58 +02:00 committed by GitHub
commit 716edc3a6a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 157 additions and 171 deletions

View File

@ -243,7 +243,7 @@ The veth pairs already exist at attach time — every adapter is born as a veth
## Capture ## Capture
Kernel links capture via uBridge's AF_PACKET module bound to the veth host end: `capture start_kernel <if> "<pcap>" [dlt]` / `capture stop_kernel`. Single capture per uBridge process (second concurrent → EALREADY): the server tracks which anchor owns the slot, refuses a second port's capture with a clear error instead of leaving it wedged, and only the owning port's stop issues the process-wide `capture stop_kernel`. Start/stop key on the **NIO type**, not on veth presence — a relay NIO riding a veth captures at its relay bridge. Kernel links capture via uBridge's AF_PACKET module bound to the veth host end: `capture start_kernel <if> "<pcap>" [dlt]` / `capture stop_kernel <if>`. One capture per interface, any number of interfaces concurrently — a node can capture several ports at once; a same-interface duplicate answers EALREADY, which the server's per-port `capturing` flag already refuses before uBridge sees it. Stops are per interface too: each port's stop names its own anchor, never touching a sibling port's capture. Start/stop key on the **NIO type**, not on veth presence — a relay NIO riding a veth captures at its relay bridge.
## Markers ## Markers

View File

@ -1057,10 +1057,6 @@ class BaseNode:
# than skipping them as "already installed". # than skipping them as "already installed".
self._marker_filter_bridges.clear() self._marker_filter_bridges.clear()
self._marker_specs.clear() self._marker_specs.clear()
# ...and so is the kernel capture, uBridge's single AF_PACKET slot
# (meaningful for the kernel-datapath node types; see
# KernelDatapathMixin._reserve_kernel_capture).
self._kernel_capture_ifc = None
async def add_ubridge_udp_connection(self, bridge_name, source_nio, destination_nio): async def add_ubridge_udp_connection(self, bridge_name, source_nio, destination_nio):
""" """

View File

@ -768,14 +768,11 @@ class EthernetSwitch(KernelDatapathMixin, BaseNode):
if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running(): if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running():
anchor = self._kernel_ports.get(port_number) anchor = self._kernel_ports.get(port_number)
if anchor is not None: if anchor is not None:
# One capture per uBridge process — see # One capture per interface; a failed start rolls the port's
# _reserve_kernel_capture; a failed start rolls the port's # flag back.
# flag back so it cannot later stop the winner's capture.
try: try:
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(anchor)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
else: else:
@ -797,12 +794,8 @@ class EthernetSwitch(KernelDatapathMixin, BaseNode):
if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running(): if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running():
anchor = self._kernel_ports.get(port_number) anchor = self._kernel_ports.get(port_number)
if anchor is not None: if anchor is not None:
# Process-wide and argument-less: only the port owning the # Per-interface stop: only this port's capture is stopped.
# slot may stop it (a second port's stop would kill this await self._ubridge_send(f"capture stop_kernel {anchor}")
# capture while its own flag still says capturing).
if self._kernel_capture_owned_by(anchor):
await self._ubridge_send("capture stop_kernel")
self._release_kernel_capture(anchor)
else: else:
ubridge_bridge = self._ubridge_bridge_name(port_number) ubridge_bridge = self._ubridge_bridge_name(port_number)
await self._ubridge_send(f"bridge stop_capture {ubridge_bridge}") await self._ubridge_send(f"bridge stop_capture {ubridge_bridge}")

View File

@ -2012,16 +2012,13 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
if isinstance(nio, NIOBridge): if isinstance(nio, NIOBridge):
# Kernel datapath: no relay bridge exists — capture via uBridge's # Kernel datapath: no relay bridge exists — capture via uBridge's
# AF_PACKET module bound to the anchor (a veth host end for the # AF_PACKET module bound to the anchor (a veth host end for the
# standard container, a persistent TAP for an IOL runner). One # standard container, a persistent TAP for an IOL runner).
# capture per uBridge process: the slot is claimed per anchor, # uBridge serves one capture per interface, any number of
# and a failed start rolls the port's capturing flag back so the # interfaces; a failed start rolls the port's capturing flag back.
# port cannot later stop the winner's capture.
host_ifc = self._kernel_host_ifc(adapter_number, port_number) host_ifc = self._kernel_host_ifc(adapter_number, port_number)
try: try:
self._reserve_kernel_capture(host_ifc)
await self._ubridge_send(f'capture start_kernel {host_ifc} "{output_file}"') await self._ubridge_send(f'capture start_kernel {host_ifc} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(host_ifc)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
return return
@ -2043,13 +2040,11 @@ class DockerVM(DockerKernelDatapathMixin, BaseNode):
raise DockerError("Cannot stop the packet capture: uBridge is not running") raise DockerError("Cannot stop the packet capture: uBridge is not running")
nio = self._ethernet_adapters[adapter_number].get_nio(port_number) nio = self._ethernet_adapters[adapter_number].get_nio(port_number)
if isinstance(nio, NIOBridge): if isinstance(nio, NIOBridge):
# Idempotent on the uBridge side (no active capture is OK), but # Per-interface stop: only this port's capture is stopped (the
# process-wide and argument-less: only the port owning the slot # capturing flag above gates it, so uBridge's 208-on-idle-
# may stop it, or a second port's stop would kill this capture. # interface can only surface on a genuine desync).
host_ifc = self._kernel_host_ifc(adapter_number, port_number) host_ifc = self._kernel_host_ifc(adapter_number, port_number)
if self._kernel_capture_owned_by(host_ifc): await self._ubridge_send(f"capture stop_kernel {host_ifc}")
await self._ubridge_send("capture stop_kernel")
self._release_kernel_capture(host_ifc)
return return
bridge_name = self._bridge_name(adapter_number, port_number) bridge_name = self._bridge_name(adapter_number, port_number)
await self._ubridge_send(f"bridge stop_capture {bridge_name}") await self._ubridge_send(f"bridge stop_capture {bridge_name}")

View File

@ -1491,15 +1491,11 @@ class Router(KernelDatapathMixin, BaseNode):
if self.ubridge: if self.ubridge:
anchor = self._kernel_host_ifc(slot_number, port_number) anchor = self._kernel_host_ifc(slot_number, port_number)
if anchor is not None: if anchor is not None:
# One capture per uBridge process — see # One capture per interface; a failed start rolls the
# _reserve_kernel_capture; a failed start rolls the # port's flag back.
# port's flag back so it cannot later stop the winner's
# capture.
try: try:
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(anchor)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
log.debug( log.debug(
@ -1538,11 +1534,9 @@ class Router(KernelDatapathMixin, BaseNode):
if isinstance(nio, NIOBridge): if isinstance(nio, NIOBridge):
nio.stop_packet_capture() nio.stop_packet_capture()
anchor = self._kernel_host_ifc(slot_number, port_number) anchor = self._kernel_host_ifc(slot_number, port_number)
if self.ubridge and anchor is not None and self._kernel_capture_owned_by(anchor): if self.ubridge and anchor is not None:
# Process-wide and argument-less: only the port owning the # Per-interface stop: only this port's capture is stopped.
# slot may stop it. await self._ubridge_send(f"capture stop_kernel {anchor}")
await self._ubridge_send("capture stop_kernel")
self._release_kernel_capture(anchor)
return return
await nio.stop_packet_capture() await nio.stop_packet_capture()

View File

@ -1828,19 +1828,16 @@ class IOUVM(KernelDatapathMixin, BaseNode):
if self.ubridge: if self.ubridge:
if isinstance(nio, NIOBridge): if isinstance(nio, NIOBridge):
# Kernel link: capture on the port's TAP anchor (AF_PACKET, # Kernel link: capture on the port's TAP anchor (AF_PACKET,
# one capture per uBridge process — see _reserve_kernel_capture; # one capture per interface — a failed start rolls the port's
# a failed start rolls the port's flag back so the port cannot # flag back).
# later stop the winner's capture).
anchor = self._kernel_host_ifc(adapter_number, port_number) anchor = self._kernel_host_ifc(adapter_number, port_number)
try: try:
if anchor is None: if anchor is None:
raise self._kernel_error( raise self._kernel_error(
f"Bay {adapter_number}/{port_number} of IOU '{self._name}' has no TAP anchor to capture on" f"Bay {adapter_number}/{port_number} of IOU '{self._name}' has no TAP anchor to capture on"
) )
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(anchor)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
else: else:
@ -1869,13 +1866,10 @@ class IOUVM(KernelDatapathMixin, BaseNode):
log.debug(f'IOU "{self._name}" [{self._id}]: stopping packet capture on {adapter_number}/{port_number}') log.debug(f'IOU "{self._name}" [{self._id}]: stopping packet capture on {adapter_number}/{port_number}')
if self.ubridge: if self.ubridge:
if isinstance(nio, NIOBridge): if isinstance(nio, NIOBridge):
# Process-wide and argument-less: only the port owning the # Per-interface stop: only this port's capture is stopped.
# slot may stop it (a second port's stop would kill this
# capture while its own flag still says capturing).
anchor = self._kernel_host_ifc(adapter_number, port_number) anchor = self._kernel_host_ifc(adapter_number, port_number)
if self._kernel_capture_owned_by(anchor): if anchor is not None:
await self._ubridge_send("capture stop_kernel") await self._ubridge_send(f"capture stop_kernel {anchor}")
self._release_kernel_capture(anchor)
else: else:
await self._ubridge_send( await self._ubridge_send(
f"iol_bridge stop_capture {self._iol_bridge_name()} {adapter_number} {port_number}" f"iol_bridge stop_capture {self._iol_bridge_name()} {adapter_number} {port_number}"

View File

@ -264,52 +264,6 @@ class KernelDatapathMixin:
return not (isinstance(nio, NIOBridge) and nio.bridge is None) return not (isinstance(nio, NIOBridge) and nio.bridge is None)
# The anchor whose kernel capture currently holds the uBridge process's
# single AF_PACKET capture slot (None = free; see _reserve_kernel_capture).
# Class-level default so every node type gets it without touching its
# __init__; BaseNode._stop_ubridge resets it when uBridge goes away.
_kernel_capture_ifc = None
def _reserve_kernel_capture(self, anchor):
"""
Claim uBridge's single kernel-capture slot for *anchor*.
uBridge serves one AF_PACKET capture per process (a static slot: a
second ``capture start_kernel`` answers EALREADY) while the server
models captures per port — a second kernel port of the same node is
refused here with a clear error. Without the claim the loser's port
would stay marked as capturing, and its stop would issue the
process-wide, argument-less ``capture stop_kernel``, killing the
winner's capture while its own flag still says capturing.
"""
if self._kernel_capture_ifc is not None and self._kernel_capture_ifc != anchor:
raise self._kernel_error(
f"Cannot start the packet capture: this node's uBridge already captures "
f"{self._kernel_capture_ifc} (one kernel capture per node); stop that capture first"
)
self._kernel_capture_ifc = anchor
def _release_kernel_capture(self, anchor):
"""
Release the slot if *anchor* holds it: after a stop, or after a
failed start whose port must not stay marked as capturing.
"""
if self._kernel_capture_ifc == anchor:
self._kernel_capture_ifc = None
def _kernel_capture_owned_by(self, anchor):
"""
Whether *anchor* may issue the process-wide ``capture stop_kernel``:
only the port holding the slot — or a node whose slot is empty (a
capture predating the tracker), where stopping is the historical
best-effort behaviour. A second port must never reach it: the
command takes no interface and would stop the owner's capture.
"""
return self._kernel_capture_ifc is None or self._kernel_capture_ifc == anchor
async def _kernel_attach(self, anchor, nio): async def _kernel_attach(self, anchor, nio):
""" """
Attach an anchor to the per-link kernel bridge carried by *nio* Attach an anchor to the per-link kernel bridge carried by *nio*
@ -342,10 +296,7 @@ class KernelDatapathMixin:
await self._ubridge_send(f'brctl addif "{nio.bridge}" "{anchor}"') await self._ubridge_send(f'brctl addif "{nio.bridge}" "{anchor}"')
if nio.capturing: if nio.capturing:
# Restore a capture that was active before a node restart # Restore a capture that was active before a node restart
# (mirrors the relay path's start_capture). The slot is claimed # (mirrors the relay path's start_capture).
# like a fresh start: a second kernel port's replay is refused
# with a clear error instead of an EALREADY deep in the wiring.
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{nio.pcap_output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{nio.pcap_output_file}"')
# Markers carried by the NIO attach to the anchor (AF_PACKET taps) — # Markers carried by the NIO attach to the anchor (AF_PACKET taps) —
# the anchor is the interface, not a relay bridge. # the anchor is the interface, not a relay bridge.

View File

@ -1758,14 +1758,11 @@ class QemuVM(KernelDatapathMixin, BaseNode):
anchor = self._kernel_host_ifc(adapter_number) anchor = self._kernel_host_ifc(adapter_number)
if isinstance(nio, NIOBridge) and anchor is not None: if isinstance(nio, NIOBridge) and anchor is not None:
# Kernel link: capture on the adapter's TAP anchor (AF_PACKET, # Kernel link: capture on the adapter's TAP anchor (AF_PACKET,
# one capture per uBridge process — see _reserve_kernel_capture; # one capture per interface — a failed start rolls the port's
# a failed start rolls the port's flag back so the port cannot # flag back).
# later stop the winner's capture).
try: try:
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(anchor)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
else: else:
@ -1792,12 +1789,8 @@ class QemuVM(KernelDatapathMixin, BaseNode):
if self.ubridge: if self.ubridge:
anchor = self._kernel_host_ifc(adapter_number) anchor = self._kernel_host_ifc(adapter_number)
if isinstance(nio, NIOBridge) and anchor is not None: if isinstance(nio, NIOBridge) and anchor is not None:
# Process-wide and argument-less: only the port owning the # Per-interface stop: only this port's capture is stopped.
# slot may stop it (a second port's stop would kill this await self._ubridge_send(f"capture stop_kernel {anchor}")
# capture while its own flag still says capturing).
if self._kernel_capture_owned_by(anchor):
await self._ubridge_send("capture stop_kernel")
self._release_kernel_capture(anchor)
else: else:
await self._ubridge_send("bridge stop_capture {name}".format(name=f"QEMU-{self._id}-{adapter_number}")) await self._ubridge_send("bridge stop_capture {name}".format(name=f"QEMU-{self._id}-{adapter_number}"))

View File

@ -659,14 +659,11 @@ class VPCSVM(KernelDatapathMixin, BaseNode):
anchor = self._kernel_host_ifc(0, port_number) anchor = self._kernel_host_ifc(0, port_number)
if isinstance(nio, NIOBridge) and anchor is not None: if isinstance(nio, NIOBridge) and anchor is not None:
# Kernel link: capture on the adapter's TAP anchor (AF_PACKET, # Kernel link: capture on the adapter's TAP anchor (AF_PACKET,
# one capture per uBridge process — see _reserve_kernel_capture; # one capture per interface — a failed start rolls the port's
# a failed start rolls the port's flag back so the port cannot # flag back).
# later stop the winner's capture).
try: try:
self._reserve_kernel_capture(anchor)
await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"') await self._ubridge_send(f'capture start_kernel {anchor} "{output_file}"')
except Exception: except Exception:
self._release_kernel_capture(anchor)
nio.stop_packet_capture() nio.stop_packet_capture()
raise raise
else: else:
@ -693,12 +690,8 @@ class VPCSVM(KernelDatapathMixin, BaseNode):
if self.ubridge: if self.ubridge:
anchor = self._kernel_host_ifc(0, port_number) anchor = self._kernel_host_ifc(0, port_number)
if isinstance(nio, NIOBridge) and anchor is not None: if isinstance(nio, NIOBridge) and anchor is not None:
# Process-wide and argument-less: only the port owning the # Per-interface stop: only this port's capture is stopped.
# slot may stop it (a second port's stop would kill this await self._ubridge_send(f"capture stop_kernel {anchor}")
# capture while its own flag still says capturing).
if self._kernel_capture_owned_by(anchor):
await self._ubridge_send("capture stop_kernel")
self._release_kernel_capture(anchor)
else: else:
await self._ubridge_send("bridge stop_capture {name}".format(name=f"VPCS-{self._id}")) await self._ubridge_send("bridge stop_capture {name}".format(name=f"VPCS-{self._id}"))

View File

@ -792,7 +792,7 @@ class TestEthernetSwitchNodesRoutes:
) )
) )
assert response.status_code == status.HTTP_204_NO_CONTENT assert response.status_code == status.HTTP_204_NO_CONTENT
node._ubridge_send.assert_any_call("capture stop_kernel") node._ubridge_send.assert_any_call(f"capture stop_kernel {anchor}")
async def test_ethernet_switch_batch_create_anchor_nio( async def test_ethernet_switch_batch_create_anchor_nio(
self, app: FastAPI, compute_client: AsyncClient, compute_project: Project, ethernet_switch: dict self, app: FastAPI, compute_client: AsyncClient, compute_project: Project, ethernet_switch: dict

View File

@ -997,14 +997,16 @@ async def test_stop_capture_kernel(vm):
await vm.stop_capture(0) await vm.stop_capture(0)
assert nio.capturing is False assert nio.capturing is False
vm._ubridge_send.assert_any_call("capture stop_kernel") vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc}")
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_start_capture_second_concurrent_fails(vm): async def test_start_capture_ubridge_ealready_rolls_the_flag_back(vm):
""" """
uBridge's kernel capture is a singleton per process: a second concurrent uBridge answers EALREADY when its interface is already being captured
capture returns EALREADY and must surface to the caller. (the server's per-port flag should have refused earlier; this is the
desync net): the error must surface and the port must not stay marked
as capturing.
""" """
async def send(command): async def send(command):
@ -1022,19 +1024,16 @@ async def test_start_capture_second_concurrent_fails(vm):
await vm.start_capture(0, "/tmp/capture.pcap") await vm.start_capture(0, "/tmp/capture.pcap")
# the failed start must not leave the port marked as capturing: a stuck # the failed start must not leave the port marked as capturing: a stuck
# flag refuses the retry and lets this port's stop kill the winner # flag refuses the retry
assert nio.capturing is False assert nio.capturing is False
assert vm._kernel_capture_ifc is None
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_start_capture_second_port_on_the_same_node_is_refused(vm): async def test_capture_two_ports_at_once_and_stop_them_independently(vm):
""" """
uBridge serves one kernel capture per process while the server models uBridge serves one kernel capture per interface: two ports of the same
captures per port: a second port's capture must be refused with a clear node capture concurrently, and each port's stop names only its own
error and its flag rolled back — otherwise its stop would issue the anchor — a sibling port's capture is never touched.
process-wide, argument-less ``capture stop_kernel`` and kill the first
port's capture while its own flag still says capturing.
""" """
vm._ubridge_hypervisor = MagicMock() vm._ubridge_hypervisor = MagicMock()
@ -1051,48 +1050,22 @@ async def test_start_capture_second_port_on_the_same_node_is_refused(vm):
vm._ethernet_adapters[1].add_nio(0, nio2) vm._ethernet_adapters[1].add_nio(0, nio2)
await vm.start_capture(0, "/tmp/capture.pcap") await vm.start_capture(0, "/tmp/capture.pcap")
await vm.start_capture(1, "/tmp/capture2.pcap")
with pytest.raises(DockerError, match="one kernel capture per node"):
await vm.start_capture(1, "/tmp/capture2.pcap")
assert nio1.capturing is True assert nio1.capturing is True
assert nio2.capturing is False assert nio2.capturing is True
vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc1} "/tmp/capture.pcap"') vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc1} "/tmp/capture.pcap"')
assert not any("stop_kernel" in str(c) for c in vm._ubridge_send.call_args_list) vm._ubridge_send.assert_any_call(f'capture start_kernel {host_ifc2} "/tmp/capture2.pcap"')
@pytest.mark.asyncio
async def test_stop_capture_on_a_port_that_does_not_own_the_slot_skips_stop_kernel(vm):
"""
``capture stop_kernel`` is process-wide and takes no interface: a stop on
a port whose flag says capturing while another port owns the slot must
not reach uBridge — it would stop the owner's capture.
"""
vm._ubridge_hypervisor = MagicMock()
vm._ubridge_send = AsyncioMagicMock()
vm.status = "started"
vm._ethernet_adapters.append(type(vm._ethernet_adapters[0])())
host_ifc1, _ = vm._veth_names(0, 0)
host_ifc2, _ = vm._veth_names(1, 0)
vm._kernel_veths[(0, 0)] = host_ifc1
vm._kernel_veths[(1, 0)] = host_ifc2
nio1 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
nio2 = vm.manager.create_nio({"type": "nio_bridge", "bridge": BRIDGE})
vm._ethernet_adapters[0].add_nio(0, nio1)
vm._ethernet_adapters[1].add_nio(0, nio2)
await vm.start_capture(0, "/tmp/capture.pcap") # port 0 owns the slot
nio2.start_packet_capture("/tmp/capture2.pcap") # a stale second flag
await vm.stop_capture(1) await vm.stop_capture(1)
assert nio1.capturing is True
assert nio2.capturing is False assert nio2.capturing is False
assert not any("stop_kernel" in str(c) for c in vm._ubridge_send.call_args_list) vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc2}")
assert not any(f"stop_kernel {host_ifc1}" in str(c) for c in vm._ubridge_send.call_args_list)
# the owner's capture is untouched and still stoppable
await vm.stop_capture(0) await vm.stop_capture(0)
vm._ubridge_send.assert_any_call("capture stop_kernel") vm._ubridge_send.assert_any_call(f"capture stop_kernel {host_ifc1}")
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@ -581,7 +581,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(router):
assert not any("bind_filter" in c for c in _hypervisor_commands(router)) assert not any("bind_filter" in c for c in _hypervisor_commands(router))
await router.stop_capture(0, 0) await router.stop_capture(0, 0)
router._ubridge_send.assert_any_call("capture stop_kernel") router._ubridge_send.assert_any_call(f"capture stop_kernel {TAP00}")
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@ -371,7 +371,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(vm):
vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP00} "/tmp/test.pcap"') vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP00} "/tmp/test.pcap"')
await vm.stop_capture(0, 0) await vm.stop_capture(0, 0)
vm._ubridge_send.assert_any_call("capture stop_kernel") vm._ubridge_send.assert_any_call(f"capture stop_kernel {TAP00}")
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@ -328,7 +328,7 @@ async def test_capture_on_a_kernel_link_uses_the_anchor(vm):
vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP0} "/tmp/test.pcap"') vm._ubridge_send.assert_any_call(f'capture start_kernel {TAP0} "/tmp/test.pcap"')
await vm.stop_capture(0) await vm.stop_capture(0)
vm._ubridge_send.assert_any_call("capture stop_kernel") vm._ubridge_send.assert_any_call(f"capture stop_kernel {TAP0}")
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@ -316,6 +316,110 @@ def test_docker_kernel_fast_path():
assert harness.wait_until(lambda: harness.bridge_members(bridge) is None, timeout=10), bridge assert harness.wait_until(lambda: harness.bridge_members(bridge) is None, timeout=10), bridge
def test_docker_kernel_ten_concurrent_captures():
"""Ten kernel captures at once on one node's uBridge. The capture side
picks the first local started node, so all ten links capture on node 1's
single uBridge process — the scale the old one-capture-per-process limit
could never reach. Each link gets its own traffic (its own subnet on its
own eth), and a partial stop must leave the untouched captures running."""
server = harness.live_server(kernel=True)
compute = server.compute
image = harness.ensure_docker_image(server)
project = compute.create_project("docker-multicap-e2e")
pid = project["project_id"]
try:
daemon = harness.DockerDaemon()
nodes = []
for suffix, address in (("1", R1_IP), ("2", R2_IP)):
node = compute.create_docker_node(pid, f"MC-{suffix}", image, adapters=10)
harness.configure_docker_interfaces(node, address)
nodes.append(_resolve_container(daemon, node))
n1, n2 = nodes
c1, c2 = n1["container_id"], n2["container_id"]
n1_id, n2_id = n1["node_id"], n2["node_id"]
links = [compute.create_link(pid, (n1_id, i, 0), (n2_id, i, 0)) for i in range(10)]
assert all(link["kernel_datapath"] is True for link in links), links
print(".. starting containers")
for node in nodes:
compute.call("POST", f"/projects/{pid}/nodes/{node['node_id']}/start")
harness.docker_wait_address(daemon, c1, R1_IP)
harness.docker_wait_address(daemon, c2, R2_IP)
# ten captures before any of the per-link interfaces is addressed, so
# every later frame (up to and including the ARP each first ping
# triggers) lands in a pcap
captures = [
compute.call(
"POST", f"/projects/{pid}/links/{link['link_id']}/capture/start", {"data_link_type": "DLT_EN10MB"}
)
for link in links
]
# eth1..eth9: one subnet each, wired by hand inside the guests; the
# settle window lets the guests' address-add chatter (each v4 addr
# fires two IGMP reports ~1 s apart) land inside an open capture
# instead of a later snapshot
for i in range(1, 10):
for container_id, host in ((c1, 1), (c2, 2)):
code, output = daemon.exec(
container_id,
["sh", "-c", f"ip link set eth{i} up && ip addr add 10.20.{i}.{host}/24 dev eth{i}"],
)
assert code == 0, output
time.sleep(2)
def ping_link(i):
# link 0 rides the init.sh-addressed eth0; the rest their own subnet
if i == 0:
result = harness.docker_ping(daemon, c1, R2_IP, count=2)
assert result["loss"] == 0, result["raw"]
else:
code, output = daemon.exec(c1, ["ping", "-c", "2", "-W", "1", f"10.20.{i}.2"])
assert code == 0, output
for i in range(10):
ping_link(i)
time.sleep(0.5)
paths = [capture["capture_file_path"] for capture in captures]
first_counts = [harness.pcap_records(path)[0] for path in paths]
# 2 ARP + 2 echo requests + 2 replies is the floor per link
assert all(count >= 4 for count in first_counts), list(zip(paths, first_counts, strict=True))
print(f".. all 10 pcaps captured concurrently: {first_counts}")
# stop half. A stop is synchronous (the reply follows the reader
# thread's join), so from its return that pcap can no longer grow —
# the post-stop counts are an exact-freeze baseline
for link in links[5:]:
compute.call("POST", f"/projects/{pid}/links/{link['link_id']}/capture/stop")
stopped_base = [harness.pcap_records(paths[i])[0] for i in range(5, 10)]
# fresh traffic on ALL segments: the stopped half must stay frozen
# even while its links carry frames, the live half must record them
for i in range(10):
ping_link(i)
time.sleep(0.5)
for i in range(5, 10):
assert harness.pcap_records(paths[i])[0] == stopped_base[i - 5], f"stopped pcap {i} grew"
for i in range(5):
assert harness.pcap_records(paths[i])[0] >= first_counts[i] + 4, f"live pcap {i} did not grow"
print(".. per-interface stop froze exactly the stopped half")
for link in links[:5]:
compute.call("POST", f"/projects/{pid}/links/{link['link_id']}/capture/stop")
assert all(
compute.call("GET", f"/projects/{pid}/links/{link['link_id']}")["capturing"] is False for link in links
)
except BaseException:
harness.release(server, pid, failed=True)
raise
harness.release(server, pid, failed=False)
def test_docker_kernel_filter_matrix(): def test_docker_kernel_filter_matrix():
"""One filter type at a time on a live kernel link, each asserted by its """One filter type at a time on a live kernel link, each asserted by its
effect on real traffic and its tc state. Types the uBridge on this effect on real traffic and its tc state. Types the uBridge on this