diff --git a/krkn/scenario_plugins/pod_disruption/models/models.py b/krkn/scenario_plugins/pod_disruption/models/models.py index 67adb868..5ffb3fe1 100644 --- a/krkn/scenario_plugins/pod_disruption/models/models.py +++ b/krkn/scenario_plugins/pod_disruption/models/models.py @@ -20,6 +20,9 @@ class InputParams: self.kill = config["kill"] if "kill" in config else 1 self.timeout = config["timeout"] if "timeout" in config else 120 self.duration = config["duration"] if "duration" in config else 10 + self.execution = config["execution"] if "execution" in config else "serial" + if self.execution not in ["serial", "parallel"]: + raise ValueError(f"Unknown execution '{self.execution}' in config. Supported values are: serial, parallel.") self.krkn_pod_recovery_time = config["krkn_pod_recovery_time"] if "krkn_pod_recovery_time" in config else 120 self.label_selector = config["label_selector"] if "label_selector" in config else "" self.namespace_pattern = config["namespace_pattern"] if "namespace_pattern" in config else "" @@ -33,6 +36,7 @@ class InputParams: timeout: int duration: int kill: int + execution: str label_selector: str name_pattern: str node_label_selector: str diff --git a/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py b/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py index 1824ab14..8830bc30 100644 --- a/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py +++ b/krkn/scenario_plugins/pod_disruption/pod_disruption_scenario_plugin.py @@ -14,20 +14,23 @@ import logging import random import time -from asyncio import Future +import queue +import threading import traceback +import concurrent.futures +from asyncio import Future +from datetime import datetime +from dataclasses import dataclass + import yaml from krkn_lib.k8s import KrknKubernetes from krkn_lib.k8s.pod_monitor import select_and_monitor_by_namespace_pattern_and_label, \ select_and_monitor_by_name_pattern_and_namespace_pattern - -from krkn.scenario_plugins.pod_disruption.models.models import InputParams from krkn_lib.models.telemetry import ScenarioTelemetry from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift from krkn_lib.models.pod_monitor.models import PodsSnapshot -from datetime import datetime -from dataclasses import dataclass +from krkn.scenario_plugins.pod_disruption.models.models import InputParams from krkn.scenario_plugins.abstract_scenario_plugin import AbstractScenarioPlugin @dataclass @@ -236,12 +239,20 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): return 1 random.shuffle(pods) + + pods_to_kill = [] for i in range(config.kill): pod = pods[i] logging.info(pod) if pod[0] in exclude_pods: logging.info(f"Excluding {pod[0]} from chaos") else: + pods_to_kill.append(pod) + + if config.execution == "parallel": + self._delete_pods_parallel(pods_to_kill, kubecli) + else: + for pod in pods_to_kill: logging.info(f'Deleting pod {pod[0]}') kubecli.delete_pod(pod[0], pod[1]) @@ -272,3 +283,33 @@ class PodDisruptionScenarioPlugin(AbstractScenarioPlugin): return 1 return 0 + + def _delete_pods_parallel(self, pods: list, kubecli: KrknKubernetes): + """Delete pods concurrently using a thread pool to avoid unbounded threads.""" + error_queue = queue.Queue() + + def _delete(pod): + try: + logging.info(f'[parallel] Deleting pod {pod[0]}') + kubecli.delete_pod(pod[0], pod[1]) + except Exception as exc: + error_queue.put(exc) + + # Cap the number of workers to prevent overloading the apiserver + max_workers = min(10, len(pods)) if pods else 1 + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [executor.submit(_delete, pod) for pod in pods] + concurrent.futures.wait(futures) + + errors = [] + while True: + try: + errors.append(error_queue.get_nowait()) + except queue.Empty: + break + + if errors: + raise Exception( + f"parallel pod deletion failed with {len(errors)} error(s): " + + "; ".join(str(e) for e in errors) + ) diff --git a/scenarios/kube/pod.yml b/scenarios/kube/pod.yml index 29d931be..7eb21bc5 100644 --- a/scenarios/kube/pod.yml +++ b/scenarios/kube/pod.yml @@ -4,4 +4,5 @@ name_pattern: ^nginx-.*$ namespace_pattern: ^default$ kill: 1 + # execution: serial # optional: serial (default) | parallel krkn_pod_recovery_time: 120 diff --git a/scenarios/openshift/etcd_quorum_loss.yml b/scenarios/openshift/etcd_quorum_loss.yml new file mode 100644 index 00000000..c19937fe --- /dev/null +++ b/scenarios/openshift/etcd_quorum_loss.yml @@ -0,0 +1,9 @@ +# yaml-language-server: $schema=../plugin.schema.json +- id: kill-pods + config: + namespace_pattern: ^openshift-etcd$ + label_selector: k8s-app=etcd + kill: 2 + execution: parallel # optional: serial (default) | parallel + krkn_pod_recovery_time: 120 + exclude_label: "" # excludes pods marked with this label from chaos diff --git a/tests/test_pod_disruption_scenario_plugin.py b/tests/test_pod_disruption_scenario_plugin.py index 30a664f8..61f2f682 100644 --- a/tests/test_pod_disruption_scenario_plugin.py +++ b/tests/test_pod_disruption_scenario_plugin.py @@ -16,6 +16,7 @@ from krkn_lib.k8s import KrknKubernetes from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift from krkn.scenario_plugins.pod_disruption.pod_disruption_scenario_plugin import PodDisruptionScenarioPlugin +from krkn.scenario_plugins.pod_disruption.models.models import InputParams class TestPodDisruptionScenarioPlugin(unittest.TestCase): @@ -39,6 +40,136 @@ class TestPodDisruptionScenarioPlugin(unittest.TestCase): self.assertEqual(result, ["pod_disruption_scenarios"]) self.assertEqual(len(result), 1) +class TestKillingPodsMode(unittest.TestCase): + def setUp(self): + """Set up test fixtures for killing_pods mode tests.""" + self.plugin = PodDisruptionScenarioPlugin() + self.kubecli = MagicMock(spec=KrknKubernetes) + self.plugin.get_pods = MagicMock() + self.plugin.wait_for_pods = MagicMock(return_value=0) + + def tearDown(self): + """Clean up after each test to prevent state leakage.""" + self.plugin = None + self.kubecli = None + + # --- InputParams.execution parsing --- + + def test_execution_defaults_to_serial(self): + """execution defaults to 'serial' when not specified in config.""" + params = InputParams({"kill": 2}) + self.assertEqual(params.execution, "serial") + + def test_execution_serial_explicit(self): + """execution is correctly parsed when explicitly set to 'serial'.""" + params = InputParams({"kill": 2, "execution": "serial"}) + self.assertEqual(params.execution, "serial") + + def test_execution_parallel_explicit(self): + """execution is correctly parsed when explicitly set to 'parallel'.""" + params = InputParams({"kill": 2, "execution": "parallel"}) + self.assertEqual(params.execution, "parallel") + + def test_execution_invalid_raises_value_error(self): + """execution raises ValueError on unknown values.""" + with self.assertRaises(ValueError) as context: + InputParams({"kill": 2, "execution": "invalid_mode"}) + + self.assertIn("Unknown execution 'invalid_mode'", str(context.exception)) + + # --- killing_pods() behaviour --- + + def test_not_enough_pods_returns_error(self): + """Returns 1 and never calls delete_pod when fewer pods exist than kill count.""" + config = InputParams({"kill": 3, "execution": "serial"}) + self.plugin.get_pods.return_value = [("pod1", "ns1"), ("pod2", "ns1")] + + result = self.plugin.killing_pods(config, self.kubecli) + + self.assertEqual(result, 1) + self.kubecli.delete_pod.assert_not_called() + + def test_serial_mode_calls_delete_in_order(self): + """Serial mode deletes all selected pods one at a time.""" + config = InputParams({"kill": 2, "execution": "serial"}) + self.plugin.get_pods.return_value = [("pod1", "ns1"), ("pod2", "ns1")] + + result = self.plugin.killing_pods(config, self.kubecli) + + self.assertEqual(result, 0) + self.assertEqual(self.kubecli.delete_pod.call_count, 2) + self.kubecli.delete_pod.assert_any_call("pod1", "ns1") + self.kubecli.delete_pod.assert_any_call("pod2", "ns1") + + def test_parallel_mode_calls_delete_concurrently(self): + """Parallel mode deletes all selected pods and calls delete_pod for each concurrently.""" + config = InputParams({"kill": 2, "execution": "parallel"}) + pods = [("pod1", "ns1"), ("pod2", "ns1")] + self.plugin.get_pods.return_value = pods + + # Use a barrier to prove threads run concurrently. If they run serially, + # the first thread will block forever waiting for the second. + import threading + barrier = threading.Barrier(2, timeout=5) + + def side_effect(name, namespace): + barrier.wait() + + self.kubecli.delete_pod.side_effect = side_effect + + result = self.plugin.killing_pods(config, self.kubecli) + + self.assertEqual(result, 0) + self.assertEqual(self.kubecli.delete_pod.call_count, 2) + self.kubecli.delete_pod.assert_any_call("pod1", "ns1") + self.kubecli.delete_pod.assert_any_call("pod2", "ns1") + + def test_parallel_mode_propagates_delete_exception(self): + """Exceptions raised during parallel deletion bubble up correctly.""" + config = InputParams({"kill": 2, "execution": "parallel"}) + self.plugin.get_pods.return_value = [("pod1", "ns1"), ("pod2", "ns1")] + + def side_effect(name, namespace): + if name == "pod1": + raise RuntimeError("failed to delete") + + self.kubecli.delete_pod.side_effect = side_effect + + with self.assertRaises(Exception) as context: + self.plugin.killing_pods(config, self.kubecli) + + self.assertIn("parallel pod deletion failed", str(context.exception)) + self.assertIn("failed to delete", str(context.exception)) + + def test_excluded_pods_are_not_deleted_in_serial_mode(self): + """Pods matched by exclude_label are skipped and never passed to delete_pod (serial).""" + config = InputParams({"kill": 2, "execution": "serial", "exclude_label": "protected=true"}) + # get_pods is called twice: first for target pods, then for excluded pods + self.plugin.get_pods.side_effect = [ + [("pod1", "ns1"), ("pod2", "ns1")], # target pods + [("pod1", "ns1")], # excluded pods + ] + + result = self.plugin.killing_pods(config, self.kubecli) + + self.assertEqual(result, 0) + # Only pod2 should be deleted; pod1 is excluded + self.kubecli.delete_pod.assert_called_once_with("pod2", "ns1") + + def test_excluded_pods_are_not_deleted_in_parallel_mode(self): + """Pods matched by exclude_label are skipped and never passed to delete_pod (parallel).""" + config = InputParams({"kill": 2, "execution": "parallel", "exclude_label": "protected=true"}) + self.plugin.get_pods.side_effect = [ + [("pod1", "ns1"), ("pod2", "ns1")], # target pods + [("pod1", "ns1")], # excluded pods + ] + + result = self.plugin.killing_pods(config, self.kubecli) + + self.assertEqual(result, 0) + # Only pod2 should be deleted; pod1 is excluded + self.kubecli.delete_pod.assert_called_once_with("pod2", "ns1") + if __name__ == "__main__": unittest.main()