test(e2e): live coverage for markers on the kernel datapath

Two Alpine containers on a kernel-only link with traffic-insight markers
attached to a veth anchor (marker add_kernel), observed on the dedicated
marker WebSocket via a new harness piece (WebSocketCollector, a sync
facade over websockets' sync client). Real ICMP proves: the marker.match
identity is complete (filter/tag/link_id/capture node/ts/len and the
tx/rx direction set of a ping), the main project channel carries no
matches, the BPF and direction are enforced (a non-matching sibling and
a tx-only marker each see exactly their traffic), the host pcap holds
exactly the matched frames, pausing silences one marker while a sibling
keeps counting (silence is never a dead channel), tag replay serves the
paused tag's timeline with display-filter/link/window narrowing and
refuses with 409 while capturing, and deleting a marker takes its pcap
and signals with it.

docs/features/marker-tag-replay.md

@@ -314,6 +314,12 @@ All error bodies are `{"message": "…"}` (the app's unified format).
 
 ## Notes
 
+- **In-repo e2e coverage (2026-10).** `tests/e2e/test_marker_kernel_datapath.py` exercises
+  the whole contract against a paused tag on real marker pcaps: the `range` aggregate
+  (frame_count, sources inventory with the node/link/marker identity, first-frame
+  src/dst/proto from sharkd), the display filter narrowing (`icmp.type == 8`), the
+  unknown-link empty stream, the `frames` window endpoint, and the 409 gate while a
+  marker under the tag is still capturing.
 - **Heterogeneous link types coexist.** Frames are never merged into a single pcap
   (mergecap is deliberately not used) — each frame carries its source and is decoded
   individually, so Ethernet and serial (cHDLC/PPP) markers can share one timeline.

docs/features/marker-traffic-insight.md

@@ -344,6 +344,15 @@ direction relative to the capture node; see [Direction](#direction).
 
 ## Notes
 
+- **Live e2e coverage.** `tests/e2e/test_marker_kernel_datapath.py` drives two real
+  Alpine containers over a kernel-only link with markers attached to a veth anchor
+  (`marker add_kernel`): it asserts the exact `marker.match` identity on the dedicated
+  marker WebSocket (filter/tag/link_id/node_id/ts/len and the tx/rx direction set of a
+  ping), that the main project channel carries no `marker.match`, that a non-matching
+  BPF and a `tx`-directioned sibling each see exactly their intended traffic, the host
+  pcap frame count, pause/resume silence counting against a still-firing sibling,
+  marker delete removing the pcap and silencing signals, and the empty marker
+  directory at the end.
 - **Marker name is immutable.** It is the identifier across the controller, the uBridge
   filter, the pcap filename, and `MARK` signal routing — so rename is a delete + recreate,
   not a field update. PUT ignores the body `name`; the `{name}` path parameter identifies

tests/e2e/README.md

@@ -47,10 +47,11 @@ fatal: the routers simply run without one).
 
 `harness.py` holds the shared pieces — REST client, server lifecycle, the
 IOS console driver, the Docker guest driver (a tiny Docker Engine API
-client over the daemon socket, plus the digest-pinned image helper),
+client over the daemon socket, plus the digest-pinned image helper), a
-host-side kernel inspections (bridge membership, tap admin state, tc
+WebSocket notification collector (a sync facade over `websockets`' sync
-qdiscs, the L2-anchor spec §E assertions) — so a new node type adds only
+client for asserting on live notification streams), host-side kernel
-its own scenario module.
+inspections (bridge membership, tap admin state, tc qdiscs, the L2-anchor
+spec §E assertions) — so a new node type adds only its own scenario module.
 
 Current scenarios:
 
@@ -93,6 +94,15 @@ Current scenarios:
   each measured through real traffic), plus a relay negative control and a
   relay→kernel reopen upgrade (the server's datapath choice is flipped
   across a restart).
