diff --git a/config/config.yaml b/config/config.yaml index ab7b87ee..49bc6ca7 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -54,6 +54,7 @@ kraken: - scenarios/kube/pod-network-chaos.yml - scenarios/kube/node_interface_down.yaml - scenarios/openshift/virt_network_chaos.yaml + - scenarios/openshift/virt_network.yaml - kubevirt_vm_outage: - scenarios/kubevirt/kubevirt-vm-outage.yaml - http_load_scenarios: diff --git a/krkn/scenario_plugins/network_chaos_ng/models.py b/krkn/scenario_plugins/network_chaos_ng/models.py index 1c45aad5..7f1c9672 100644 --- a/krkn/scenario_plugins/network_chaos_ng/models.py +++ b/krkn/scenario_plugins/network_chaos_ng/models.py @@ -66,8 +66,14 @@ class BaseNetworkChaosConfig: @dataclass class NetworkFilterConfig(BaseNetworkChaosConfig): - ports: list[int] - protocols: list[str] + ports: list[int] = None + protocols: list[str] = None + + def __post_init__(self): + if self.ports is None: + self.ports = [] + if self.protocols is None: + self.protocols = ["tcp", "udp"] def validate(self) -> list[str]: errors = super().validate() diff --git a/krkn/scenario_plugins/network_chaos_ng/modules/abstract_network_chaos_module.py b/krkn/scenario_plugins/network_chaos_ng/modules/abstract_network_chaos_module.py index 8a3104e3..e437e756 100644 --- a/krkn/scenario_plugins/network_chaos_ng/modules/abstract_network_chaos_module.py +++ b/krkn/scenario_plugins/network_chaos_ng/modules/abstract_network_chaos_module.py @@ -91,6 +91,35 @@ class AbstractNetworkChaosModule(abc.ABC): ) return [config.target] + def get_vmi_targets(self, config: BaseNetworkChaosConfig) -> list[str]: + """ + Returns the list of VMI targets in "namespace/vmi-name" format. + Supports regex matching on both name (via `target`) and namespace, + and optional post-filtering by `label_selector` in "key=value" format. + """ + if not config.namespace: + raise Exception("namespace not specified for VMI scenario, aborting") + name_regex = config.target if config.target else ".*" + vmis = self.kubecli.get_lib_kubernetes().get_vmis(name_regex, config.namespace) + if not vmis: + return [] + if config.label_selector: + try: + label_key, label_value = config.label_selector.split("=", 1) + except ValueError: + raise Exception( + f"invalid label_selector format: '{config.label_selector}', expected 'key=value'" + ) + vmis = [ + vmi + for vmi in vmis + if vmi.get("metadata", {}).get("labels", {}).get(label_key) == label_value + ] + return [ + f"{vmi['metadata']['namespace']}/{vmi['metadata']['name']}" + for vmi in vmis + ] + def __init__( self, base_network_config: BaseNetworkChaosConfig, diff --git a/krkn/scenario_plugins/network_chaos_ng/modules/node_interface_down.py b/krkn/scenario_plugins/network_chaos_ng/modules/node_interface_down.py index e714f868..c9aa3918 100644 --- a/krkn/scenario_plugins/network_chaos_ng/modules/node_interface_down.py +++ b/krkn/scenario_plugins/network_chaos_ng/modules/node_interface_down.py @@ -11,6 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. + import queue import time from typing import Tuple @@ -114,13 +115,6 @@ class NodeInterfaceDownModule(AbstractNetworkChaosModule): target, ) - log_info( - f"waiting {self.config.test_duration} seconds for interface(s) to recover", - parallel, - target, - ) - time.sleep(self.config.test_duration) - log_info( f"waiting for node {target} to become Ready after interface recovery", parallel, diff --git a/krkn/scenario_plugins/network_chaos_ng/modules/utils.py b/krkn/scenario_plugins/network_chaos_ng/modules/utils.py index f4f45049..a2511aca 100644 --- a/krkn/scenario_plugins/network_chaos_ng/modules/utils.py +++ b/krkn/scenario_plugins/network_chaos_ng/modules/utils.py @@ -18,7 +18,6 @@ from typing import Optional, Tuple import yaml from jinja2 import FileSystemLoader, Environment from krkn_lib.k8s import KrknKubernetes -from krkn_lib.models.k8s import Pod from krkn.scenario_plugins.network_chaos_ng.models import ( BaseNetworkChaosConfig, @@ -122,6 +121,9 @@ def find_virt_launcher_netns_pid( namespace — some helper processes run in the host netns. Entering one of those would target the node's physical NIC instead of the bridge slave inside the virt-launcher netns. + + tap0 is a KubeVirt-specific tap device that only exists inside the + virt-launcher's netns, so its presence is a reliable probe. """ for pid in pids: try: @@ -161,6 +163,56 @@ def get_vmi_tap_interface( return None +def get_vmi_tap_interface( + chaos_pod_name: str, namespace: str, pid: str, kubecli: KrknKubernetes +) -> str: + """Find the VMI's primary tap interface inside the virt-launcher network namespace. + + The tap device is the VM-facing member of the KubeVirt bridge: + ovn-udn1-nic -> k6t-ovn-udn1 (bridge) -> tap0 -> QEMU (VM guest) + + We locate it by finding the tap member of the k6t-* bridge rather than + grepping for any tap-prefixed device, so the detection works regardless + of how many interfaces the VM has. + + Blocking the tap interface isolates only this VMI. Blocking the bridge + slave (ovn-udn1-nic) would also sever OVN's BFD heartbeats and trigger + a node-wide network reconvergence. + """ + # Find the k6t-* bridge name first, then find its tap member. + bridge_cmd = ( + f"nsenter --target {pid} --net -- " + f"ip link show | grep ': k6t-' | head -1 | cut -d: -f2 | tr -d ' '" + ) + bridge = kubecli.exec_cmd_in_pod([bridge_cmd], chaos_pod_name, namespace).strip() + if not bridge: + return "" + + tap_cmd = ( + f"nsenter --target {pid} --net -- " + f"ip link show master {bridge} | grep ': tap' | head -1 | cut -d: -f2 | tr -d ' '" + ) + output = kubecli.exec_cmd_in_pod([tap_cmd], chaos_pod_name, namespace) + return output.strip() + + +def get_vmi_masquerade_interface( + chaos_pod_name: str, namespace: str, netns_pid: str, kubecli: KrknKubernetes +) -> str: + """Return the default-route interface inside the virt-launcher netns (masquerade mode).""" + result = kubecli.exec_cmd_in_pod( + [f"nsenter --target {netns_pid} --net -- ip route show default"], + chaos_pod_name, + namespace, + ) + parts = result.split() if result else [] + if "dev" in parts: + idx = parts.index("dev") + if idx + 1 < len(parts): + return parts[idx + 1] + return "" + + def setup_network_chaos_ng_scenario( config: BaseNetworkChaosConfig, node_name: str, diff --git a/krkn/scenario_plugins/network_chaos_ng/modules/utils_network_filter.py b/krkn/scenario_plugins/network_chaos_ng/modules/utils_network_filter.py index f1f67d2f..152c8844 100644 --- a/krkn/scenario_plugins/network_chaos_ng/modules/utils_network_filter.py +++ b/krkn/scenario_plugins/network_chaos_ng/modules/utils_network_filter.py @@ -25,17 +25,29 @@ def generate_rules( input_rules = [] output_rules = [] for interface in interfaces: - for port in config.ports: + if config.ports: + for port in config.ports: + if config.egress: + for protocol in set(config.protocols): + output_rules.append( + f"iptables -I OUTPUT 1 -p {protocol} --dport {port} -m state --state NEW,RELATED,ESTABLISHED -j DROP" + ) + if config.ingress: + for protocol in set(config.protocols): + input_rules.append( + f"iptables -I INPUT 1 -i {interface} -p {protocol} --dport {port} -m state --state NEW,RELATED,ESTABLISHED -j DROP" + ) + else: + # empty ports means block all traffic on all ports if config.egress: for protocol in set(config.protocols): output_rules.append( - f"iptables -I OUTPUT 1 -p {protocol} --dport {port} -m state --state NEW,RELATED,ESTABLISHED -j DROP" + f"iptables -I OUTPUT 1 -p {protocol} -m state --state NEW,RELATED,ESTABLISHED -j DROP" ) - if config.ingress: for protocol in set(config.protocols): input_rules.append( - f"iptables -I INPUT 1 -i {interface} -p {protocol} --dport {port} -m state --state NEW,RELATED,ESTABLISHED -j DROP" + f"iptables -I INPUT 1 -i {interface} -p {protocol} -m state --state NEW,RELATED,ESTABLISHED -j DROP" ) return input_rules, output_rules @@ -115,3 +127,50 @@ def generate_namespaced_rules( namespaced_output_rules.extend(ns_output_rules) return namespaced_input_rules, namespaced_output_rules + + +def apply_tc_vmi_chaos( + kubecli: KrknKubernetes, + chaos_pod_name: str, + namespace: str, + pid: str, + iface: str, + config: NetworkFilterConfig, + parallel: bool, + vmi_name: str, +) -> Tuple[list[str], list[str]]: + """Apply iptables rules inside the virt-launcher netns via nsenter. + + Targets the tap interface (tap0) rather than the bridge slave (ovn-udn1-nic) + so that OVN's BFD heartbeats on the bridge are unaffected. Rules are applied + inside the virt-launcher's network namespace using nsenter, matching the same + iptables approach used by pod_network_filter and node_network_filter. + + Returns (input_rules, output_rules) needed for cleanup. + """ + log_info( + f"applying iptables rules on {iface} " + f"(ports:{config.ports}, protocols:{config.protocols})", + parallel, + vmi_name, + ) + input_rules, output_rules = generate_namespaced_rules([iface], config, [pid]) + apply_network_rules(kubecli, input_rules, output_rules, chaos_pod_name, namespace, parallel, vmi_name) + return input_rules, output_rules + + +def clean_tc_vmi_chaos( + kubecli: KrknKubernetes, + chaos_pod_name: str, + namespace: str, + pid: str, + iface: str, + input_rules: list[str], + output_rules: list[str], +): + """Remove iptables rules applied by apply_tc_vmi_chaos.""" + clean_network_rules_namespaced( + kubecli, input_rules, output_rules, chaos_pod_name, namespace, [pid] + ) + + diff --git a/krkn/scenario_plugins/network_chaos_ng/modules/vmi_network_filter.py b/krkn/scenario_plugins/network_chaos_ng/modules/vmi_network_filter.py new file mode 100644 index 00000000..82a4ea49 --- /dev/null +++ b/krkn/scenario_plugins/network_chaos_ng/modules/vmi_network_filter.py @@ -0,0 +1,338 @@ +# Copyright 2025 The Krkn Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +import dataclasses +import queue +import re +import time +from typing import Tuple + +from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift +from krkn_lib.utils import get_random_string + +from krkn.scenario_plugins.network_chaos_ng.models import ( + NetworkChaosScenarioType, + BaseNetworkChaosConfig, + NetworkFilterConfig, +) +from krkn.scenario_plugins.network_chaos_ng.modules.abstract_network_chaos_module import ( + AbstractNetworkChaosModule, +) +from krkn.scenario_plugins.network_chaos_ng.modules.utils import ( + log_info, + log_error, + deploy_network_chaos_ng_pod, + find_virt_launcher_netns_pid, + get_vmi_tap_interface, + get_vmi_masquerade_interface, +) +from krkn.scenario_plugins.network_chaos_ng.modules.utils_network_filter import ( + apply_tc_vmi_chaos, + clean_tc_vmi_chaos, +) + +_UNSUPPORTED_MODES = frozenset({"sriov", "macvtap"}) +_KNOWN_BINDING_KEYS = frozenset({"bridge", "masquerade", "sriov", "macvtap", "slirp", "passt"}) + + +def _get_vmi_network_mode(vmi: dict) -> str: + """Return the network binding mode of the first VMI interface, defaulting to 'bridge'.""" + ifaces = ( + vmi.get("spec", {}) + .get("domain", {}) + .get("devices", {}) + .get("interfaces", []) + ) + if not ifaces: + return "bridge" + for key in _KNOWN_BINDING_KEYS: + if key in ifaces[0]: + return key + return "bridge" + + +class VmiNetworkFilterModule(AbstractNetworkChaosModule): + + def __init__(self, config: NetworkFilterConfig, kubecli: KrknTelemetryOpenshift): + super().__init__(config, kubecli) + self.config = config + + def _rollback( + self, + namespace: str, + network_chaos_pod_name: str, + netns_pid: str = None, + iface: str = None, + input_rules: list = None, + output_rules: list = None, + ): + if netns_pid and iface and input_rules is not None and output_rules is not None: + clean_tc_vmi_chaos( + self.kubecli.get_lib_kubernetes(), + network_chaos_pod_name, + namespace, + netns_pid, + iface, + input_rules, + output_rules, + ) + self.kubecli.get_lib_kubernetes().delete_pod( + network_chaos_pod_name, namespace + ) + + def run(self, target: str, error_queue: queue.Queue = None): + # target is "namespace/vmi-name" as produced by get_vmi_targets() + parallel = False + if error_queue: + parallel = True + + network_chaos_pod_name = None + netns_pid = None + iface = None + input_rules = None + output_rules = None + namespace = "" + vmi_name = "" + + try: + namespace, vmi_name = target.split("/", 1) + # Create a scoped config with the resolved namespace so that all + # Kubernetes calls use the actual namespace, not the regex pattern. + scoped_config = dataclasses.replace(self.config, namespace=namespace) + network_chaos_pod_name = f"vmi-network-chaos-{get_random_string(5)}" + container_name = f"fedora-container-{get_random_string(5)}" + + log_info( + f"creating workload to filter VMI {vmi_name} network " + f"ports:{','.join([str(p) for p in self.config.ports])}, " + f"ingress:{str(self.config.ingress)}, " + f"egress:{str(self.config.egress)}", + parallel, + network_chaos_pod_name, + ) + + # Resolve which node the VMI is running on + vmi = self.kubecli.get_lib_kubernetes().get_vmi(vmi_name, namespace) + if not vmi: + raise Exception( + f"VMI {vmi_name} not found in namespace {namespace}" + ) + + node_name = vmi.get("status", {}).get("nodeName") + if not node_name: + raise Exception( + f"unable to determine node for VMI {vmi_name} in namespace {namespace}; " + "VMI may not be in Running phase" + ) + + log_info( + f"VMI {vmi_name} is running on node {node_name}", + parallel, + network_chaos_pod_name, + ) + + # Fail fast for unsupported binding modes before deploying the chaos pod. + # sriov/macvtap interfaces don't live in a netns we can nsenter into. + if not scoped_config.interfaces: + network_mode = _get_vmi_network_mode(vmi) + if network_mode in _UNSUPPORTED_MODES: + raise Exception( + f"VMI network binding mode '{network_mode}' is not supported; " + "cannot inject chaos" + ) + + # The virt-launcher pod carries the VMI's network namespace. + # It is labelled vm.kubevirt.io/name=. + virt_launcher_pods = self.kubecli.get_lib_kubernetes().list_pods( + namespace, label_selector=f"vm.kubevirt.io/name={vmi_name}" + ) + if not virt_launcher_pods: + raise Exception( + f"no virt-launcher pod found for VMI {vmi_name} in namespace {namespace}" + ) + virt_launcher_pod_name = virt_launcher_pods[0] + + log_info( + f"resolved virt-launcher pod {virt_launcher_pod_name} for VMI {vmi_name}", + parallel, + network_chaos_pod_name, + ) + + # Deploy the privileged chaos pod onto the VMI's node. + # hostPID=True (via template) allows nsenter into the virt-launcher's + # network namespace using any of the compute container's host PIDs. + deploy_network_chaos_ng_pod( + scoped_config, + node_name, + network_chaos_pod_name, + self.kubecli.get_lib_kubernetes(), + container_name, + host_network=False, + ) + + # Prefer the 'compute' container (the QEMU process in KubeVirt). + # 'virt-launcher' is a sidecar monitor that may not be running; + # using its cgroup ID would cause get_pod_pids to return nothing. + pod_info = self.kubecli.get_lib_kubernetes().get_pod_info( + virt_launcher_pod_name, namespace + ) + if not pod_info: + raise Exception( + f"impossible to retrieve info for virt-launcher pod " + f"{virt_launcher_pod_name} in namespace {namespace}" + ) + + target_container_id = None + for container in pod_info.containers: + if container.name == "compute" and container.ready and container.containerId: + target_container_id = re.sub(r".*://", "", container.containerId) + break + if not target_container_id: + raise Exception( + f"compute container in virt-launcher pod {virt_launcher_pod_name} " + f"in namespace {namespace} is not ready" + ) + + log_info( + f"targeting compute container {target_container_id}", + parallel, + network_chaos_pod_name, + ) + + pids = self.kubecli.get_lib_kubernetes().get_pod_pids( + base_pod_name=network_chaos_pod_name, + base_pod_namespace=namespace, + base_pod_container_name=container_name, + pod_name=virt_launcher_pod_name, + pod_namespace=namespace, + pod_container_id=target_container_id, + ) + if not pids: + raise Exception( + f"impossible to resolve PIDs for virt-launcher pod {virt_launcher_pod_name}" + ) + + log_info( + f"resolved PIDs {pids} on node {node_name} for VMI {vmi_name}", + parallel, + network_chaos_pod_name, + ) + + # Not all PIDs from get_pod_pids are in the virt-launcher's netns — + # some helper/privileged processes run in the host netns. Entering + # one of those would target the node's physical NIC (e.g. ens4) + # instead of the bridge slave inside the virt-launcher's netns. + netns_pid = find_virt_launcher_netns_pid( + network_chaos_pod_name, + namespace, + pids, + self.kubecli.get_lib_kubernetes(), + ) + if not netns_pid: + raise Exception( + f"could not find a PID in the virt-launcher netns for VMI {vmi_name}; " + "none of the compute container PIDs contain tap0" + ) + + log_info( + f"using PID {netns_pid} for netns entry (virt-launcher netns confirmed via tap0)", + parallel, + network_chaos_pod_name, + ) + + if len(scoped_config.interfaces) == 0: + # network_mode was computed (and validated) in the early-exit block above. + if network_mode == "masquerade": + iface = get_vmi_masquerade_interface( + network_chaos_pod_name, + namespace, + netns_pid, + self.kubecli.get_lib_kubernetes(), + ) + else: + iface = get_vmi_tap_interface( + network_chaos_pod_name, + namespace, + netns_pid, + self.kubecli.get_lib_kubernetes(), + ) + if not iface: + log_error( + f"could not detect interface for '{network_mode}' mode in " + "virt-launcher netns; impossible to execute the VMI network filter scenario", + parallel, + network_chaos_pod_name, + ) + self._rollback(namespace, network_chaos_pod_name) + return + else: + iface = scoped_config.interfaces[0] + + log_info( + f"targeting interface: {iface}", + parallel, + network_chaos_pod_name, + ) + + input_rules, output_rules = apply_tc_vmi_chaos( + self.kubecli.get_lib_kubernetes(), + network_chaos_pod_name, + namespace, + netns_pid, + iface, + scoped_config, + parallel, + vmi_name, + ) + + log_info( + f"waiting {self.config.test_duration} seconds before removing iptables rules", + parallel, + network_chaos_pod_name, + ) + + time.sleep(self.config.test_duration) + + log_info("removing iptables rules", parallel, network_chaos_pod_name) + + self._rollback(namespace, network_chaos_pod_name, netns_pid, iface, input_rules, output_rules) + + except Exception as e: + if network_chaos_pod_name: + self._rollback(namespace, network_chaos_pod_name, netns_pid, iface, input_rules, output_rules) + if error_queue is None: + raise e + else: + error_queue.put(str(e)) + + def get_config(self) -> Tuple[NetworkChaosScenarioType, BaseNetworkChaosConfig]: + return NetworkChaosScenarioType.VMI, self.config + + def get_targets(self) -> list[str]: + if not self.config.namespace: + raise Exception("namespace not specified for VMI scenario, aborting") + name_regex = self.config.target if self.config.target else ".*" + label_selector = self.config.label_selector or None + + vmis = self.kubecli.get_lib_kubernetes().get_vmis( + name_regex, self.config.namespace, label_selector=label_selector + ) + return [ + f"{vmi['metadata']['namespace']}/{vmi['metadata']['name']}" + for vmi in vmis + if re.match(name_regex, vmi.get("metadata", {}).get("name", "")) + and re.match( + self.config.namespace, + vmi.get("metadata", {}).get("namespace", ""), + ) + ] diff --git a/krkn/scenario_plugins/network_chaos_ng/network_chaos_factory.py b/krkn/scenario_plugins/network_chaos_ng/network_chaos_factory.py index 50a7a38a..1fc3ab18 100644 --- a/krkn/scenario_plugins/network_chaos_ng/network_chaos_factory.py +++ b/krkn/scenario_plugins/network_chaos_ng/network_chaos_factory.py @@ -39,6 +39,9 @@ from krkn.scenario_plugins.network_chaos_ng.modules.pod_network_filter import ( from krkn.scenario_plugins.network_chaos_ng.modules.vmi_network_chaos import ( VmiNetworkChaosModule, ) +from krkn.scenario_plugins.network_chaos_ng.modules.vmi_network_filter import ( + VmiNetworkFilterModule, +) supported_modules = [ "node_network_filter", @@ -47,6 +50,7 @@ supported_modules = [ "node_network_chaos", "node_interface_down", "vmi_network_chaos", + "vmi_network_filter", ] @@ -97,5 +101,11 @@ class NetworkChaosFactory: if len(errors) > 0: raise Exception(f"config validation errors: [{';'.join(errors)}]") return VmiNetworkChaosModule(scenario_config, kubecli) + if config["id"] == "vmi_network_filter": + scenario_config = NetworkFilterConfig(**config) + errors = scenario_config.validate() + if len(errors) > 0: + raise Exception(f"config validation errors: [{';'.join(errors)}]") + return VmiNetworkFilterModule(scenario_config, kubecli) else: raise Exception(f"invalid network chaos id {config['id']}") diff --git a/scenarios/openshift/virt_network.yaml b/scenarios/openshift/virt_network.yaml new file mode 100644 index 00000000..c413c0dd --- /dev/null +++ b/scenarios/openshift/virt_network.yaml @@ -0,0 +1,15 @@ +- id: vmi_network_filter + image: "quay.io/krkn-chaos/krkn-network-chaos:latest" + wait_duration: 300 + test_duration: 120 + label_selector: "" + service_account: "" + taints: [] + namespace: + instance_count: 1 + execution: serial + target: "" + interfaces: [] + ingress: true + egress: true + ports: [] diff --git a/tests/test_node_interface_down.py b/tests/test_node_interface_down.py index 654a42a6..678dbd80 100644 --- a/tests/test_node_interface_down.py +++ b/tests/test_node_interface_down.py @@ -167,13 +167,15 @@ class TestNodeInterfaceDownModule(unittest.TestCase): @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.deploy_network_chaos_ng_pod") @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.log_info") def test_run_sleeps_test_duration(self, mock_log, mock_deploy, mock_sleep): + # test_duration is embedded in the shell command (sleep {n} && ip link set up), + # so no Python time.sleep(test_duration) should be called. self.config.test_duration = 45 self.config.recovery_time = 0 self.module.run("worker-1") sleep_values = [c[0][0] for c in mock_sleep.call_args_list] - self.assertIn(45, sleep_values) + self.assertNotIn(45, sleep_values) @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.time.sleep") @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.deploy_network_chaos_ng_pod") @@ -185,7 +187,7 @@ class TestNodeInterfaceDownModule(unittest.TestCase): self.module.run("worker-1") sleep_values = [c[0][0] for c in mock_sleep.call_args_list] - self.assertIn(30, sleep_values) + self.assertNotIn(30, sleep_values) self.assertIn(15, sleep_values) @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.time.sleep") @@ -198,7 +200,7 @@ class TestNodeInterfaceDownModule(unittest.TestCase): self.module.run("worker-1") sleep_values = [c[0][0] for c in mock_sleep.call_args_list] - self.assertIn(30, sleep_values) + self.assertNotIn(30, sleep_values) self.assertNotIn(0, sleep_values) @patch("krkn.scenario_plugins.network_chaos_ng.modules.node_interface_down.time.sleep") diff --git a/tests/test_vmi_network_filter.py b/tests/test_vmi_network_filter.py new file mode 100644 index 00000000..7f5525a7 --- /dev/null +++ b/tests/test_vmi_network_filter.py @@ -0,0 +1,792 @@ +#!/usr/bin/env python3 + +""" +Test suite for VmiNetworkFilterModule + +Usage: + python -m unittest tests/test_vmi_network_filter.py -v + python -m coverage run -a -m unittest tests/test_vmi_network_filter.py -v +""" + +import queue +import unittest +from unittest.mock import MagicMock, call, patch + +from krkn.scenario_plugins.network_chaos_ng.models import ( + NetworkChaosScenarioType, + NetworkFilterConfig, +) +from krkn.scenario_plugins.network_chaos_ng.modules.vmi_network_filter import ( + VmiNetworkFilterModule, +) + +MODULE = "krkn.scenario_plugins.network_chaos_ng.modules.vmi_network_filter" + + +def _make_config(**overrides): + defaults = dict( + id="vmi_network_filter", + image="quay.io/krkn-chaos/krkn-network-chaos:latest", + wait_duration=300, + test_duration=60, + label_selector="", + service_account="", + taints=[], + namespace="virt-density-udn-3", + instance_count=1, + execution="serial", + target=".*", + interfaces=[], + ingress=True, + egress=True, + ports=[], + protocols=["tcp", "udp"], + ) + defaults.update(overrides) + return NetworkFilterConfig(**defaults) + + +def _make_container(name, ready=True, container_id="containerd://abc123"): + c = MagicMock() + c.name = name + c.ready = ready + c.containerId = container_id + return c + + +class TestVmiNetworkFilterModuleInit(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.config = _make_config() + self.module = VmiNetworkFilterModule(self.config, self.mock_kubecli) + + def test_initialization(self): + self.assertEqual(self.module.config, self.config) + self.assertEqual(self.module.kubecli, self.mock_kubecli) + self.assertEqual(self.module.base_network_config, self.config) + + def test_get_config(self): + scenario_type, config = self.module.get_config() + self.assertEqual(scenario_type, NetworkChaosScenarioType.VMI) + self.assertEqual(config, self.config) + + +class TestVmiNetworkFilterModuleGetTargets(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.mock_kubernetes = MagicMock() + self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes + self.config = _make_config( + namespace="virt-density-udn-3", + target="virt-server-.*", + ) + self.module = VmiNetworkFilterModule(self.config, self.mock_kubecli) + + def test_get_targets_success(self): + vmis = [ + {"metadata": {"name": "virt-server-1", "namespace": "virt-density-udn-3"}}, + {"metadata": {"name": "virt-server-2", "namespace": "virt-density-udn-3"}}, + ] + self.mock_kubernetes.get_vmis.return_value = vmis + + result = self.module.get_targets() + + self.assertEqual( + result, + [ + "virt-density-udn-3/virt-server-1", + "virt-density-udn-3/virt-server-2", + ], + ) + self.mock_kubernetes.get_vmis.assert_called_once_with( + "virt-server-.*", "virt-density-udn-3", label_selector=None + ) + + def test_get_targets_no_namespace_raises(self): + self.config.namespace = None + with self.assertRaises(Exception) as ctx: + self.module.get_targets() + self.assertIn("namespace not specified", str(ctx.exception)) + + def test_get_targets_no_vmis_returns_empty(self): + self.mock_kubernetes.get_vmis.return_value = [] + result = self.module.get_targets() + self.assertEqual(result, []) + + def test_get_targets_regex_filters_name(self): + self.config.target = "virt-server-1" + vmis = [ + {"metadata": {"name": "virt-server-1", "namespace": "virt-density-udn-3"}}, + {"metadata": {"name": "virt-server-10", "namespace": "virt-density-udn-3"}}, + ] + self.mock_kubernetes.get_vmis.return_value = vmis + + result = self.module.get_targets() + + # re.match("virt-server-1", "virt-server-10") matches (prefix), both included + self.assertIn("virt-density-udn-3/virt-server-1", result) + + def test_get_targets_regex_filters_namespace(self): + self.config.namespace = "virt-density-udn-3" + vmis = [ + {"metadata": {"name": "virt-server-1", "namespace": "virt-density-udn-3"}}, + {"metadata": {"name": "virt-server-2", "namespace": "other-namespace"}}, + ] + self.mock_kubernetes.get_vmis.return_value = vmis + + result = self.module.get_targets() + + self.assertIn("virt-density-udn-3/virt-server-1", result) + self.assertNotIn("other-namespace/virt-server-2", result) + + def test_get_targets_passes_label_selector(self): + self.config.label_selector = "app=myapp" + self.mock_kubernetes.get_vmis.return_value = [] + + self.module.get_targets() + + self.mock_kubernetes.get_vmis.assert_called_once_with( + "virt-server-.*", "virt-density-udn-3", label_selector="app=myapp" + ) + + def test_get_targets_empty_label_selector_passes_none(self): + self.config.label_selector = "" + self.mock_kubernetes.get_vmis.return_value = [] + + self.module.get_targets() + + self.mock_kubernetes.get_vmis.assert_called_once_with( + "virt-server-.*", "virt-density-udn-3", label_selector=None + ) + + +class TestVmiNetworkFilterModuleRun(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.mock_kubernetes = MagicMock() + self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes + self.config = _make_config( + namespace="virt-density-udn-.*", + target="virt-server-.*", + test_duration=60, + interfaces=[], + ) + self.module = VmiNetworkFilterModule(self.config, self.mock_kubecli) + + # common happy-path mocks + self.mock_kubernetes.get_vmi.return_value = { + "status": {"nodeName": "worker-1"} + } + self.mock_kubernetes.list_pods.return_value = [ + "virt-launcher-virt-server-3-abc12" + ] + + compute = _make_container("compute", ready=True, container_id="containerd://deadbeef") + virt_launcher = _make_container("virt-launcher", ready=False, container_id="") + mock_pod_info = MagicMock() + mock_pod_info.containers = [virt_launcher, compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + + self.mock_kubernetes.get_pod_pids.return_value = ["100", "101", "102"] + + # ------------------------------------------------------------------ helpers + + def _patch_run(self): + """Return a context-manager stack that patches all external calls in run.""" + return [ + patch(f"{MODULE}.deploy_network_chaos_ng_pod"), + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), + patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0"), + patch(f"{MODULE}.apply_tc_vmi_chaos"), + patch(f"{MODULE}.clean_tc_vmi_chaos"), + patch(f"{MODULE}.time.sleep"), + patch(f"{MODULE}.log_info"), + patch(f"{MODULE}.log_error"), + ] + + # ------------------------------------------------------------------ success + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_success( + self, + mock_log, + mock_sleep, + mock_deploy, + mock_find_pid, + mock_tap, + mock_apply, + mock_clean, + ): + self.module.run("virt-density-udn-3/virt-server-3") + + mock_deploy.assert_called_once() + mock_find_pid.assert_called_once() + mock_tap.assert_called_once() + mock_apply.assert_called_once() + mock_sleep.assert_called_once_with(60) + mock_clean.assert_called_once() + self.mock_kubernetes.delete_pod.assert_called_once() + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_uses_resolved_namespace_not_regex( + self, + mock_log, + mock_sleep, + mock_deploy, + mock_find_pid, + mock_tap, + mock_apply, + mock_clean, + ): + """Kubernetes calls must use the real namespace, not the regex pattern.""" + self.module.run("virt-density-udn-3/virt-server-3") + + # get_vmi called with the resolved namespace + self.mock_kubernetes.get_vmi.assert_called_once_with( + "virt-server-3", "virt-density-udn-3" + ) + # deploy called with scoped_config (namespace = resolved), not regex + deploy_config = mock_deploy.call_args[0][0] + self.assertEqual(deploy_config.namespace, "virt-density-udn-3") + self.assertNotEqual(deploy_config.namespace, "virt-density-udn-.*") + + # ------------------------------------------------------------------ vmi not found + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_vmi_not_found_raises(self, mock_log, mock_deploy): + self.mock_kubernetes.get_vmi.return_value = None + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("not found", str(ctx.exception)) + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_vmi_no_node_raises(self, mock_log, mock_deploy): + self.mock_kubernetes.get_vmi.return_value = {"status": {}} + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("unable to determine node", str(ctx.exception)) + + # ------------------------------------------------------------------ virt-launcher pod + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_no_virt_launcher_pod_raises(self, mock_log, mock_deploy): + self.mock_kubernetes.list_pods.return_value = [] + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("no virt-launcher pod found", str(ctx.exception)) + + # ------------------------------------------------------------------ compute container + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_no_pod_info_raises(self, mock_log, mock_deploy): + self.mock_kubernetes.get_pod_info.return_value = None + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("impossible to retrieve info", str(ctx.exception)) + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_compute_not_ready_raises(self, mock_log, mock_deploy): + compute = _make_container("compute", ready=False, container_id="containerd://abc") + mock_pod_info = MagicMock() + mock_pod_info.containers = [compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("compute container", str(ctx.exception)) + self.assertIn("not ready", str(ctx.exception)) + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_no_compute_container_raises(self, mock_log, mock_deploy): + other = _make_container("virt-launcher", ready=True, container_id="containerd://abc") + mock_pod_info = MagicMock() + mock_pod_info.containers = [other] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("compute container", str(ctx.exception)) + + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_strips_container_id_prefix(self, mock_log, mock_deploy, mock_find): + """containerd:// prefix must be stripped before passing to get_pod_pids.""" + compute = _make_container( + "compute", ready=True, container_id="containerd://deadbeef123" + ) + mock_pod_info = MagicMock() + mock_pod_info.containers = [compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + self.mock_kubernetes.get_pod_pids.return_value = ["100"] + + with patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])), \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"): + self.module.run("virt-density-udn-3/virt-server-3") + + call_kwargs = self.mock_kubernetes.get_pod_pids.call_args[1] + self.assertEqual(call_kwargs["pod_container_id"], "deadbeef123") + + # ------------------------------------------------------------------ pids / netns + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_no_pids_raises(self, mock_log, mock_deploy): + self.mock_kubernetes.get_pod_pids.return_value = None + + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("impossible to resolve PIDs", str(ctx.exception)) + + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value=None) + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_no_netns_pid_raises(self, mock_log, mock_deploy, mock_find): + with self.assertRaises(Exception) as ctx: + self.module.run("virt-density-udn-3/virt-server-3") + + self.assertIn("could not find a PID", str(ctx.exception)) + + # ------------------------------------------------------------------ error queue + + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_error_queue_captures_exception(self, mock_log, mock_deploy): + self.mock_kubernetes.get_vmi.return_value = None + error_queue = queue.Queue() + + self.module.run("virt-density-udn-3/virt-server-3", error_queue) + + self.assertFalse(error_queue.empty()) + self.assertIn("not found", error_queue.get()) + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_no_error_queue_raises_directly( + self, mock_log, mock_sleep, mock_deploy, mock_find, mock_tap, mock_apply, mock_clean + ): + mock_apply.side_effect = RuntimeError("tc failed") + + with self.assertRaises(RuntimeError): + self.module.run("virt-density-udn-3/virt-server-3") + + # ------------------------------------------------------------------ apply / clean called correctly + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_apply_and_clean_called_with_tap_and_pid( + self, mock_log, mock_sleep, mock_deploy, mock_find, mock_tap, mock_apply, mock_clean + ): + self.module.run("virt-density-udn-3/virt-server-3") + + apply_args = mock_apply.call_args[0] + self.assertEqual(apply_args[3], "101") # pid + self.assertEqual(apply_args[4], "tap0") # iface + + clean_args = mock_clean.call_args[0] + self.assertEqual(clean_args[3], "101") # pid + self.assertEqual(clean_args[4], "tap0") # iface + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_chaos_pod_deleted_after_clean( + self, mock_log, mock_sleep, mock_deploy, mock_find, mock_tap, mock_apply, mock_clean + ): + """Chaos pod must be cleaned up even after a successful run.""" + self.module.run("virt-density-udn-3/virt-server-3") + self.mock_kubernetes.delete_pod.assert_called_once() + # namespace passed to delete_pod must be the resolved one + delete_ns = self.mock_kubernetes.delete_pod.call_args[0][1] + self.assertEqual(delete_ns, "virt-density-udn-3") + + +class TestVmiNetworkFilterModuleRollback(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.mock_kubernetes = MagicMock() + self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes + self.config = _make_config( + namespace="virt-density-udn-3", + target="virt-server-.*", + test_duration=60, + interfaces=[], + ) + self.module = VmiNetworkFilterModule(self.config, self.mock_kubecli) + + self.mock_kubernetes.get_vmi.return_value = { + "status": {"nodeName": "worker-1"} + } + self.mock_kubernetes.list_pods.return_value = [ + "virt-launcher-virt-server-3-abc12" + ] + + compute = _make_container("compute", ready=True, container_id="containerd://deadbeef") + mock_pod_info = MagicMock() + mock_pod_info.containers = [compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + self.mock_kubernetes.get_pod_pids.return_value = ["100", "101", "102"] + + # ------------------------------------------------------------------ _rollback directly + + def test_rollback_calls_clean_then_delete_when_tc_applied(self): + input_rules = ["nsenter ... iptables -I INPUT 1 -i tap0 -p tcp -j DROP"] + output_rules = ["nsenter ... iptables -I OUTPUT 1 -p tcp -j DROP"] + with patch(f"{MODULE}.clean_tc_vmi_chaos") as mock_clean: + self.module._rollback("ns", "chaos-pod", "101", "tap0", input_rules, output_rules) + + mock_clean.assert_called_once_with( + self.mock_kubernetes, "chaos-pod", "ns", "101", "tap0", input_rules, output_rules + ) + self.mock_kubernetes.delete_pod.assert_called_once_with("chaos-pod", "ns") + + def test_rollback_skips_clean_when_no_rules(self): + with patch(f"{MODULE}.clean_tc_vmi_chaos") as mock_clean: + self.module._rollback("ns", "chaos-pod") + + mock_clean.assert_not_called() + self.mock_kubernetes.delete_pod.assert_called_once_with("chaos-pod", "ns") + + def test_rollback_skips_clean_when_rules_none_but_pid_set(self): + """Chaos pod deployed, netns_pid found, but rules never applied: skip clean.""" + with patch(f"{MODULE}.clean_tc_vmi_chaos") as mock_clean: + self.module._rollback("ns", "chaos-pod", "101", "tap0", None, None) + + mock_clean.assert_not_called() + self.mock_kubernetes.delete_pod.assert_called_once_with("chaos-pod", "ns") + + # ------------------------------------------------------------------ rollback from run: error before tc + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_rollback_deletes_pod_on_error_before_tc( + self, mock_log, mock_deploy, mock_clean + ): + """Pod deployed but setup fails before tc is applied: delete only, no clean.""" + self.mock_kubernetes.get_pod_info.return_value = None + + with self.assertRaises(Exception): + self.module.run("virt-density-udn-3/virt-server-3") + + self.mock_kubernetes.delete_pod.assert_called_once() + mock_clean.assert_not_called() + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.log_info") + def test_run_rollback_does_not_clean_when_no_netns_pid( + self, mock_log, mock_deploy, mock_clean + ): + """No netns_pid resolved means tc was never applied: clean must not be called.""" + with patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value=None): + with self.assertRaises(Exception): + self.module.run("virt-density-udn-3/virt-server-3") + + mock_clean.assert_not_called() + self.mock_kubernetes.delete_pod.assert_called_once() + + # ------------------------------------------------------------------ rollback from run: error after tc + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=(["in_rule"], ["out_rule"])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_rollback_cleans_tc_and_deletes_pod_on_error_after_tc( + self, mock_log, mock_sleep, mock_deploy, mock_find, mock_tap, mock_apply, mock_clean + ): + """If interrupted after tc is applied, both clean and delete must be called.""" + mock_sleep.side_effect = RuntimeError("interrupted") + + with self.assertRaises(RuntimeError): + self.module.run("virt-density-udn-3/virt-server-3") + + mock_clean.assert_called_once() + self.mock_kubernetes.delete_pod.assert_called_once() + + @patch(f"{MODULE}.clean_tc_vmi_chaos") + @patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=(["in_rule"], ["out_rule"])) + @patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") + @patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101") + @patch(f"{MODULE}.deploy_network_chaos_ng_pod") + @patch(f"{MODULE}.time.sleep") + @patch(f"{MODULE}.log_info") + def test_run_rollback_passes_correct_pid_iface_and_rules_to_clean( + self, mock_log, mock_sleep, mock_deploy, mock_find, mock_tap, mock_apply, mock_clean + ): + """Clean must receive the resolved netns_pid, iface, and the rules returned by apply.""" + mock_sleep.side_effect = RuntimeError("interrupted") + + with self.assertRaises(RuntimeError): + self.module.run("virt-density-udn-3/virt-server-3") + + clean_args = mock_clean.call_args[0] + self.assertEqual(clean_args[3], "101") # netns_pid + self.assertEqual(clean_args[4], "tap0") # iface + self.assertEqual(clean_args[5], ["in_rule"]) # input_rules from apply + self.assertEqual(clean_args[6], ["out_rule"]) # output_rules from apply + + +class TestVmiNetworkFilterNetworkMode(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.mock_kubernetes = MagicMock() + self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes + + self.mock_kubernetes.list_pods.return_value = ["virt-launcher-virt-server-3-abc12"] + compute = _make_container("compute", ready=True, container_id="containerd://deadbeef") + mock_pod_info = MagicMock() + mock_pod_info.containers = [compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + self.mock_kubernetes.get_pod_pids.return_value = ["100", "101", "102"] + + def _vmi_with_mode(self, mode: str) -> dict: + binding = {mode: {}} + return { + "status": {"nodeName": "worker-1"}, + "spec": {"domain": {"devices": {"interfaces": [{"name": "default", **binding}]}}}, + } + + def _run_success(self, vmi: dict, **config_overrides): + config = _make_config(**config_overrides) + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = vmi + with patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), \ + patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0"), \ + patch(f"{MODULE}.get_vmi_masquerade_interface", return_value="eth0"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) as mock_apply, \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"), \ + patch(f"{MODULE}.log_info"): + module.run("virt-density-udn-3/virt-server-3") + return mock_apply + + # ------------------------------------------------------------------ mode detection + + def test_bridge_mode_uses_tap_interface(self): + with patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0") as mock_tap, \ + patch(f"{MODULE}.get_vmi_masquerade_interface") as mock_masq, \ + patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])), \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"), \ + patch(f"{MODULE}.log_info"): + config = _make_config() + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = self._vmi_with_mode("bridge") + module.run("virt-density-udn-3/virt-server-3") + + mock_tap.assert_called_once() + mock_masq.assert_not_called() + + def test_masquerade_mode_uses_default_interface(self): + with patch(f"{MODULE}.get_vmi_tap_interface") as mock_tap, \ + patch(f"{MODULE}.get_vmi_masquerade_interface", return_value="eth0") as mock_masq, \ + patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])), \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"), \ + patch(f"{MODULE}.log_info"): + config = _make_config() + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = self._vmi_with_mode("masquerade") + module.run("virt-density-udn-3/virt-server-3") + + mock_masq.assert_called_once() + mock_tap.assert_not_called() + + def test_masquerade_mode_apply_called_with_eth0(self): + mock_apply = self._run_success(self._vmi_with_mode("masquerade")) + iface_arg = mock_apply.call_args[0][4] + self.assertEqual(iface_arg, "eth0") + + def test_bridge_mode_apply_called_with_tap0(self): + mock_apply = self._run_success(self._vmi_with_mode("bridge")) + iface_arg = mock_apply.call_args[0][4] + self.assertEqual(iface_arg, "tap0") + + def test_sriov_mode_raises(self): + config = _make_config() + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = self._vmi_with_mode("sriov") + with patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.log_info"): + with self.assertRaises(Exception) as ctx: + module.run("virt-density-udn-3/virt-server-3") + self.assertIn("sriov", str(ctx.exception)) + self.assertIn("not supported", str(ctx.exception)) + + def test_macvtap_mode_raises(self): + config = _make_config() + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = self._vmi_with_mode("macvtap") + with patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.log_info"): + with self.assertRaises(Exception) as ctx: + module.run("virt-density-udn-3/virt-server-3") + self.assertIn("macvtap", str(ctx.exception)) + self.assertIn("not supported", str(ctx.exception)) + + def test_explicit_interface_skips_mode_detection(self): + """If interfaces is set in config, neither tap nor masquerade detection is called.""" + with patch(f"{MODULE}.get_vmi_tap_interface") as mock_tap, \ + patch(f"{MODULE}.get_vmi_masquerade_interface") as mock_masq, \ + patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) as mock_apply, \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"), \ + patch(f"{MODULE}.log_info"): + config = _make_config(interfaces=["bond0"]) + module = VmiNetworkFilterModule(config, self.mock_kubecli) + self.mock_kubernetes.get_vmi.return_value = self._vmi_with_mode("masquerade") + module.run("virt-density-udn-3/virt-server-3") + + mock_tap.assert_not_called() + mock_masq.assert_not_called() + iface_arg = mock_apply.call_args[0][4] + self.assertEqual(iface_arg, "bond0") + + def test_no_mode_in_spec_defaults_to_bridge(self): + """VMI with no binding in spec should be treated as bridge mode.""" + vmi = {"status": {"nodeName": "worker-1"}, "spec": {}} + mock_apply = self._run_success(vmi) + iface_arg = mock_apply.call_args[0][4] + self.assertEqual(iface_arg, "tap0") + + +class TestVmiNetworkFilterPortsProtocols(unittest.TestCase): + + def setUp(self): + self.mock_kubecli = MagicMock() + self.mock_kubernetes = MagicMock() + self.mock_kubecli.get_lib_kubernetes.return_value = self.mock_kubernetes + + self.mock_kubernetes.get_vmi.return_value = {"status": {"nodeName": "worker-1"}} + self.mock_kubernetes.list_pods.return_value = ["virt-launcher-virt-server-3-abc12"] + compute = _make_container("compute", ready=True, container_id="containerd://deadbeef") + mock_pod_info = MagicMock() + mock_pod_info.containers = [compute] + self.mock_kubernetes.get_pod_info.return_value = mock_pod_info + self.mock_kubernetes.get_pod_pids.return_value = ["100", "101", "102"] + + def _run_and_capture_apply(self, **config_overrides): + config = _make_config(**config_overrides) + module = VmiNetworkFilterModule(config, self.mock_kubecli) + with patch(f"{MODULE}.deploy_network_chaos_ng_pod"), \ + patch(f"{MODULE}.find_virt_launcher_netns_pid", return_value="101"), \ + patch(f"{MODULE}.get_vmi_tap_interface", return_value="tap0"), \ + patch(f"{MODULE}.apply_tc_vmi_chaos", return_value=([], [])) as mock_apply, \ + patch(f"{MODULE}.clean_tc_vmi_chaos"), \ + patch(f"{MODULE}.time.sleep"), \ + patch(f"{MODULE}.log_info"): + module.run("virt-density-udn-3/virt-server-3") + return mock_apply + + def test_apply_receives_specific_ports(self): + mock_apply = self._run_and_capture_apply(ports=[53, 80, 443]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.ports, [53, 80, 443]) + + def test_apply_receives_empty_ports_for_all_traffic(self): + mock_apply = self._run_and_capture_apply(ports=[]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.ports, []) + + def test_apply_receives_tcp_only_protocol(self): + mock_apply = self._run_and_capture_apply(protocols=["tcp"]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.protocols, ["tcp"]) + + def test_apply_receives_udp_only_protocol(self): + mock_apply = self._run_and_capture_apply(protocols=["udp"]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.protocols, ["udp"]) + + def test_apply_receives_both_protocols(self): + mock_apply = self._run_and_capture_apply(protocols=["tcp", "udp"]) + config_arg = mock_apply.call_args[0][5] + self.assertIn("tcp", config_arg.protocols) + self.assertIn("udp", config_arg.protocols) + + def test_apply_receives_dns_ports_with_both_protocols(self): + """DNS blackout: port 53 on tcp and udp.""" + mock_apply = self._run_and_capture_apply(ports=[53], protocols=["tcp", "udp"]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.ports, [53]) + self.assertIn("tcp", config_arg.protocols) + self.assertIn("udp", config_arg.protocols) + + def test_apply_receives_management_ports(self): + """Management plane loss: SSH + HTTPS + k8s API.""" + mock_apply = self._run_and_capture_apply(ports=[22, 443, 6443], protocols=["tcp"]) + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.ports, [22, 443, 6443]) + self.assertEqual(config_arg.protocols, ["tcp"]) + + def test_apply_config_has_resolved_namespace_not_regex(self): + """The config passed to apply must use the real namespace from the target string.""" + mock_apply = self._run_and_capture_apply(namespace="virt-density-udn-.*") + config_arg = mock_apply.call_args[0][5] + self.assertEqual(config_arg.namespace, "virt-density-udn-3") + self.assertNotEqual(config_arg.namespace, "virt-density-udn-.*") + + +if __name__ == "__main__": + unittest.main()