fix: stabilize large topology and device-down lifecycle

Author: yyyyyt123

.github/workflows/test.yml

@@ -84,7 +84,7 @@ jobs:
           python -m venv /tmp/netopsbench-wheel-smoke
           /tmp/netopsbench-wheel-smoke/bin/pip install dist/netopsbench-*.whl
           cd /tmp
-          /tmp/netopsbench-wheel-smoke/bin/python -c "from netopsbench.sdk import NetOpsBench; from netopsbench.platform.incident import DiagnosticSession, IncidentEngine"
+          /tmp/netopsbench-wheel-smoke/bin/python -c "from netopsbench.sdk import DiagnosticSession, IncidentEngine, NetOpsBench"
           /tmp/netopsbench-wheel-smoke/bin/netopsbench --help >/dev/null
 
   lint:

netopsbench/models/scale_profiles.yaml

@@ -21,9 +21,9 @@ profiles:
   - name: large
     topology: {family: clos, spines: 4, leafs: 16, clients_per_leaf: 4}
     management: {prefix: 24, subnet_base: 160}
-    pingmesh: {destination_batch_size: null, rtt_port_pool_size: 16, rtt_ports_per_cycle: 4, cycle_interval_seconds: 1}
-    traffic: {max_pps_per_client: 150}
-    runtime: {deploy_timeout_seconds: 2700, worker_deploy_parallelism: 1, health_timeout_seconds: 180}
+    pingmesh: {destination_batch_size: 16, rtt_port_pool_size: 16, rtt_ports_per_cycle: 4, cycle_interval_seconds: 3}
+    traffic: {max_pps_per_client: 100}
+    runtime: {deploy_timeout_seconds: 2700, worker_deploy_parallelism: 1, health_timeout_seconds: 180, containerlab_max_workers: 16}
   - name: xlarge
     topology: {family: clos, spines: 16, leafs: 128, clients_per_leaf: 1}
     management: {prefix: 23, subnet_base: 180}

netopsbench/platform/faults/handlers/system.py

@@ -18,7 +18,10 @@
 class SystemHandler:
     """Handles device-level (system) fault injection and recovery."""
 
-    _CONTAINERLAB_TIMEOUT = "20s"
+    _CONTAINERLAB_TIMEOUT = "120s"
+    _CONTAINERLAB_COMMAND_TIMEOUT_SECONDS = 150
+    _STOP_SETTLE_TIMEOUT_SECONDS = 30
+    _STOP_SETTLE_POLL_SECONDS = 1
     _ACTIVATION_MAX_TRIES = 36
     _BGP_MAX_TRIES = 20
 
@@ -47,11 +50,108 @@ def _containerlab_node_command(self, operation: str, device: str) -> list[str]:
             self._CONTAINERLAB_TIMEOUT,
         ]
 
