mirror of
https://github.com/krkn-chaos/krkn.git
synced 2026-08-25 09:27:36 +00:00
feat(pod-disruption): support parallel pod deletion (#1536)
* feat(pod-disruption): support parallel pod deletion Introduce 'kill_mode' field to pod disruption scenario to support parallel deletion of pods. By default 'kill_mode' is 'sequential' preserving existing behavior. When set to 'parallel', it concurrently deletes pods using threading and queue, enabling effective testing of disruption scenarios like etcd quorum loss where simultaneous disruption is necessary. Resolves: #1516 Assisted-by: Claude <noreply@anthropic.com> Signed-off-by: swagatobauri <swagato731123@gmail.com> * fix(pod-disruption): validate kill_mode and address PR feedback - Validate 'kill_mode' at parse time: fallback to 'sequential' and log a warning if an unknown string is provided, preventing silent failures. - Update test_parallel_mode_calls_delete_concurrently to use a threading.Barrier to definitively prove concurrent thread execution. - Revert 'scenarios/openshift/etcd.yml' to default non-destructive behavior and extract parallel quorum loss example into a new explicitly named file 'etcd_quorum_loss.yml'. Assisted-by: Claude <noreply@anthropic.com> Signed-off-by: swagatobauri <swagato731123@gmail.com> * refactor(pod-disruption): optimize imports and cap threads - Reorganize imports in pod_disruption_scenario_plugin.py to comply with standard PEP-8 grouping (stdlib, third-party, local). - Replace unbounded OS thread spawning in _delete_pods_parallel with concurrent.futures.ThreadPoolExecutor. Capped max_workers to 10 to prevent excessive concurrent API calls that could overload the Kubernetes API server under aggressive scenario configurations. Assisted-by: Claude <noreply@anthropic.com> Signed-off-by: swagatobauri <swagato731123@gmail.com> * refactor(pod-disruption): align execution config with network_chaos_ng - Renamed config field from 'kill_mode' to 'execution' for consistency with the network_chaos_ng plugin. - Changed valid values from 'sequential|parallel' to 'serial|parallel'. - Replaced fallback warning with strict validation, raising ValueError on invalid execution values, aligning with the strict validation pattern used elsewhere in the codebase. - Updated relevant tests and scenario YAML templates. Assisted-by: Claude <noreply@anthropic.com> Signed-off-by: swagatobauri <swagato731123@gmail.com> * fix(pod-disruption): remove unused logging import in models.py Following the switch from a warning log to raising a ValueError for invalid execution types, the logging import was no longer used. Assisted-by: Claude <noreply@anthropic.com> Signed-off-by: swagatobauri <swagato731123@gmail.com> --------- Signed-off-by: swagatobauri <swagato731123@gmail.com> Co-authored-by: Darshan Jain <darjain@redhat.com>
This commit is contained in:
SWAGATO BAURI
co-authored by
Darshan Jain
Darshan Jain
parent
502cf958e6
commit
ee32a179cf
@@ -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
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -4,4 +4,5 @@
|
||||
name_pattern: ^nginx-.*$
|
||||
namespace_pattern: ^default$
|
||||
kill: 1
|
||||
# execution: serial # optional: serial (default) | parallel
|
||||
krkn_pod_recovery_time: 120
|
||||
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user