+* `test_marker_kernel_datapath.py` — two real Alpine containers on a
+  kernel-only link with traffic-insight markers attached to a veth anchor
+  (`marker add_kernel`): `marker.match` signals collected off the dedicated
+  marker WebSocket with their full identity (filter/tag/link_id/node_id/len
+  and the tx/rx direction set of a ping), the main project channel asserted
+  free of matches, BPF and direction enforcement, host pcap exactness,
+  pause/resume silence counting against a still-firing sibling, tag replay
+  (timeline, display filter, link narrowing, window, the 409 gate) and
+  marker delete taking pcap plus signals with it.
 
 The Docker scenarios run a digest-pinned `alpine:3`, so every developer
 tests the same bytes: a cached copy in the local Docker daemon costs no

tests/e2e/harness.py

@@ -62,12 +62,14 @@ import socket
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import urllib.error
 import urllib.parse
 import urllib.request
 
 import pytest
+from websockets.sync.client import connect as ws_connect
 
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
@@ -468,6 +470,70 @@ def release(server, project_id, failed=False):
         server.stop()
 
 
+# ---------------------------------------------------------------------------
+# WebSocket notification collector
+# ---------------------------------------------------------------------------
+
+
+class WebSocketCollector:
+    """Collects server notification messages from one WebSocket.
+
+    The e2e tests are synchronous; this is a sync facade over ``websockets``'
+    sync client: a background thread drains the socket into a list and the
+    test polls :meth:`events` / :meth:`wait_events`. Messages are the
+    server's ``{"action": ..., "event": ...}`` JSON objects.
+    """
+
+    def __init__(self, server, path, open_timeout=15):
+        base = server.compute.url.replace("https", "wss", 1).replace("http", "ws", 1)
+        self._ws = ws_connect(f"{base}{path}?token={server.compute.token}", open_timeout=open_timeout)
+        self._messages = []
+        self._stop = threading.Event()
+        self._thread = threading.Thread(target=self._pump, daemon=True)
+        self._thread.start()
+
+    def _pump(self):
+        while not self._stop.is_set():
+            try:
+                raw = self._ws.recv(timeout=0.5)
+            except TimeoutError:
+                continue
+            except Exception:
+                break  # socket closed under us: stop draining
+            try:
+                self._messages.append(json.loads(raw))
+            except ValueError:
+                continue
+
+    def messages(self, action=None):
+        """Every message received so far, optionally only for one action."""
+        messages = list(self._messages)
+        if action is not None:
+            messages = [m for m in messages if m.get("action") == action]
+        return messages
+
+    def events(self, action):
+        """The ``event`` payload of every message carrying the action."""
+        return [m.get("event") for m in self.messages(action)]
+
+    def wait_events(self, action, count=1, timeout=10):
+        """Poll until at least *count* events of the action arrived (or the
+        timeout passes); returns whatever was collected either way."""
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            events = self.events(action)
+            if len(events) >= count:
+                return events
+            time.sleep(0.2)
+        return self.events(action)
+
+    def close(self):
+        self._stop.set()
+        with contextlib.suppress(Exception):
+            self._ws.close()
+        self._thread.join(timeout=5)
+
+
 # ---------------------------------------------------------------------------
 # Dynamips idle-PC (CPU)
 # ---------------------------------------------------------------------------

tests/e2e/test_marker_kernel_datapath.py