-    def _start_and_wait(self, device: str, container: str) -> tuple[bool, str]:
-        start = self._cmd.run_cmd(self._containerlab_node_command("start", device), timeout=60)
+    @staticmethod
+    def _parking_namespace(container: str) -> str:
+        return f"clab-park-{container}"
+
+    def _parking_namespace_exists(self, container: str) -> bool | None:
+        # Listing named namespaces is read-only and does not require root.
+        # Using ``sudo -n`` here made a healthy parking namespace unreadable on
+        # hosts whose sudoers policy grants Containerlab but not arbitrary
+        # ``ip`` commands.
+        result = self._cmd.run_cmd(["ip", "netns", "list"], timeout=15)
+        if result.returncode != 0:
+            return None
+        expected = self._parking_namespace(container)
+        return any(line.split(maxsplit=1)[0] == expected for line in (result.stdout or "").splitlines() if line.strip())
+
+    def _settled_stop_state(self, container: str) -> tuple[bool | None, bool | None]:
+        """Wait for Docker's state and Containerlab's parking state to settle."""
+        deadline = time.monotonic() + self._STOP_SETTLE_TIMEOUT_SECONDS
+        state: bool | None = None
+        parking: bool | None = None
+        responsive_polls = 0
+        while True:
+            state = self._cmd.container_is_running(container)
+            parking = self._parking_namespace_exists(container)
+            if state is False:
+                return state, parking
+            if state is True and parking is False:
+                responsive = self._cmd.docker_exec(container, ["/bin/true"], timeout=10)
+                if responsive.returncode == 0:
+                    responsive_polls += 1
+                    if responsive_polls >= 3:
+                        return state, parking
+                else:
+                    responsive_polls = 0
+            else:
+                responsive_polls = 0
+            if time.monotonic() >= deadline:
+                return state, parking
+            time.sleep(self._STOP_SETTLE_POLL_SECONDS)
+
+    def _expected_dataplane_interface_count(self, device: str) -> int:
+        return sum(
+            1 for link in self._ctx.manifest.links if any(endpoint.device == device for endpoint in link.endpoints)
+        )
+
+    def _observed_dataplane_interface_count(self, container: str) -> int | None:
+        result = self._cmd.docker_exec(
+            container,
+            [
+                "bash",
+                "-lc",
+                "count=0; for path in /sys/class/net/eth*; do "
+                '[ -e "$path" ] || continue; [ "${path##*/}" = eth0 ] && continue; '
+                "count=$((count + 1)); done; "
+                "printf '%s\\n' \"$count\"",
+            ],
+            timeout=15,
+        )
+        if result.returncode != 0:
+            return None
+        try:
+            return int((result.stdout or "").strip())
+        except ValueError:
+            return None
+
+    def _start_and_wait(self, device: str, container: str) -> tuple[bool, str, bool]:
+        running_before = self._cmd.container_is_running(container)
+        parking_before = self._parking_namespace_exists(container)
+        if running_before is False and parking_before is False:
+            return False, "device container is stopped but its parking namespace is missing", False
+        if parking_before is None:
+            return False, "unable to inspect the device parking namespace", True
+        if running_before is None:
+            return False, "unable to inspect the device container state", True
+        if running_before is True:
+            return False, "device container is already running while the device-down fault is active", False
+
+        start = self._cmd.run_cmd(
+            self._containerlab_node_command("start", device),
+            timeout=self._CONTAINERLAB_COMMAND_TIMEOUT_SECONDS,
+        )
         running_state = self._cmd.container_is_running(container)
-        if start.returncode != 0 and running_state is not True:
-            return False, (start.stderr or start.stdout or "").strip() or "containerlab node start failed"
+        parking_after = self._parking_namespace_exists(container)
+        if running_state is not True or parking_after is not False:
+            detail = (start.stderr or start.stdout or "").strip() or "containerlab node start failed"
+            retryable = running_state is False and parking_after is True
+            if parking_after is False and running_state is not True:
+                retryable = False
+                detail = f"{detail}; parking namespace was lost before the container recovered"
+            elif parking_after is True and running_state is True:
+                retryable = False
+                detail = f"{detail}; container is running while dataplane interfaces remain parked"
+            return False, detail, retryable
+
+        expected_interfaces = self._expected_dataplane_interface_count(device)
+        observed_interfaces = self._observed_dataplane_interface_count(container)
+        if observed_interfaces != expected_interfaces:
+            return (
+                False,
+                f"restored dataplane interface count is {observed_interfaces}, expected {expected_interfaces}",
+                False,
+            )
 
         if not self._sonic.supervisord_ready(container):
             supervisor = self._cmd.run_cmd(
@@ -66,11 +166,15 @@ def _start_and_wait(self, device: str, container: str) -> tuple[bool, str]:
                 timeout=30,
             )
             if supervisor.returncode != 0:
-                return False, (supervisor.stderr or supervisor.stdout or "").strip() or "supervisord start failed"
+                return (
+                    False,
+                    (supervisor.stderr or supervisor.stdout or "").strip() or "supervisord start failed",
+                    True,
+                )
 
         manifest_device = self._ctx.manifest.device(device)
         if manifest_device is None:
-            return False, f"device {device!r} is missing from topology manifest"
+            return False, f"device {device!r} is missing from topology manifest", False
         ecmp_hash_policy = self._ctx.manifest.routing.ecmp_hash_policy_by_role[manifest_device.role]
         activated, activation_error = activate_device(
             device,
@@ -80,68 +184,73 @@ def _start_and_wait(self, device: str, container: str) -> tuple[bool, str]:
             readiness_max_tries=self._ACTIVATION_MAX_TRIES,
         )
         if not activated:
-            return False, activation_error
+            return False, activation_error, True
 
         last_error = ""
         for _attempt in range(self._BGP_MAX_TRIES):
             running = self._cmd.container_is_running(container)
             supervisor_ready = running is True and self._sonic.supervisord_ready(container)
             if supervisor_ready and self._sonic.bgp_neighbors_established(device):
-                return True, ""
+                return True, "", False
             if running is True:
                 bgp_result = self._sonic.vtysh(device, ["show ip bgp summary"])
                 last_error = (bgp_result.stderr or bgp_result.stdout or "").strip() or last_error
             elif running is None:
                 last_error = "unable to read container running state"
             time.sleep(5)
-        return False, last_error or "device did not recover after containerlab node start"
+        return False, last_error or "device did not recover after containerlab node start", True
 
     def inject_device_down(self, device: str) -> dict[str, Any]:
         container = self._ctx.container_names.get(device)
         if not container:
             raise ValueError(f"Unknown device: {device}")
 
-        result = self._cmd.run_cmd(self._containerlab_node_command("stop", device), timeout=60)
-        running_state = self._cmd.container_is_running(container)
-        success = result.returncode == 0 and running_state is False
+        result = self._cmd.run_cmd(
+            self._containerlab_node_command("stop", device),
+            timeout=self._CONTAINERLAB_COMMAND_TIMEOUT_SECONDS,
+        )
+        running_state, parking_exists = self._settled_stop_state(container)
+        success = running_state is False and parking_exists is True
+        parking_namespace = self._parking_namespace(container)
+        if success:
+            error = None
+        else:
+            detail = (result.stderr or result.stdout or "").strip()
+            state_detail = (
+                f"container_running={running_state!r}, "
+                f"parking_namespace={parking_namespace!r}, parking_exists={parking_exists!r}"
+            )
+            error = "; ".join(filter(None, [detail, state_detail]))
         fault_info = {
             "type": "device_down",
             "device": device,
             "container": container,
             "mode": "containerlab_node_stop",
             "success": success,
-            "error": (
-                None
-                if success
-                else (result.stderr or result.stdout or "").strip()
-                or "container remained running after containerlab node stop"
-            ),
+            "parking_namespace": parking_namespace,
+            "container_running": running_state,
+            "parking_exists": parking_exists,
+            "management_unavailable": running_state is False,
+            "control_plane_unavailable": running_state is False,
+            "data_plane_unavailable": parking_exists is True,
+            "error": error,
         }
         if success:
             self._tracker.track(fault_info)
             return fault_info
 
-        compensated, compensation_error = self._start_and_wait(device, container)
-        if not compensated:
-            error = "; ".join(
-                filter(
-                    None,
-                    [
-                        str(fault_info.get("error") or ""),
-                        compensation_error or "device-down compensation failed",
-                    ],
-                )
-            )
-            fault_info["error"] = error
-            self._tracker.track_residual(fault_info, error)
+        clean_failure = running_state is True and parking_exists is False
+        if not clean_failure:
+            fault_info["retryable"] = False
+            self._tracker.track_residual(fault_info, str(error or "device-down state is inconsistent"))
         return fault_info
 
     def recover_device_down(self, device: str) -> dict[str, Any]:
         container = self._ctx.container_names.get(device)
         if not container:
             raise ValueError(f"Unknown device: {device}")
 
-        ready, last_error = self._start_and_wait(device, container)
+        ready, last_error, retryable = self._start_and_wait(device, container)
 
         if ready:
             self._tracker.remove_faults(lambda fault: fault["type"] == "device_down" and fault["device"] == device)
@@ -152,5 +261,6 @@ def recover_device_down(self, device: str) -> dict[str, Any]:
             "recovered": ready,
             "container_running": self._cmd.container_is_running(container),
             "sonic_ready": ready,
+            "retryable": retryable,
             "error": None if ready else last_error,
         }

netopsbench/platform/runtime/deployment.py

@@ -6,6 +6,7 @@
 import ipaddress
 import json
 import os
+import shlex
 import signal
 import tempfile
 import time