@@ -0,0 +1,280 @@
+#
+# 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/>.
+#
+
+"""
+Live end-to-end test for traffic-insight markers on the kernel datapath
+(pytest marker ``e2e``): two real Alpine containers on a kernel-only link,
+markers attached to a veth anchor (``marker add_kernel``) and observed on
+the dedicated marker notification WebSocket. Real ICMP proves:
+
+* a marker match arrives as a ``marker.match`` on
+  ``/projects/{id}/notifications/markers/ws`` with the full identity —
+  filter name, tag, link_id, capture node, timestamp, packet length,
+  travel direction — and **nothing** on the main project channel (the
+  whole point of the dedicated stream: matches must not head-of-line-block
+  topology events);
+* the BPF is enforced on the anchor: a sibling marker whose expression
+  cannot match the traffic stays silent, a directioned marker only sees
+  its direction (``tx`` = the capture node sending), and the pcap on the
+  host holds exactly the matched frames;
+* pausing a marker silences its signals while keeping its pcap; resuming
+  restores them — and other markers' matches keep flowing throughout, so
+  silence is never just "the channel died";
+* tag replay serves the paused tag's aggregate timeline off that pcap
+  (merged frames with src/dst/proto, per-source inventory, display-filter
+  and link narrowing, the window endpoint) and refuses with 409 while any
+  marker under the tag is still capturing (skipped when sharkd is absent);
+* deleting a marker removes its pcap and silences its signals; deleting
+  the project leaves no marker state behind.
+"""
+
+import json
+import os
+import shutil
+import urllib.error
+import urllib.parse
+import urllib.request
+
+import pytest
+
+from tests.e2e import harness
+
+pytestmark = pytest.mark.e2e
+
+# every standard Docker adapter has a single port
+ETH = (0, 0)
+R1_IP = "10.1.1.1"
+R2_IP = "10.1.1.2"
+TAG = 4242
+
+# a busybox `ping` default frame: 14 Ethernet + 20 IP + 8 ICMP + 56 data.
+PING_FRAME_LEN = 98
+
+
+def _resolve_container(daemon, node):
+    """Fill in the node's container id from the daemon (the controller's
+    node payload does not carry the compute-side id), by the deterministic
+    name the server gives the container."""
+    container_id = daemon.container_id(f"GNS3.{node['name']}.{node['project_id']}")
+    assert container_id, f"container for {node['name']} not found on the Docker daemon"
+    node["container_id"] = container_id
+    return node
+
+
+def _create_topology(server, pid, image):
+    compute = server.compute
+    daemon = harness.DockerDaemon()
+    nodes = []
+    for suffix, address in (("1", R1_IP), ("2", R2_IP)):
+        node = compute.create_docker_node(pid, f"MARKER-E2E-{suffix}", image)
+        harness.configure_docker_interfaces(node, address)
+        nodes.append(_resolve_container(daemon, node))
+    return nodes[0], nodes[1], daemon
+
+
+def _start_topology(compute, pid, daemon, nodes):
+    """Start the containers and wait for init.sh to have applied their
+    addresses (busybox ifup runs a moment after container start)."""
+    for node in nodes:
+        compute.call("POST", f"/projects/{pid}/nodes/{node['node_id']}/start")
+    for node, address in zip(nodes, (R1_IP, R2_IP), strict=False):
+        harness.docker_wait_address(daemon, node["container_id"], address)
+
+
+def _raw(compute, path):
+    """GET without raising on an error status: returns ``(status, body)``
+    (body parsed as JSON when possible, else the raw text). The replay
+    endpoints are asserted on their status codes (409 gate / 200 data)."""
+    # S310: the base URL comes from the operator (GNS3_E2E_URL or the
+    # isolated instance this test just started), not untrusted input
+    req = urllib.request.Request(compute.url + path, method="GET")  # noqa: S310
+    req.add_header("Authorization", f"Bearer {compute.token}")
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:  # noqa: S310
+            payload = resp.read()
+            return resp.status, json.loads(payload) if payload else None
+    except urllib.error.HTTPError as e:
+        return e.code, e.read().decode(errors="replace")
+
+
+def _matches(marker_ws, marker_name):
+    return [e for e in marker_ws.events("marker.match") if e["filter"] == marker_name]
+
+
+def _close(*collectors):
+    for collector in collectors:
+        if collector is not None:
+            collector.close()
+
+
+def test_marker_kernel_datapath():
+    server = harness.live_server(kernel=True)
+    compute = server.compute
+    image = harness.ensure_docker_image(server)
+
+    project = compute.create_project("marker-e2e")
+    pid = project["project_id"]
+    marker_ws = main_ws = None
+    try:
+        n1, n2, daemon = _create_topology(server, pid, image)
+        n1_id, n2_id = n1["node_id"], n2["node_id"]
+        c1 = n1["container_id"]
+
+        link = compute.create_link(pid, (n1_id, *ETH), (n2_id, *ETH))
+        lid = link["link_id"]
+        assert link["kernel_datapath"] is True, link
+        _start_topology(compute, pid, daemon, (n1, n2))
+        baseline = harness.docker_wait_ping(daemon, c1, R2_IP)
+        assert baseline["loss"] == 0, baseline["raw"]
+
+        a1 = harness.docker_anchor_name(n1_id, *ETH)
+        markers_dir = os.path.join(compute.call("GET", f"/projects/{pid}")["path"], "project-files", "markers")
+
+        # Both channels open before the markers exist so no match is missed.
+        marker_ws = harness.WebSocketCollector(server, f"/projects/{pid}/notifications/markers/ws")
+        main_ws = harness.WebSocketCollector(server, f"/projects/{pid}/notifications/ws")
+
+        print(".. attaching markers to the kernel link")
+        m = compute.call(
+            "POST",
+            f"/projects/{pid}/links/{lid}/markers",
+            {"name": "m-icmp", "bpf": "icmp", "tag": TAG, "capture_node_id": n1_id},
+        )
+        assert m["capture_node_id"] == n1_id and m["tag"] == TAG and m["enabled"] is True, m
+        # a sibling whose expression cannot match ICMP traffic — it must stay
+        # silent while the others fire (the BPF is the filter, not the marker)
+        compute.call(
+            "POST", f"/projects/{pid}/links/{lid}/markers", {"name": "m-tcp", "bpf": "tcp port 9999", "tag": 4343}
+        )
+        # direction is relative to the capture node: node1 sending only
+        m_tx = compute.call(
+            "POST",
+            f"/projects/{pid}/links/{lid}/markers",
+            {"name": "m-tx", "bpf": "icmp", "direction": "tx", "tag": 4344, "capture_node_id": n1_id},
+        )
+        assert m_tx["direction"] == "tx", m_tx
+
+        print(".. sending 5 ICMP echoes and collecting marker.match signals")
+        result = harness.docker_ping(daemon, c1, R2_IP, count=5)
+        assert result["loss"] == 0, result["raw"]
+
+        harness.wait_until(lambda: len(_matches(marker_ws, "m-icmp")) >= 10, timeout=10)
+        icmp = _matches(marker_ws, "m-icmp")
+        assert len(icmp) == 10, len(icmp)  # 5 requests (tx) + 5 replies (rx)
+        assert not _matches(marker_ws, "m-tcp"), "a non-matching BPF must stay silent"
+        assert len(_matches(marker_ws, "m-tx")) == 5, "the tx direction sees requests only"
+
+        for event in icmp:
+            assert event["node_id"] == n1_id, event
+            assert event["link_id"] == lid, event
+            assert event["tag"] == TAG, event
+            assert event["ts"] > 0, event
+            assert event["len"] == PING_FRAME_LEN, event
+        assert {e["dir"] for e in icmp} == {"tx", "rx"}, [e["dir"] for e in icmp]
+        assert {e["dir"] for e in _matches(marker_ws, "m-tx")} == {"tx"}
+
+        # The dedicated channel carries matches; the main project stream must
+        # not (and must still be alive — its link.updated proves the socket).
+        assert main_ws.messages("marker.match") == [], main_ws.messages("marker.match")[:3]
+        assert main_ws.messages("link.updated"), "the main channel is not receiving at all"
+
+        # Host pcap: exactly the matched frames, off the anchor
+        pcap = os.path.join(markers_dir, f"{n1_id}_{lid}_m-icmp.pcap")
+        assert os.path.exists(pcap), os.listdir(markers_dir)
+        count, ethertypes = harness.pcap_records(pcap)
+        assert count == 10, count
+        assert set(ethertypes) == {"0800"}, ethertypes
+
+        # pause: this marker's signals stop, its pcap survives, and the
+        # sibling marker's matches keep flowing (the channel is alive —
+        # silence is the pause, not a dead stream)
+        print(".. pausing m-icmp: its signals stop while the siblings keep firing")
+        compute.call("PUT", f"/projects/{pid}/links/{lid}/markers/m-icmp", {"enabled": False})
+        harness.docker_ping(daemon, c1, R2_IP, count=3)
+        harness.wait_until(lambda: len(_matches(marker_ws, "m-tx")) >= 8, timeout=10)
+        assert len(_matches(marker_ws, "m-icmp")) == 10, "a paused marker must not signal"
+        assert os.path.exists(pcap)
+
+        # resume: signals come back, pcap keeps appending
+        print(".. resuming m-icmp")
+        compute.call("PUT", f"/projects/{pid}/links/{lid}/markers/m-icmp", {"enabled": True})
+        harness.docker_ping(daemon, c1, R2_IP, count=2)
+        harness.wait_until(lambda: len(_matches(marker_ws, "m-icmp")) >= 14, timeout=10)
+        assert len(_matches(marker_ws, "m-icmp")) == 14
+        assert harness.pcap_records(pcap)[0] == 14
+
+        if shutil.which("sharkd") is None:
+            print(".. sharkd is not installed: skipping the tag replay section")
+        else:
+            # Tag replay reads the pcaps at rest: pause every marker of the
+            # tag. m-icmp is the only one carrying TAG.
+            print(f".. pausing m-icmp and replaying tag {TAG}")
+            compute.call("PUT", f"/projects/{pid}/links/{lid}/markers/m-icmp", {"enabled": False})
+            status, timeline = _raw(compute, f"/projects/{pid}/markers/tags/{TAG}/replay/range")
+            assert status == 200, timeline
+            assert timeline["tag"] == TAG
+            assert timeline["frame_count"] == 14, timeline["frame_count"]
+            assert timeline["start"] and timeline["end"], timeline
+            assert timeline["sources"] == [
+                {
+                    "node_id": n1_id,
+                    "link_id": lid,
+                    "marker": "m-icmp",
+                    "data_link_type": "DLT_EN10MB",
+                    "count": 14,
+                }
+            ], timeline["sources"]
+            first = timeline["frames"][0]
+            assert first["src"] == R1_IP and first["dst"] == R2_IP, first
+            assert "ICMP" in (first["proto"] or ""), first
+
+            # display filter: echo requests only (5 + 2 of the 14)
+            query = urllib.parse.urlencode({"filter": "icmp.type == 8"})
+            status, filtered = _raw(compute, f"/projects/{pid}/markers/tags/{TAG}/replay/range?{query}")
+            assert status == 200, filtered
+            assert filtered["frame_count"] == 7, filtered["frame_count"]
+            # unknown link: an empty stream, same shape as a zero-match filter
+            query = urllib.parse.urlencode({"link": "00000000-0000-0000-0000-000000000000"})
+            status, empty = _raw(compute, f"/projects/{pid}/markers/tags/{TAG}/replay/range?{query}")
+            assert status == 200 and empty["frame_count"] == 0 and empty["frames"] == [], empty
+            # window endpoint: the whole capture inside one window
+            query = urllib.parse.urlencode({"ts": timeline["start"], "window_ms": 120000, "limit": 1000})
+            status, window = _raw(compute, f"/projects/{pid}/markers/tags/{TAG}/replay/frames?{query}")
+            assert status == 200 and len(window["frames"]) == 14, window
+
+            # the gate: replay refuses while the tag is still capturing
+            compute.call("PUT", f"/projects/{pid}/links/{lid}/markers/m-icmp", {"enabled": True})
+            status, gate = _raw(compute, f"/projects/{pid}/markers/tags/{TAG}/replay/range")
+            assert status == 409, (status, gate)
+
+        # delete: the pcap goes with the marker and the signals stop
+        print(".. deleting m-icmp: pcap and signals go with it")
+        compute.call("DELETE", f"/projects/{pid}/links/{lid}/markers/m-icmp")
+        assert harness.wait_until(lambda: not os.path.exists(pcap), timeout=10), pcap
+        harness.docker_ping(daemon, c1, R2_IP, count=2)
+        harness.wait_until(lambda: len(_matches(marker_ws, "m-tx")) >= 12, timeout=10)
+        assert len(_matches(marker_ws, "m-icmp")) == 14, "a deleted marker must not signal"
+
+        compute.call("DELETE", f"/projects/{pid}/links/{lid}/markers/m-tcp")
+        compute.call("DELETE", f"/projects/{pid}/links/{lid}/markers/m-tx")
+        assert not os.listdir(markers_dir), os.listdir(markers_dir)
+    except BaseException:
+        harness.release(server, pid, failed=True)
+        _close(marker_ws, main_ws)
+        raise
+    _close(marker_ws, main_ws)
+    harness.release(server, pid, failed=False)