@@ -16,8 +17,9 @@
 from netopsbench.models.profiles import ScaleProfile, ScaleRegistry, get_scale_profile
 from netopsbench.models.runtime import RuntimeIdentity
 from netopsbench.platform.runtime.apply_configs import apply_configs
+from netopsbench.platform.topology.config import SONIC_PID1_COMMAND
 from netopsbench.platform.topology.generator import generate_topology
-from netopsbench.platform.topology.topology_utils import load_topology_manifest
+from netopsbench.platform.topology.topology_utils import clab_container_name, load_topology_manifest
 from netopsbench.platform.utils.proc import docker_prefix, safe_run, sudo_prefix
 
 APPLY_CONFIG_PARALLELISM = 32
@@ -27,6 +29,10 @@
 logger = get_logger(__name__)
 
 
+def _read_process_comm(pid: int) -> str:
+    return Path(f"/proc/{pid}/comm").read_text(encoding="utf-8").strip()
+
+
 def management_subnet_stride(scale: str, registry: ScaleRegistry | None = None) -> int:
     prefix = get_scale_profile(scale, registry).management_prefix
     return 1 if prefix >= 24 else 2 ** (24 - prefix)
@@ -148,11 +154,72 @@ def deploy_worker_lab(worker: RuntimeIdentity, scale: str, registry: ScaleRegist
         details = (deploy_result.stderr or deploy_result.stdout or "no diagnostic output").strip()
         raise RuntimeError(f"Containerlab deploy failed ({deploy_result.returncode}): {details[-4000:]}")
 
+    _verify_sonic_pid1_contract(worker)
     result = apply_configs(str(topology_dir), APPLY_CONFIG_PARALLELISM, worker.lab_name)
     if result.failed:
         raise RuntimeError(f"SONiC activation failed for: {', '.join(result.failed)}")
 
 
+def _verify_sonic_pid1_contract(worker: RuntimeIdentity) -> None:
+    """Require the generated, SIGTERM-responsive PID 1 on every SONiC node."""
+    manifest = load_topology_manifest(worker.topology_dir)
+    containers = [clab_container_name(worker.lab_name, device.name) for device in manifest.routing_devices()]
+    if not containers:
+        raise RuntimeError(f"Topology {worker.lab_name!r} has no SONiC routing devices")
+
+    inspected = safe_run(
+        [
+            *docker_prefix(),
+            "docker",
+            "inspect",
+            "--format",
+            "{{.Name}}\t{{.State.Running}}\t{{.State.Pid}}\t{{json .Config.Cmd}}",
+            *containers,
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=120,
+    )
+    if inspected.returncode != 0:
+        details = (inspected.stderr or inspected.stdout or "no diagnostic output").strip()
+        raise RuntimeError(f"Unable to verify SONiC PID 1 contract: {details[-2000:]}")
+
+    failures: list[str] = []
+    seen: set[str] = set()
+    for line in inspected.stdout.splitlines():
+        parts = line.split("\t", 3)
+        if len(parts) != 4:
+            failures.append(f"malformed docker inspect output: {line!r}")
+            continue
+        raw_name, running, raw_pid, command = parts
+        name = raw_name.lstrip("/")
+        seen.add(name)
+        if running.lower() != "true":
+            failures.append(f"{name}: container is not running")
+            continue
+        try:
+            parsed_command = json.loads(command)
+        except json.JSONDecodeError:
+            parsed_command = None
+        if parsed_command != shlex.split(SONIC_PID1_COMMAND):
+            failures.append(f"{name}: unexpected command {command}")
+            continue
+        try:
+            pid = int(raw_pid)
+            pid1 = _read_process_comm(pid)
+        except (OSError, ValueError) as exc:
+            failures.append(f"{name}: unable to inspect PID 1: {exc}")
+            continue
+        if pid1 != "bash":
+            failures.append(f"{name}: PID 1 is {pid1!r}, expected the signal-handling 'bash' wrapper")
+
+    missing = sorted(set(containers) - seen)
+    failures.extend(f"{name}: missing docker inspect result" for name in missing)
+    if failures:
+        raise RuntimeError("SONiC PID 1 contract failed: " + "; ".join(failures[:12]))
+
+
 def assert_worker_slot_available(worker: RuntimeIdentity) -> None:
     """Reject globally conflicting Containerlab resources before deployment."""
     docker = docker_prefix()