From eb2efa84eab3f6f7e19cc3600b6b8a4b1bc61496 Mon Sep 17 00:00:00 2001 From: Darshan Jain Date: Tue, 19 May 2026 00:24:55 +0530 Subject: [PATCH] feat: storage I/O throttle scenario (cgroups v1/v2) for PVC-backed workloads (#1296) * feat(storage-throttle): add storage throttle scenario and tests Consolidate the storage-throttle implementation, scenario configs, CI v2 coverage, and krkn-lib 6.0.10 dependency update into a single signed commit for cleaner PR history. Signed-off-by: ddjain Co-authored-by: Cursor Signed-off-by: ddjain * adding need dco auto review (#1329) Signed-off-by: Paige Patton Signed-off-by: ddjain * fix: start_klusterlet_scenario action calls start instead of stop (#1324) The start_klusterlet_scenario branch in inject_managedcluster_scenario was calling stop_klusterlet_scenario on the scenarios object instead of start_klusterlet_scenario. Any user configuring this action would stop the klusterlet (scale to 0) instead of starting it (scale to 3). Fixes #1323 Signed-off-by: v0idheaven Co-authored-by: Paige Patton <64206430+paigerube14@users.noreply.github.com> Signed-off-by: ddjain * adding links to issue of completed roadmap items (#1328) Signed-off-by: Paige Patton Signed-off-by: ddjain * container scenario template image update (#1342) Signed-off-by: Paige Patton Signed-off-by: ddjain --------- Signed-off-by: ddjain Signed-off-by: Paige Patton Signed-off-by: v0idheaven Co-authored-by: Cursor Co-authored-by: Paige Patton <64206430+paigerube14@users.noreply.github.com> Co-authored-by: Varun Yadav --- CI/tests_v2/pytest.ini | 1 + .../scenarios/storage_throttle/resource.yaml | 39 + .../storage_throttle/scenario_base.yaml | 13 + .../storage_throttle/test_storage_throttle.py | 139 ++ config/config.yaml | 2 + .../storage_throttle/__init__.py | 0 .../storage_throttle_scenario_plugin.py | 774 +++++++++++ .../storage_throttle_utils.py | 147 +++ requirements.txt | 2 +- scenarios/kind/storage_throttle.yaml | 12 + scenarios/kube/storage_throttle.yaml | 12 + scenarios/openshift/storage_throttle.yaml | 12 + .../test_storage_throttle_scenario_plugin.py | 1159 +++++++++++++++++ 13 files changed, 2311 insertions(+), 1 deletion(-) create mode 100644 CI/tests_v2/scenarios/storage_throttle/resource.yaml create mode 100644 CI/tests_v2/scenarios/storage_throttle/scenario_base.yaml create mode 100644 CI/tests_v2/scenarios/storage_throttle/test_storage_throttle.py create mode 100644 krkn/scenario_plugins/storage_throttle/__init__.py create mode 100644 krkn/scenario_plugins/storage_throttle/storage_throttle_scenario_plugin.py create mode 100644 krkn/scenario_plugins/storage_throttle/storage_throttle_utils.py create mode 100644 scenarios/kind/storage_throttle.yaml create mode 100644 scenarios/kube/storage_throttle.yaml create mode 100644 scenarios/openshift/storage_throttle.yaml create mode 100644 tests/test_storage_throttle_scenario_plugin.py diff --git a/CI/tests_v2/pytest.ini b/CI/tests_v2/pytest.ini index 888ae5cb..db6bc9b6 100644 --- a/CI/tests_v2/pytest.ini +++ b/CI/tests_v2/pytest.ini @@ -9,6 +9,7 @@ markers = functional: marks a test as a functional test (deselect with '-m "not functional"') pod_disruption: marks a test as a pod disruption scenario test application_outage: marks a test as an application outage scenario test + storage_throttle: marks a test as a storage throttle scenario test no_workload: skip workload deployment for this test (e.g. negative tests) order: set test order (pytest-order) junit_family = xunit2 diff --git a/CI/tests_v2/scenarios/storage_throttle/resource.yaml b/CI/tests_v2/scenarios/storage_throttle/resource.yaml new file mode 100644 index 00000000..cb6fceb3 --- /dev/null +++ b/CI/tests_v2/scenarios/storage_throttle/resource.yaml @@ -0,0 +1,39 @@ +# PVC + Deployment for storage throttle integration test. +# Namespace is patched at deploy time by the test framework. +# Uses the default StorageClass (e.g. local-path on KinD). +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: krkn-throttle-pvc +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 128Mi +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: krkn-throttle-target +spec: + replicas: 1 + selector: + matchLabels: + app: krkn-throttle-target + template: + metadata: + labels: + app: krkn-throttle-target + spec: + containers: + - name: app + image: nginx:alpine + command: ["/bin/sh", "-c", "trap : TERM INT; sleep infinity & wait"] + volumeMounts: + - name: data + mountPath: /data + volumes: + - name: data + persistentVolumeClaim: + claimName: krkn-throttle-pvc diff --git a/CI/tests_v2/scenarios/storage_throttle/scenario_base.yaml b/CI/tests_v2/scenarios/storage_throttle/scenario_base.yaml new file mode 100644 index 00000000..64099751 --- /dev/null +++ b/CI/tests_v2/scenarios/storage_throttle/scenario_base.yaml @@ -0,0 +1,13 @@ +# Base storage_throttle scenario. Tests load this and patch namespace, pvc_name, etc. +storage_throttle_scenario: + pvc_name: krkn-throttle-pvc + pod_name: "" + namespace: default + mount_path: /data + throttle_type: bandwidth + read_bps: 1Mi + write_bps: 512Ki + read_iops: 100 + write_iops: 50 + duration: 15 + image: quay.io/krkn-chaos/krkn:tools diff --git a/CI/tests_v2/scenarios/storage_throttle/test_storage_throttle.py b/CI/tests_v2/scenarios/storage_throttle/test_storage_throttle.py new file mode 100644 index 00000000..0c019487 --- /dev/null +++ b/CI/tests_v2/scenarios/storage_throttle/test_storage_throttle.py @@ -0,0 +1,139 @@ +""" +Functional test for storage throttle scenario (cgroup I/O throttle on PVC-backed volume). + +Deploys a PVC + Deployment in an ephemeral namespace, runs the storage_throttle +scenario, and verifies: + - Krkn exits 0 (throttle applied and removed cleanly) + - Target pods survive (running and ready after scenario) + - Negative cases: bad namespace and invalid throttle_type fail gracefully + +Follows the CI/tests_v2 BaseScenarioTest pattern. +""" + +import pytest + +from lib.base import BaseScenarioTest, READINESS_TIMEOUT +from lib.utils import ( + assert_all_pods_running_and_ready, + assert_kraken_failure, + assert_kraken_success, + assert_pod_count_unchanged, + get_pods_list, +) + + +@pytest.mark.functional +@pytest.mark.storage_throttle +class TestStorageThrottle(BaseScenarioTest): + """Storage throttle scenario: apply I/O cgroup limits on a PVC mount and verify recovery.""" + + WORKLOAD_MANIFEST = "CI/tests_v2/scenarios/storage_throttle/resource.yaml" + WORKLOAD_IS_PATH = True + LABEL_SELECTOR = "app=krkn-throttle-target" + SCENARIO_NAME = "storage_throttle" + SCENARIO_TYPE = "storage_throttle_scenarios" + NAMESPACE_KEY_PATH = ["storage_throttle_scenario", "namespace"] + NAMESPACE_IS_REGEX = False + OVERRIDES_KEY_PATH = ["storage_throttle_scenario"] + + @pytest.mark.order(1) + def test_bandwidth_throttle_and_recovery(self, wait_for_pods_running): + """Bandwidth throttle: apply read/write bps limits, verify Krkn success and pod recovery.""" + ns = self.ns + before = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + + result = self.run_scenario(self.tmp_path, ns, overrides={ + "throttle_type": "bandwidth", + "read_bps": "1Mi", + "write_bps": "512Ki", + "duration": 15, + }) + assert_kraken_success(result, context=f"bandwidth namespace={ns}", tmp_path=self.tmp_path) + + wait_for_pods_running(ns, self.LABEL_SELECTOR, timeout=READINESS_TIMEOUT) + after = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + assert_pod_count_unchanged(before, after, namespace=ns) + assert_all_pods_running_and_ready(after, namespace=ns) + + @pytest.mark.order(2) + def test_iops_throttle_and_recovery(self, wait_for_pods_running): + """IOPS throttle: apply read/write iops limits, verify Krkn success and pod recovery.""" + ns = self.ns + before = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + + result = self.run_scenario( + self.tmp_path, ns, + overrides={ + "throttle_type": "iops", + "read_iops": 50, + "write_iops": 25, + "duration": 15, + }, + config_filename="test_iops_config.yaml", + ) + assert_kraken_success(result, context=f"iops namespace={ns}", tmp_path=self.tmp_path) + + wait_for_pods_running(ns, self.LABEL_SELECTOR, timeout=READINESS_TIMEOUT) + after = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + assert_pod_count_unchanged(before, after, namespace=ns) + assert_all_pods_running_and_ready(after, namespace=ns) + + @pytest.mark.order(3) + def test_both_throttle_and_recovery(self, wait_for_pods_running): + """Combined throttle: apply both bps and iops limits, verify Krkn success and pod recovery.""" + ns = self.ns + before = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + + result = self.run_scenario( + self.tmp_path, ns, + overrides={ + "throttle_type": "both", + "read_bps": "1Mi", + "write_bps": "512Ki", + "read_iops": 50, + "write_iops": 25, + "duration": 15, + }, + config_filename="test_both_config.yaml", + ) + assert_kraken_success(result, context=f"both namespace={ns}", tmp_path=self.tmp_path) + + wait_for_pods_running(ns, self.LABEL_SELECTOR, timeout=READINESS_TIMEOUT) + after = get_pods_list(self.k8s_core, ns, self.LABEL_SELECTOR) + assert_pod_count_unchanged(before, after, namespace=ns) + assert_all_pods_running_and_ready(after, namespace=ns) + + @pytest.mark.no_workload + def test_bad_namespace_fails(self): + """Scenario targeting non-existent namespace causes Krkn to exit non-zero.""" + scenario = self.load_and_patch_scenario( + self.repo_root, "nonexistent-namespace-xyz-99999", + pod_name="nonexistent-pod", + ) + scenario_path = self.write_scenario(self.tmp_path, scenario, suffix="_bad_ns") + config_path = self.build_config( + self.SCENARIO_TYPE, str(scenario_path), + filename="storage_throttle_bad_ns_config.yaml", + ) + result = self.run_kraken(config_path) + assert_kraken_failure( + result, context=f"bad namespace test", tmp_path=self.tmp_path, + ) + + @pytest.mark.no_workload + def test_invalid_throttle_type_fails(self): + """Invalid throttle_type causes Krkn to exit non-zero.""" + scenario = self.load_and_patch_scenario( + self.repo_root, self.ns, + throttle_type="invalid_type", + pod_name="doesnt-matter", + ) + scenario_path = self.write_scenario(self.tmp_path, scenario, suffix="_bad_type") + config_path = self.build_config( + self.SCENARIO_TYPE, str(scenario_path), + filename="storage_throttle_bad_type_config.yaml", + ) + result = self.run_kraken(config_path) + assert_kraken_failure( + result, context=f"invalid throttle_type test", tmp_path=self.tmp_path, + ) diff --git a/config/config.yaml b/config/config.yaml index 49bc6ca7..fbac08ee 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -41,6 +41,8 @@ kraken: - scenarios/openshift/zone_outage.yaml - pvc_scenarios: - scenarios/openshift/pvc_scenario.yaml + - storage_throttle_scenarios: + - scenarios/openshift/storage_throttle.yaml - network_chaos_scenarios: - scenarios/openshift/network_chaos.yaml - service_hijacking_scenarios: diff --git a/krkn/scenario_plugins/storage_throttle/__init__.py b/krkn/scenario_plugins/storage_throttle/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/krkn/scenario_plugins/storage_throttle/storage_throttle_scenario_plugin.py b/krkn/scenario_plugins/storage_throttle/storage_throttle_scenario_plugin.py new file mode 100644 index 00000000..5936061c --- /dev/null +++ b/krkn/scenario_plugins/storage_throttle/storage_throttle_scenario_plugin.py @@ -0,0 +1,774 @@ +# Copyright 2026 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 base64 +import json +import logging +import random +import time +import traceback +from dataclasses import dataclass +from typing import Optional + +import yaml +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.models.telemetry import ScenarioTelemetry +from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift +from krkn_lib.utils import get_yaml_item_value + +from krkn.scenario_plugins.abstract_scenario_plugin import AbstractScenarioPlugin +from krkn.rollback.config import RollbackContent +from krkn.rollback.handler import set_rollback_context_decorator +from krkn.scenario_plugins.storage_throttle.storage_throttle_utils import ( + parse_byte_value, + parse_duration_value, + validate_cgroup_path, + validate_container_id, + validate_maj_min, + validate_mount_path, +) + +# Backward-compatible aliases so existing imports of the underscore-prefixed +# names from this module continue to work (e.g. in tests). +_validate_mount_path = validate_mount_path +_validate_cgroup_path = validate_cgroup_path +_validate_container_id = validate_container_id +_validate_maj_min = validate_maj_min +_parse_byte_value = parse_byte_value +_parse_duration_value = parse_duration_value + + +@dataclass(frozen=True) +class ThrottleParams: + """Parsed and validated scenario configuration for storage throttle.""" + pvc_name: str + pod_name: str + namespace: str + throttle_type: str + read_iops: int + write_iops: int + read_bps: int + write_bps: int + duration: int + mount_path: str + image: str + + +class StorageThrottleScenarioPlugin(AbstractScenarioPlugin): + """Chaos scenario that throttles I/O on PVC-backed volumes via Linux cgroups (v1/v2).""" + + DEFAULT_IMAGE = "quay.io/krkn-chaos/krkn:tools" + _V1_BLKIO_FILES = [ + "blkio.throttle.read_bps_device", + "blkio.throttle.write_bps_device", + "blkio.throttle.read_iops_device", + "blkio.throttle.write_iops_device", + ] + + def __init__(self, scenario_type: str = None): + super().__init__(scenario_type="storage_throttle_scenarios") + + # ------------------------------------------------------------------ + # Config parsing + # ------------------------------------------------------------------ + + def _parse_scenario_config( + self, scenario: str + ) -> Optional[ThrottleParams]: + """Parse and validate the scenario YAML file. + + Returns a ThrottleParams on success or None on validation failure + (errors are logged). + """ + with open(scenario, "r") as f: + config_yaml = yaml.safe_load(f) + + scenario_config = config_yaml["storage_throttle_scenario"] + pvc_name = get_yaml_item_value(scenario_config, "pvc_name", "") + pod_name = get_yaml_item_value(scenario_config, "pod_name", "") + namespace = get_yaml_item_value(scenario_config, "namespace", "") + throttle_type = get_yaml_item_value( + scenario_config, "throttle_type", "bandwidth" + ) + try: + read_iops = int( + get_yaml_item_value(scenario_config, "read_iops", 100) + ) + write_iops = int( + get_yaml_item_value(scenario_config, "write_iops", 50) + ) + read_bps = parse_byte_value( + get_yaml_item_value(scenario_config, "read_bps", 1048576) + ) + write_bps = parse_byte_value( + get_yaml_item_value(scenario_config, "write_bps", 524288) + ) + duration = parse_duration_value( + get_yaml_item_value(scenario_config, "duration", 60) + ) + except (ValueError, TypeError) as exc: + logging.error("Invalid numeric config value: %s", exc) + return None + mount_path = get_yaml_item_value( + scenario_config, "mount_path", "" + ) + image = get_yaml_item_value( + scenario_config, "image", self.DEFAULT_IMAGE + ) + + if not namespace: + logging.error("You must specify the namespace") + return None + if not pvc_name and not pod_name: + logging.error("You must specify pvc_name or pod_name") + return None + if throttle_type not in ("iops", "bandwidth", "both"): + logging.error( + "throttle_type must be 'iops', 'bandwidth', or 'both', " + "got '%s'" % throttle_type + ) + return None + if mount_path and not validate_mount_path(str(mount_path)): + logging.error( + "mount_path contains invalid characters or format: %r. " + "Use an absolute path with only letters, digits, " + "._/- (e.g. /data)" % mount_path + ) + return None + for name, val in [ + ("read_iops", read_iops), ("write_iops", write_iops), + ("read_bps", read_bps), ("write_bps", write_bps), + ("duration", duration), + ]: + if val <= 0: + logging.error( + "%s must be a positive value, got %d" % (name, val) + ) + return None + + return ThrottleParams( + pvc_name=pvc_name, + pod_name=pod_name, + namespace=namespace, + throttle_type=throttle_type, + read_iops=read_iops, + write_iops=write_iops, + read_bps=read_bps, + write_bps=write_bps, + duration=duration, + mount_path=mount_path, + image=image, + ) + + # ------------------------------------------------------------------ + # Main entry point + # ------------------------------------------------------------------ + + @set_rollback_context_decorator + def run( + self, + run_uuid: str, + scenario: str, + lib_telemetry: KrknTelemetryOpenshift, + scenario_telemetry: ScenarioTelemetry, + ) -> int: + try: + params = self._parse_scenario_config(scenario) + if params is None: + return 1 + + lib_k8s = lib_telemetry.get_lib_kubernetes() + + pod_name = self._resolve_pod_name( + lib_k8s, params.pvc_name, params.pod_name, params.namespace + ) + if not pod_name: + return 1 + + pod = lib_k8s.get_pod_info( + name=pod_name, namespace=params.namespace + ) + if pod is None: + logging.error( + "Pod '%s' doesn't exist in namespace '%s'" + % (pod_name, params.namespace) + ) + return 1 + + container_name, vol_mount_path = self._find_pvc_mount( + pod, params.mount_path, params.pvc_name + ) + if not container_name: + logging.error( + "Pod '%s' has no PVC volume mount" % pod_name + ) + return 1 + if not validate_mount_path(vol_mount_path): + logging.error( + "Resolved mount path is invalid or unsafe: %r" + % vol_mount_path + ) + return 1 + + logging.info( + "Target: pod=%s container=%s mount=%s namespace=%s" + % (pod_name, container_name, vol_mount_path, params.namespace) + ) + + maj_min = self._get_device_maj_min( + lib_k8s, pod_name, params.namespace, + container_name, vol_mount_path, + ) + if not maj_min: + logging.error( + "Could not determine major:minor for mount %s" + % vol_mount_path + ) + return 1 + if not validate_maj_min(maj_min): + logging.error( + "Device major:minor from mountinfo is invalid: %r" + % maj_min + ) + return 1 + logging.info("Block device major:minor = %s" % maj_min) + + node_name = pod.nodeName + if not node_name: + logging.error( + "Pod '%s' has no nodeName yet (still pending?)" + % pod_name + ) + return 1 + logging.info("Target node: %s" % node_name) + + container_id = self._get_container_id(pod, container_name) + if not container_id: + logging.error( + "Could not get container ID for %s" % pod_name + ) + return 1 + logging.info("Container ID: %s" % container_id) + + # Deploy privileged pod — all subsequent work is guarded by + # try/finally so the pod is always cleaned up. + priv_pod_name = lib_k8s.deploy_io_throttle_pod( + node_name=node_name, + image=params.image, + namespace=params.namespace, + ) + logging.info("Privileged pod deployed: %s" % priv_pod_name) + + throttle_applied = False + cgroup_path = "" + cgroup_version = "v1" + try: + cgroup_version = self._detect_cgroup_version( + lib_k8s, priv_pod_name, params.namespace + ) + logging.info("Detected cgroups %s" % cgroup_version) + + cgroup_path = self._find_host_cgroup_path( + lib_k8s, priv_pod_name, container_id, + cgroup_version, params.namespace, + ) + if not cgroup_path: + logging.error( + "Could not find host cgroup path for container %s" + % container_id + ) + return 1 + if not validate_cgroup_path(cgroup_path): + logging.error( + "Discovered cgroup path is invalid or unsafe: %r" + % cgroup_path + ) + return 1 + logging.info("Host cgroup path: %s" % cgroup_path) + + self._register_rollback( + priv_pod_name, maj_min, cgroup_path, + cgroup_version, params.namespace, + ) + + self._apply_throttle( + lib_k8s, priv_pod_name, cgroup_path, + cgroup_version, maj_min, params, params.namespace, + ) + throttle_applied = True + logging.info( + "I/O throttle applied (type=%s) for %ds" + % (params.throttle_type, params.duration) + ) + + self._wait_with_progress(params.duration) + + self._remove_throttle( + lib_k8s, priv_pod_name, cgroup_path, + cgroup_version, maj_min, params.namespace, + ) + throttle_applied = False + logging.info("I/O throttle removed") + finally: + if throttle_applied: + try: + self._remove_throttle( + lib_k8s, priv_pod_name, cgroup_path, + cgroup_version, maj_min, params.namespace, + ) + logging.info("I/O throttle removed during cleanup") + except Exception as e: + logging.warning( + "Best-effort throttle removal failed: %s. " + "I/O limits may persist on device %s at " + "cgroup path %s (%s). To clear manually, " + "run a privileged pod on the node and reset " + "the cgroup: %s" + % ( + e, maj_min, cgroup_path, cgroup_version, + "echo '%s rbps=max wbps=max riops=max wiops=max'" + " > /sys/fs/cgroup%s/io.max" % (maj_min, cgroup_path) + if cgroup_version == "v2" + else "echo '%s 0' > /sys/fs/cgroup/blkio%s/" + "" % (maj_min, cgroup_path), + ) + ) + self._cleanup_privileged_pod( + lib_k8s, priv_pod_name, params.namespace + ) + logging.info("Privileged pod cleaned up") + + except Exception as e: + logging.error("Stack trace:\n%s", traceback.format_exc()) + logging.error( + "StorageThrottleScenarioPlugin exception: %s" % e + ) + return 1 + else: + return 0 + + def get_scenario_types(self) -> list[str]: + return ["storage_throttle_scenarios"] + + def _resolve_pod_name( + self, + lib_k8s: KrknKubernetes, + pvc_name: str, + pod_name: str, + namespace: str, + ) -> str: + if pvc_name: + if pod_name: + logging.info( + "pod_name '%s' will be overridden by pod from PVC" % pod_name + ) + pvc = lib_k8s.get_pvc_info(pvc_name, namespace) + if pvc is None or not pvc.podNames: + logging.error( + "No pod associated with PVC '%s' in namespace '%s'" + % (pvc_name, namespace) + ) + return "" + pod_name = random.choice(pvc.podNames) # nosec + logging.info("Resolved pod from PVC: %s" % pod_name) + return pod_name + + @staticmethod + def _find_pvc_mount(pod, mount_path: str, pvc_name: str = ""): + """Find the container name and mount path for a PVC volume. + + When *pvc_name* is set, only volumes backed by that PVC are considered. + """ + for volume in pod.volumes: + if volume.pvcName is None: + continue + if pvc_name and volume.pvcName != pvc_name: + continue + vol_name = volume.name + for container in pod.containers: + for vol_mount in container.volumeMounts: + if vol_mount.name == vol_name: + if mount_path and vol_mount.mountPath != mount_path: + continue + return container.name, vol_mount.mountPath + return None, None + + @staticmethod + def _get_device_maj_min( + lib_k8s: KrknKubernetes, + pod_name: str, + namespace: str, + container_name: str, + mount_path: str, + ) -> str: + """Extract device major:minor from /proc/self/mountinfo.""" + if not _validate_mount_path(mount_path): + logging.error("Refusing to exec with invalid mount_path: %r", mount_path) + return "" + cmd = "grep -F ' %s ' /proc/self/mountinfo | awk '{print $3}' | head -1" + output = lib_k8s.exec_cmd_in_pod( + [cmd % mount_path], pod_name, namespace, container_name + ) + if output: + return output.strip() + return "" + + @staticmethod + def _get_container_id(pod, container_name: str) -> str: + """Get the container ID from krkn_lib Pod.containers, stripping runtime prefix.""" + if pod is None: + return "" + for c in pod.containers: + if c.name == container_name: + cid = c.containerId or "" + if not cid: + return "" + if "://" in cid: + cid = cid.split("://", 1)[1] + if not validate_container_id(cid): + logging.warning("Container ID is not valid hex: %r", cid) + return "" + return cid + return "" + + def _cleanup_privileged_pod( + self, lib_k8s: KrknKubernetes, priv_pod_name: str, namespace: str + ): + """Delete the privileged pod.""" + try: + lib_k8s.delete_pod(priv_pod_name, namespace) + except Exception as e: + logging.warning("Failed to delete privileged pod %s: %s" % (priv_pod_name, e)) + + def _detect_cgroup_version( + self, lib_k8s: KrknKubernetes, priv_pod_name: str, namespace: str + ) -> str: + """Detect whether the node uses cgroups v1 or v2.""" + output = lib_k8s.exec_cmd_in_pod( + ["/host", "stat", "-f", "-c", "%T", "/sys/fs/cgroup"], + priv_pod_name, + namespace, + base_command="chroot", + ) + if output and "cgroup2fs" in output: + return "v2" + return "v1" + + def _find_host_cgroup_path( + self, + lib_k8s: KrknKubernetes, + priv_pod_name: str, + container_id: str, + cgroup_version: str, + namespace: str, + ) -> str: + """ + Find the real cgroup path on the host for the target container. + Excludes conmon (CRI-O container monitor) paths. + """ + short_id = container_id[:12] + + if cgroup_version == "v2": + search_base = "/sys/fs/cgroup" + strip_prefix = "/sys/fs/cgroup" + else: + search_base = "/sys/fs/cgroup/blkio" + strip_prefix = "/sys/fs/cgroup/blkio" + + # Search for the container's scope directory, excluding conmon + find_cmd = ( + "find %s -name '*.scope' -path '*%s*' ! -name '*conmon*' 2>/dev/null | head -1" + % (search_base, short_id) + ) + output = lib_k8s.exec_cmd_in_pod( + ["/host", "bash", "-c", find_cmd], + priv_pod_name, + namespace, + base_command="chroot", + ) + if output and output.strip(): + return output.strip().replace(strip_prefix, "", 1) + + # Fallback: search directories (containerd doesn't use .scope files) + find_cmd = ( + "find %s -type d -name '*%s*' ! -name '*conmon*' 2>/dev/null | head -1" + % (search_base, short_id) + ) + output = lib_k8s.exec_cmd_in_pod( + ["/host", "bash", "-c", find_cmd], + priv_pod_name, + namespace, + base_command="chroot", + ) + if output and output.strip(): + return output.strip().replace(strip_prefix, "", 1) + + return "" + + # ------------------------------------------------------------------ + # Rollback registration + # ------------------------------------------------------------------ + + def _register_rollback( + self, + priv_pod_name: str, + maj_min: str, + cgroup_path: str, + cgroup_version: str, + namespace: str, + ): + """Register rollback data so throttle can be undone on failure.""" + rollback_data = { + "priv_pod_name": priv_pod_name, + "maj_min": maj_min, + "cgroup_path": cgroup_path, + "cgroup_version": cgroup_version, + } + encoded_data = base64.b64encode( + json.dumps(rollback_data).encode("utf-8") + ).decode("utf-8") + self.rollback_handler.set_rollback_callable( + self.rollback_throttle, + RollbackContent( + namespace=namespace, + resource_identifier=encoded_data, + ), + ) + + # ------------------------------------------------------------------ + # Duration hold + # ------------------------------------------------------------------ + + @staticmethod + def _wait_with_progress(duration: int, interval: int = 30): + """Sleep for *duration* seconds, logging progress every *interval*.""" + elapsed = 0 + while elapsed < duration: + chunk = min(interval, duration - elapsed) + time.sleep(chunk) + elapsed += chunk + logging.info( + "Throttle active: %d/%ds elapsed" % (elapsed, duration) + ) + + # ------------------------------------------------------------------ + # Throttle apply / remove + # ------------------------------------------------------------------ + + def _apply_throttle( + self, + lib_k8s: KrknKubernetes, + priv_pod_name: str, + cgroup_path: str, + cgroup_version: str, + maj_min: str, + params: ThrottleParams, + namespace: str, + ): + """Apply I/O throttle via cgroup writes.""" + if cgroup_version == "v2": + self._apply_throttle_v2( + lib_k8s, priv_pod_name, cgroup_path, maj_min, + params, namespace, + ) + else: + self._apply_throttle_v1( + lib_k8s, priv_pod_name, cgroup_path, maj_min, + params, namespace, + ) + + def _apply_throttle_v2( + self, lib_k8s, priv_pod_name, cgroup_path, maj_min, + params: ThrottleParams, namespace, + ): + io_max_path = "/sys/fs/cgroup%s/io.max" % cgroup_path + + if params.throttle_type == "iops": + value = "%s riops=%d wiops=%d" % ( + maj_min, params.read_iops, params.write_iops, + ) + elif params.throttle_type == "bandwidth": + value = "%s rbps=%d wbps=%d" % ( + maj_min, params.read_bps, params.write_bps, + ) + else: # both + value = "%s rbps=%d wbps=%d riops=%d wiops=%d" % ( + maj_min, params.read_bps, params.write_bps, + params.read_iops, params.write_iops, + ) + + logging.info("Setting io.max: %s" % value) + self._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s' > %s" % (value, io_max_path), + namespace, + ) + + result = self._chroot_exec( + lib_k8s, priv_pod_name, "cat %s" % io_max_path, namespace + ) + logging.info("Verified io.max: %s" % result) + if result and maj_min not in result: + logging.warning( + "Throttle may not have been applied; io.max readback: %s" + % result + ) + + def _apply_throttle_v1( + self, lib_k8s, priv_pod_name, cgroup_path, maj_min, + params: ThrottleParams, namespace, + ): + blkio_path = "/sys/fs/cgroup/blkio%s" % cgroup_path + + if params.throttle_type in ("iops", "both"): + self._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s %d' > %s/blkio.throttle.read_iops_device" + % (maj_min, params.read_iops, blkio_path), + namespace, + ) + self._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s %d' > %s/blkio.throttle.write_iops_device" + % (maj_min, params.write_iops, blkio_path), + namespace, + ) + + if params.throttle_type in ("bandwidth", "both"): + self._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s %d' > %s/blkio.throttle.read_bps_device" + % (maj_min, params.read_bps, blkio_path), + namespace, + ) + self._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s %d' > %s/blkio.throttle.write_bps_device" + % (maj_min, params.write_bps, blkio_path), + namespace, + ) + + logging.info("Verified blkio settings:") + for f in self._V1_BLKIO_FILES: + result = self._chroot_exec( + lib_k8s, priv_pod_name, + "cat %s/%s 2>/dev/null" % (blkio_path, f), + namespace, + ) + if result: + logging.info(" %s: %s" % (f, result.strip())) + if maj_min not in result: + logging.warning( + "Throttle may not be active for %s; readback: %s" + % (f, result.strip()) + ) + + @staticmethod + def _remove_throttle( + lib_k8s: KrknKubernetes, + priv_pod_name: str, + cgroup_path: str, + cgroup_version: str, + maj_min: str, + namespace: str, + ): + """Remove the I/O throttle by resetting cgroup values.""" + if cgroup_version == "v2": + io_max_path = "/sys/fs/cgroup%s/io.max" % cgroup_path + reset_value = "%s rbps=max wbps=max riops=max wiops=max" % maj_min + StorageThrottleScenarioPlugin._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s' > %s" % (reset_value, io_max_path), + namespace, + ) + else: + blkio_path = "/sys/fs/cgroup/blkio%s" % cgroup_path + for f in StorageThrottleScenarioPlugin._V1_BLKIO_FILES: + StorageThrottleScenarioPlugin._chroot_exec( + lib_k8s, priv_pod_name, + "echo '%s 0' > %s/%s" % (maj_min, blkio_path, f), + namespace, + ) + + @staticmethod + def _chroot_exec( + lib_k8s: KrknKubernetes, priv_pod_name: str, cmd: str, + namespace: str, + ) -> str: + """Execute a command inside the privileged pod via chroot /host.""" + return lib_k8s.exec_cmd_in_pod( + ["/host", "bash", "-c", cmd], + priv_pod_name, + namespace, + base_command="chroot", + ) + + @staticmethod + def rollback_throttle( + rollback_content: RollbackContent, + lib_telemetry: KrknTelemetryOpenshift, + ): + """ + Rollback: remove any applied throttle and delete the privileged pod. + + :param rollback_content: Contains namespace and encoded rollback data. + :param lib_telemetry: KrknTelemetryOpenshift instance. + """ + try: + namespace = rollback_content.namespace + decoded = base64.b64decode( + rollback_content.resource_identifier.encode("utf-8") + ).decode("utf-8") + data = json.loads(decoded) + priv_pod_name = data["priv_pod_name"] + maj_min = data["maj_min"] + + lib_k8s = lib_telemetry.get_lib_kubernetes() + logging.info( + "Rolling back storage throttle: removing limits and " + "deleting pod %s" % priv_pod_name + ) + + if not _validate_maj_min(maj_min): + logging.warning( + "Invalid maj_min during rollback, skipping throttle removal" + ) + else: + cgroup_path = data.get("cgroup_path") + cgroup_version = data.get("cgroup_version") + if cgroup_path and cgroup_version in ("v1", "v2"): + try: + StorageThrottleScenarioPlugin._remove_throttle( + lib_k8s, + priv_pod_name, + cgroup_path, + cgroup_version, + maj_min, + namespace, + ) + logging.info("Throttle limits removed during rollback") + except Exception as rem_exc: + logging.warning( + "Rollback throttle removal failed: %s" % rem_exc + ) + else: + logging.warning( + "Rollback data missing cgroup_path or cgroup_version, " + "cannot remove throttle" + ) + + lib_k8s.delete_pod(priv_pod_name, namespace) + logging.info("Privileged pod deleted during rollback") + + except Exception as e: + logging.error("Failed to rollback storage throttle: %s" % e) diff --git a/krkn/scenario_plugins/storage_throttle/storage_throttle_utils.py b/krkn/scenario_plugins/storage_throttle/storage_throttle_utils.py new file mode 100644 index 00000000..cc0d1b0e --- /dev/null +++ b/krkn/scenario_plugins/storage_throttle/storage_throttle_utils.py @@ -0,0 +1,147 @@ +# Copyright 2026 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. + +""" +Parsing and validation utilities for the storage throttle scenario plugin. + +Kubernetes-style binary byte suffixes (powers of 1024): + Ki = Kibibyte = 1024 bytes + Mi = Mebibyte = 1024^2 = 1,048,576 bytes + Gi = Gibibyte = 1024^3 = 1,073,741,824 bytes + +SI decimal suffixes (powers of 1000): + K = Kilobyte = 1,000 bytes + M = Megabyte = 1,000,000 bytes + G = Gigabyte = 1,000,000,000 bytes + +Duration suffixes: + s = seconds + m = minutes (x60) + h = hours (x3600) +""" + +import re + +_BYTE_UNITS = { + "Ki": 1024, + "Mi": 1024 ** 2, + "Gi": 1024 ** 3, + "K": 1000, + "M": 1000 ** 2, + "G": 1000 ** 3, +} + +_DURATION_UNITS = { + "s": 1, + "m": 60, + "h": 3600, +} + +_BYTE_PATTERN = re.compile( + r"^\s*(\d+(?:\.\d+)?)\s*(Ki|Mi|Gi|K|M|G)?\s*$" +) +_DURATION_PATTERN = re.compile( + r"^\s*(\d+(?:\.\d+)?)\s*(s|m|h)?\s*$" +) + +_SAFE_PATH_RE = re.compile(r"^/[a-zA-Z0-9._/\-]+$") +_CGROUP_PATH_RE = re.compile(r"^/[a-zA-Z0-9._/\-]+$") +_MAJ_MIN_RE = re.compile(r"^\d+:\d+$") +_CONTAINER_ID_RE = re.compile(r"^[a-f0-9]+$") + + +def validate_mount_path(path: str) -> bool: + """Validate mount path contains only safe characters and no traversal.""" + if ".." in path: + return False + return bool(_SAFE_PATH_RE.match(path)) + + +def validate_cgroup_path(path: str) -> bool: + """Validate cgroup path contains only safe characters and no traversal.""" + if ".." in path: + return False + return bool(_CGROUP_PATH_RE.match(path)) + + +def validate_maj_min(value: str) -> bool: + """Validate device major:minor format (e.g. '8:16').""" + return bool(_MAJ_MIN_RE.match(value)) + + +def validate_container_id(value: str) -> bool: + """Validate container ID is a hex string (CRI-O/containerd format).""" + return bool(value) and bool(_CONTAINER_ID_RE.match(value)) + + +def parse_byte_value(value) -> int: + """Parse a byte value that may use Kubernetes-style unit suffixes. + + Accepts: + - Plain integers: 1048576 -> 1048576 + - String with suffix: "1Mi" -> 1048576, "512Ki" -> 524288, "5Gi" -> 5368709120 + - String without suffix: "1048576" -> 1048576 + + Supported suffixes (binary, Kubernetes-style): + Ki = 1024, Mi = 1024^2 (1,048,576), Gi = 1024^3 (1,073,741,824) + Supported suffixes (decimal, SI): + K = 1000, M = 1000^2, G = 1000^3 + """ + if isinstance(value, (int, float)): + return int(value) + if not isinstance(value, str): + raise ValueError("byte value must be an int or string, got: %r" % value) + + match = _BYTE_PATTERN.match(value) + if not match: + raise ValueError( + "invalid byte value '%s'. Use a number optionally followed by " + "Ki, Mi, Gi (binary) or K, M, G (decimal). " + "Examples: 1048576, '1Mi', '512Ki', '5Gi'" % value + ) + number = float(match.group(1)) + suffix = match.group(2) + if suffix: + return int(number * _BYTE_UNITS[suffix]) + return int(number) + + +def parse_duration_value(value) -> int: + """Parse a duration value into seconds. + + Accepts: + - Plain integers: 120 -> 120 (seconds) + - String with suffix: "2m" -> 120, "30s" -> 30, "1h" -> 3600 + - String without suffix: "120" -> 120 (seconds) + + Supported suffixes: + s = seconds, m = minutes (x60), h = hours (x3600) + """ + if isinstance(value, (int, float)): + return int(value) + if not isinstance(value, str): + raise ValueError("duration must be an int or string, got: %r" % value) + + match = _DURATION_PATTERN.match(value) + if not match: + raise ValueError( + "invalid duration '%s'. Use a number optionally followed by " + "s (seconds), m (minutes), or h (hours). " + "Examples: 120, '2m', '30s', '1h'" % value + ) + number = float(match.group(1)) + suffix = match.group(2) + if suffix: + return int(number * _DURATION_UNITS[suffix]) + return int(number) diff --git a/requirements.txt b/requirements.txt index 20f8eaac..6c3ba263 100644 --- a/requirements.txt +++ b/requirements.txt @@ -17,7 +17,7 @@ ibm_vpc==0.26.3 # Requires ibm_cloud_sdk_core jinja2==3.1.6 lxml==6.1.0 kubernetes>=35.0.0 -krkn-lib==6.0.9 +krkn-lib==6.0.10 numpy==1.26.4 pandas==2.2.0 openshift-client==1.0.21 diff --git a/scenarios/kind/storage_throttle.yaml b/scenarios/kind/storage_throttle.yaml new file mode 100644 index 00000000..8596b5c5 --- /dev/null +++ b/scenarios/kind/storage_throttle.yaml @@ -0,0 +1,12 @@ +storage_throttle_scenario: + pvc_name: throttle-test-pvc # Target PVC name + pod_name: "" # Auto-resolved from PVC + namespace: nginx-test # Namespace of the target PVC/pod + mount_path: /data # Mount path to throttle + throttle_type: bandwidth # "iops", "bandwidth", or "both" + read_iops: 100 # Max read IOPS + write_iops: 50 # Max write IOPS + read_bps: 1048576 # Max read bytes/sec (1 MB/s) + write_bps: 524288 # Max write bytes/sec (512 KB/s) + duration: 30 # How long to hold the throttle in seconds + image: quay.io/krkn-chaos/krkn:tools diff --git a/scenarios/kube/storage_throttle.yaml b/scenarios/kube/storage_throttle.yaml new file mode 100644 index 00000000..a571252a --- /dev/null +++ b/scenarios/kube/storage_throttle.yaml @@ -0,0 +1,12 @@ +storage_throttle_scenario: + pvc_name: "" # Target PVC name. If set, pod_name is auto-resolved from PVC. + pod_name: my-app-pod # Target pod name. Ignored if pvc_name is set. + namespace: default # Namespace of the target PVC/pod + mount_path: "" # Specific mount path to throttle. If empty, first PVC mount is used. + throttle_type: bandwidth # "iops", "bandwidth", or "both" + read_iops: 100 # Max read IOPS (used when throttle_type is "iops" or "both") + write_iops: 50 # Max write IOPS (used when throttle_type is "iops" or "both") + read_bps: 1Mi # Max read bytes/sec (used when throttle_type is "bandwidth" or "both") + write_bps: 512Ki # Max write bytes/sec (used when throttle_type is "bandwidth" or "both") + duration: 1m # How long to hold the throttle (supports suffixes: 30s, 2m, 1h) + # image: quay.io/krkn-chaos/krkn:tools # (optional) override helper pod image diff --git a/scenarios/openshift/storage_throttle.yaml b/scenarios/openshift/storage_throttle.yaml new file mode 100644 index 00000000..e8cf9760 --- /dev/null +++ b/scenarios/openshift/storage_throttle.yaml @@ -0,0 +1,12 @@ +storage_throttle_scenario: + pvc_name: # Target PVC name. If set, pod_name is auto-resolved from PVC. + pod_name: # Target pod name. Ignored if pvc_name is set. + namespace: # Namespace of the target PVC/pod + mount_path: "" # Specific mount path to throttle. If empty, first PVC mount is used. + throttle_type: bandwidth # "iops", "bandwidth", or "both" + read_iops: 100 # Max read IOPS (used when throttle_type is "iops" or "both") + write_iops: 50 # Max write IOPS (used when throttle_type is "iops" or "both") + read_bps: 1048576 # Max read bytes/sec, 1 MB/s (used when throttle_type is "bandwidth" or "both") + write_bps: 524288 # Max write bytes/sec, 512 KB/s (used when throttle_type is "bandwidth" or "both") + duration: 120 # How long to hold the throttle in seconds + image: quay.io/krkn-chaos/krkn:tools # Image for the privileged helper pod diff --git a/tests/test_storage_throttle_scenario_plugin.py b/tests/test_storage_throttle_scenario_plugin.py new file mode 100644 index 00000000..c8d7590e --- /dev/null +++ b/tests/test_storage_throttle_scenario_plugin.py @@ -0,0 +1,1159 @@ +#!/usr/bin/env python3 + +""" +Test suite for StorageThrottleScenarioPlugin class + +Usage: + python -m coverage run -a -m unittest tests/test_storage_throttle_scenario_plugin.py -v +""" + +import base64 +import json +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch, call + +import yaml +from krkn_lib.k8s import KrknKubernetes +from krkn_lib.telemetry.ocp import KrknTelemetryOpenshift + +from krkn.scenario_plugins.storage_throttle.storage_throttle_scenario_plugin import ( + StorageThrottleScenarioPlugin, + ThrottleParams, +) +from krkn.scenario_plugins.storage_throttle.storage_throttle_utils import ( + validate_cgroup_path as _validate_cgroup_path, + validate_container_id as _validate_container_id, + validate_maj_min as _validate_maj_min, + validate_mount_path as _validate_mount_path, + parse_byte_value as _parse_byte_value, + parse_duration_value as _parse_duration_value, +) +from krkn.rollback.config import RollbackContent + + +class TestValidateShellInputs(unittest.TestCase): + """Sanity checks for mount path and major:minor validation.""" + + def test_validate_mount_path_ok(self): + self.assertTrue(_validate_mount_path("/data")) + self.assertTrue(_validate_mount_path("/var/lib/kubelet/pods/abc/volumes")) + + def test_validate_mount_path_rejects_shell_chars(self): + self.assertFalse(_validate_mount_path("/data;rm -rf /")) + self.assertFalse(_validate_mount_path("$(whoami)")) + self.assertFalse(_validate_mount_path("")) + + def test_validate_mount_path_rejects_traversal(self): + self.assertFalse(_validate_mount_path("/data/../etc/shadow")) + self.assertFalse(_validate_mount_path("/..")) + + def test_validate_cgroup_path_ok(self): + self.assertTrue(_validate_cgroup_path("/kubepods.slice/crio-abc.scope")) + self.assertTrue(_validate_cgroup_path("/kubepods/abc123def456")) + + def test_validate_cgroup_path_rejects_bad(self): + self.assertFalse(_validate_cgroup_path("")) + self.assertFalse(_validate_cgroup_path("/kubepods/../etc")) + self.assertFalse(_validate_cgroup_path("relative/path")) + + def test_validate_container_id_ok(self): + self.assertTrue(_validate_container_id("abc123def456")) + self.assertTrue(_validate_container_id("0123456789abcdef")) + + def test_validate_container_id_rejects_bad(self): + self.assertFalse(_validate_container_id("")) + self.assertFalse(_validate_container_id("abc;rm -rf /")) + self.assertFalse(_validate_container_id("ABC123")) # uppercase + self.assertFalse(_validate_container_id("abc 123")) + + def test_validate_maj_min_ok(self): + self.assertTrue(_validate_maj_min("8:16")) + self.assertTrue(_validate_maj_min("259:0")) + + def test_validate_maj_min_rejects_bad(self): + self.assertFalse(_validate_maj_min("8:16 foo")) + self.assertFalse(_validate_maj_min("abc")) + + +class TestParseByteValue(unittest.TestCase): + """Tests for _parse_byte_value — Kubernetes-style unit parsing. + + Supported binary suffixes: Ki (1024), Mi (1024^2), Gi (1024^3) + Supported decimal suffixes: K (1000), M (1000^2), G (1000^3) + """ + + def test_plain_int(self): + self.assertEqual(_parse_byte_value(1048576), 1048576) + + def test_plain_int_zero(self): + self.assertEqual(_parse_byte_value(0), 0) + + def test_string_no_suffix(self): + self.assertEqual(_parse_byte_value("1048576"), 1048576) + + def test_ki_suffix(self): + self.assertEqual(_parse_byte_value("512Ki"), 524288) + + def test_mi_suffix(self): + self.assertEqual(_parse_byte_value("1Mi"), 1048576) + + def test_gi_suffix(self): + self.assertEqual(_parse_byte_value("1Gi"), 1073741824) + + def test_decimal_k_suffix(self): + self.assertEqual(_parse_byte_value("500K"), 500000) + + def test_decimal_m_suffix(self): + self.assertEqual(_parse_byte_value("5M"), 5000000) + + def test_decimal_g_suffix(self): + self.assertEqual(_parse_byte_value("1G"), 1000000000) + + def test_fractional_value(self): + self.assertEqual(_parse_byte_value("1.5Mi"), 1572864) + + def test_whitespace_handling(self): + self.assertEqual(_parse_byte_value(" 1Mi "), 1048576) + + def test_invalid_suffix(self): + with self.assertRaises(ValueError): + _parse_byte_value("100Ti") + + def test_invalid_type(self): + with self.assertRaises(ValueError): + _parse_byte_value([100]) + + def test_float_passthrough(self): + self.assertEqual(_parse_byte_value(1048576.7), 1048576) + + +class TestParseDurationValue(unittest.TestCase): + """Tests for _parse_duration_value — time unit parsing. + + Supported suffixes: s (seconds), m (minutes x60), h (hours x3600) + """ + + def test_plain_int(self): + self.assertEqual(_parse_duration_value(120), 120) + + def test_string_no_suffix(self): + self.assertEqual(_parse_duration_value("120"), 120) + + def test_seconds_suffix(self): + self.assertEqual(_parse_duration_value("30s"), 30) + + def test_minutes_suffix(self): + self.assertEqual(_parse_duration_value("2m"), 120) + + def test_hours_suffix(self): + self.assertEqual(_parse_duration_value("1h"), 3600) + + def test_fractional_minutes(self): + self.assertEqual(_parse_duration_value("1.5m"), 90) + + def test_whitespace_handling(self): + self.assertEqual(_parse_duration_value(" 5m "), 300) + + def test_invalid_suffix(self): + with self.assertRaises(ValueError): + _parse_duration_value("10d") + + def test_invalid_type(self): + with self.assertRaises(ValueError): + _parse_duration_value([60]) + + def test_float_passthrough(self): + self.assertEqual(_parse_duration_value(60.9), 60) + + +class TestStorageThrottleScenarioPlugin(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def test_get_scenario_types(self): + result = self.plugin.get_scenario_types() + self.assertEqual(result, ["storage_throttle_scenarios"]) + self.assertEqual(len(result), 1) + + +class TestResolveTargetPod(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def test_resolve_by_pod_name(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + result = self.plugin._resolve_pod_name( + mock_k8s, "", "my-pod", "default" + ) + self.assertEqual(result, "my-pod") + mock_k8s.get_pvc_info.assert_not_called() + + def test_resolve_by_pvc_name(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_pvc = MagicMock() + mock_pvc.podNames = ["pvc-pod-1", "pvc-pod-2"] + mock_k8s.get_pvc_info.return_value = mock_pvc + + result = self.plugin._resolve_pod_name( + mock_k8s, "my-pvc", "", "default" + ) + self.assertIn(result, ["pvc-pod-1", "pvc-pod-2"]) + + def test_resolve_pvc_no_pods(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_pvc = MagicMock() + mock_pvc.podNames = [] + mock_k8s.get_pvc_info.return_value = mock_pvc + + result = self.plugin._resolve_pod_name( + mock_k8s, "my-pvc", "", "default" + ) + self.assertEqual(result, "") + + def test_resolve_pvc_overrides_pod_name(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_pvc = MagicMock() + mock_pvc.podNames = ["from-pvc"] + mock_k8s.get_pvc_info.return_value = mock_pvc + + result = self.plugin._resolve_pod_name( + mock_k8s, "my-pvc", "ignored-pod", "default" + ) + self.assertEqual(result, "from-pvc") + + def test_resolve_pvc_not_found(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.get_pvc_info.return_value = None + + result = self.plugin._resolve_pod_name( + mock_k8s, "missing-pvc", "", "default" + ) + self.assertEqual(result, "") + + +class TestFindPvcMount(unittest.TestCase): + + def test_find_mount_success(self): + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "test-pvc" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_container = MagicMock() + mock_container.name = "app" + mock_vol_mount = MagicMock() + mock_vol_mount.name = "vol1" + mock_vol_mount.mountPath = "/data" + mock_container.volumeMounts = [mock_vol_mount] + mock_pod.containers = [mock_container] + + container, path = StorageThrottleScenarioPlugin._find_pvc_mount( + mock_pod, "" + ) + self.assertEqual(container, "app") + self.assertEqual(path, "/data") + + def test_find_mount_specific_path(self): + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "test-pvc" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_container = MagicMock() + mock_container.name = "app" + mock_mount1 = MagicMock() + mock_mount1.name = "vol1" + mock_mount1.mountPath = "/data" + mock_mount2 = MagicMock() + mock_mount2.name = "other-vol" + mock_mount2.mountPath = "/logs" + mock_container.volumeMounts = [mock_mount2, mock_mount1] + mock_pod.containers = [mock_container] + + container, path = StorageThrottleScenarioPlugin._find_pvc_mount( + mock_pod, "/data" + ) + self.assertEqual(container, "app") + self.assertEqual(path, "/data") + + def test_find_mount_no_pvc(self): + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = None + mock_pod.volumes = [mock_volume] + + container, path = StorageThrottleScenarioPlugin._find_pvc_mount( + mock_pod, "" + ) + self.assertIsNone(container) + self.assertIsNone(path) + + def test_find_mount_filters_by_pvc_name(self): + """With multiple PVCs, only the requested pvc_name is selected.""" + mock_pod = MagicMock() + + vol_logs = MagicMock() + vol_logs.pvcName = "logs-pvc" + vol_logs.name = "logs-vol" + + vol_data = MagicMock() + vol_data.pvcName = "data-pvc" + vol_data.name = "data-vol" + + mock_pod.volumes = [vol_logs, vol_data] + + mount_logs = MagicMock() + mount_logs.name = "logs-vol" + mount_logs.mountPath = "/logs" + + mount_data = MagicMock() + mount_data.name = "data-vol" + mount_data.mountPath = "/data" + + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mount_logs, mount_data] + mock_pod.containers = [mock_container] + + container, path = StorageThrottleScenarioPlugin._find_pvc_mount( + mock_pod, "", "data-pvc" + ) + self.assertEqual(container, "app") + self.assertEqual(path, "/data") + + def test_find_mount_pvc_name_not_found(self): + """Returns None when requested pvc_name doesn't match any volume.""" + mock_pod = MagicMock() + vol = MagicMock() + vol.pvcName = "other-pvc" + vol.name = "vol1" + mock_pod.volumes = [vol] + + mount = MagicMock() + mount.name = "vol1" + mount.mountPath = "/data" + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mount] + mock_pod.containers = [mock_container] + + container, path = StorageThrottleScenarioPlugin._find_pvc_mount( + mock_pod, "", "missing-pvc" + ) + self.assertIsNone(container) + self.assertIsNone(path) + + +class TestGetDeviceMajMin(unittest.TestCase): + + def test_success(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "8:16\n" + + result = StorageThrottleScenarioPlugin._get_device_maj_min( + mock_k8s, "pod", "ns", "container", "/data" + ) + self.assertEqual(result, "8:16") + + def test_empty_output(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + result = StorageThrottleScenarioPlugin._get_device_maj_min( + mock_k8s, "pod", "ns", "container", "/data" + ) + self.assertEqual(result, "") + + def test_none_output(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = None + + result = StorageThrottleScenarioPlugin._get_device_maj_min( + mock_k8s, "pod", "ns", "container", "/data" + ) + self.assertEqual(result, "") + + def test_invalid_mount_path_rejected(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + result = StorageThrottleScenarioPlugin._get_device_maj_min( + mock_k8s, "pod", "ns", "container", "/data/../etc" + ) + self.assertEqual(result, "") + mock_k8s.exec_cmd_in_pod.assert_not_called() + + +class TestGetContainerId(unittest.TestCase): + + def test_crio_container_id(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "app" + mock_c.containerId = "cri-o://abc123def456" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "abc123def456") + + def test_containerd_container_id(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "app" + mock_c.containerId = "containerd://abc123def456" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "abc123def456") + + def test_container_not_found(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "other" + mock_c.containerId = "cri-o://abc123" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "") + + def test_empty_container_id(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "app" + mock_c.containerId = "" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "") + + def test_pod_not_found(self): + result = StorageThrottleScenarioPlugin._get_container_id(None, "app") + self.assertEqual(result, "") + + def test_non_hex_container_id_rejected(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "app" + mock_c.containerId = "cri-o://UPPERCASE_NOT_HEX" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "") + + def test_shell_injection_container_id_rejected(self): + mock_pod = MagicMock() + mock_c = MagicMock() + mock_c.name = "app" + mock_c.containerId = "cri-o://abc;rm -rf /" + mock_pod.containers = [mock_c] + + result = StorageThrottleScenarioPlugin._get_container_id(mock_pod, "app") + self.assertEqual(result, "") + + +class TestDetectCgroupVersion(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def test_cgroup_v2(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "cgroup2fs\n" + + result = self.plugin._detect_cgroup_version(mock_k8s, "priv-pod", "default") + self.assertEqual(result, "v2") + + def test_cgroup_v1(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "tmpfs\n" + + result = self.plugin._detect_cgroup_version(mock_k8s, "priv-pod", "default") + self.assertEqual(result, "v1") + + def test_cgroup_unknown(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = None + + result = self.plugin._detect_cgroup_version(mock_k8s, "priv-pod", "default") + self.assertEqual(result, "v1") + + +class TestFindHostCgroupPath(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def test_cgroup_v2_scope_found(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = ( + "/sys/fs/cgroup/kubepods.slice/crio-abc123def456.scope\n" + ) + + result = self.plugin._find_host_cgroup_path( + mock_k8s, "priv-pod", "abc123def456789", "v2", "default" + ) + self.assertEqual(result, "/kubepods.slice/crio-abc123def456.scope") + + def test_cgroup_v2_dir_fallback(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.side_effect = [ + "", # scope search returns nothing + "/sys/fs/cgroup/kubepods/abc123def456\n", # dir search + ] + + result = self.plugin._find_host_cgroup_path( + mock_k8s, "priv-pod", "abc123def456789", "v2", "default" + ) + self.assertEqual(result, "/kubepods/abc123def456") + + def test_cgroup_v1(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = ( + "/sys/fs/cgroup/blkio/kubepods/abc123def456.scope\n" + ) + + result = self.plugin._find_host_cgroup_path( + mock_k8s, "priv-pod", "abc123def456789", "v1", "default" + ) + self.assertEqual(result, "/kubepods/abc123def456.scope") + + def test_not_found(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + result = self.plugin._find_host_cgroup_path( + mock_k8s, "priv-pod", "abc123def456789", "v2", "default" + ) + self.assertEqual(result, "") + + +class TestApplyRemoveThrottle(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def _make_params(self, **overrides): + defaults = dict( + pvc_name="", pod_name="app-pod", namespace="default", + throttle_type="bandwidth", read_iops=100, write_iops=50, + read_bps=1048576, write_bps=524288, duration=60, + mount_path="", image="quay.io/krkn-chaos/krkn:tools", + ) + defaults.update(overrides) + return ThrottleParams(**defaults) + + def test_apply_throttle_v2_bandwidth(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "8:16 rbps=1048576 wbps=524288" + + self.plugin._apply_throttle( + mock_k8s, "priv-pod", "/kubepods/crio-abc.scope", "v2", + "8:16", self._make_params(throttle_type="bandwidth"), "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + self.assertEqual(len(calls), 2) # echo + cat + echo_cmd = calls[0][0][0][3] + self.assertIn("rbps=1048576", echo_cmd) + self.assertIn("wbps=524288", echo_cmd) + + def test_apply_throttle_v2_iops(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "8:16 riops=100 wiops=50" + + self.plugin._apply_throttle( + mock_k8s, "priv-pod", "/kubepods/crio-abc.scope", "v2", + "8:16", self._make_params(throttle_type="iops"), "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + echo_cmd = calls[0][0][0][3] + self.assertIn("riops=100", echo_cmd) + self.assertIn("wiops=50", echo_cmd) + + def test_apply_throttle_v2_both(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + self.plugin._apply_throttle( + mock_k8s, "priv-pod", "/kubepods/crio-abc.scope", "v2", + "8:16", self._make_params(throttle_type="both"), "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + echo_cmd = calls[0][0][0][3] + self.assertIn("rbps=1048576", echo_cmd) + self.assertIn("wiops=50", echo_cmd) + + def test_apply_throttle_v1_bandwidth(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + self.plugin._apply_throttle( + mock_k8s, "priv-pod", "/kubepods/abc", "v1", + "8:16", self._make_params(throttle_type="bandwidth"), "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + # 2 writes (read_bps, write_bps) + 4 cat verifications + self.assertEqual(len(calls), 6) + + def test_remove_throttle_v2(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + self.plugin._remove_throttle( + mock_k8s, "priv-pod", "/kubepods/crio-abc.scope", "v2", "8:16", + "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + self.assertEqual(len(calls), 1) + echo_cmd = calls[0][0][0][3] + self.assertIn("rbps=max", echo_cmd) + self.assertIn("wbps=max", echo_cmd) + + def test_remove_throttle_v1(self): + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_k8s.exec_cmd_in_pod.return_value = "" + + self.plugin._remove_throttle( + mock_k8s, "priv-pod", "/kubepods/abc", "v1", "8:16", "default", + ) + + calls = mock_k8s.exec_cmd_in_pod.call_args_list + # 4 writes (one for each blkio file) + self.assertEqual(len(calls), 4) + + +class TestRunScenario(unittest.TestCase): + + def setUp(self): + self.plugin = StorageThrottleScenarioPlugin() + + def tearDown(self): + self.plugin = None + + def create_scenario_file(self, config: dict, temp_dir: str) -> str: + path = os.path.join(temp_dir, "scenario.yaml") + with open(path, "w") as f: + yaml.dump(config, f) + return path + + def test_run_missing_namespace(self): + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "pvc_name": "test-pvc", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_missing_pvc_and_pod(self): + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_invalid_throttle_type(self): + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "throttle_type": "invalid", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_pod_not_found(self): + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "nonexistent", + "throttle_type": "bandwidth", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock() + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.get_pod_info.return_value = None + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_scenario_file_not_found(self): + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario="/nonexistent/path.yaml", + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_invalid_config_mount_path(self): + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "mount_path": "/bad path", + "throttle_type": "bandwidth", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_telemetry.get_lib_kubernetes.assert_not_called() + + def test_run_zero_duration_rejected(self): + """run() returns 1 when duration is zero.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "throttle_type": "bandwidth", + "duration": 0, + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_telemetry.get_lib_kubernetes.assert_not_called() + + def test_run_negative_iops_rejected(self): + """run() returns 1 when an IOPS value is negative.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "throttle_type": "iops", + "read_iops": -10, + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_telemetry.get_lib_kubernetes.assert_not_called() + + def test_run_invalid_byte_suffix_rejected(self): + """run() returns 1 with clean error when byte value has invalid suffix.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "throttle_type": "bandwidth", + "read_bps": "10Ti", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_telemetry.get_lib_kubernetes.assert_not_called() + + def test_run_invalid_duration_suffix_rejected(self): + """run() returns 1 with clean error when duration has invalid suffix.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "test-pod", + "throttle_type": "bandwidth", + "duration": "10d", + } + } + path = self.create_scenario_file(config, temp_dir) + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_telemetry.get_lib_kubernetes.assert_not_called() + + def test_run_invalid_maj_min_from_mountinfo(self): + """run() returns 1 when /proc/self/mountinfo returns invalid major:minor.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "app-pod", + "throttle_type": "bandwidth", + } + } + path = self.create_scenario_file(config, temp_dir) + + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "pvc1" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_vol_mount = MagicMock() + mock_vol_mount.name = "vol1" + mock_vol_mount.mountPath = "/data" + + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mock_vol_mount] + mock_pod.containers = [mock_container] + + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.get_pod_info.return_value = mock_pod + + mock_k8s.exec_cmd_in_pod.return_value = "BADVALUE\n" + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + + def test_run_cgroup_path_not_found(self): + """run() returns 1 and cleans up when cgroup path discovery fails.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "app-pod", + "throttle_type": "bandwidth", + } + } + path = self.create_scenario_file(config, temp_dir) + + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "pvc1" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_vol_mount = MagicMock() + mock_vol_mount.name = "vol1" + mock_vol_mount.mountPath = "/data" + + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mock_vol_mount] + mock_container.containerId = "cri-o://abc123def4567890123456789012" + mock_pod.containers = [mock_container] + mock_pod.nodeName = "worker-1" + + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.get_pod_info.return_value = mock_pod + + mock_k8s.deploy_io_throttle_pod.return_value = "io-throttle-12345" + mock_k8s.exec_cmd_in_pod.side_effect = [ + "8:16\n", # _get_device_maj_min + "cgroup2fs\n", # _detect_cgroup_version + "", # _find_host_cgroup_path scope search + "", # _find_host_cgroup_path dir fallback + ] + mock_st = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + self.assertEqual(result, 1) + mock_k8s.deploy_io_throttle_pod.assert_called_once() + mock_k8s.delete_pod.assert_called_once() + + @patch( + "krkn.scenario_plugins.storage_throttle." + "storage_throttle_scenario_plugin.time.sleep" + ) + def test_run_happy_path(self, mock_sleep): + """Full success path through run() with mocks.""" + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "app-pod", + "throttle_type": "bandwidth", + "duration": 45, + } + } + path = self.create_scenario_file(config, temp_dir) + + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "pvc1" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_vol_mount = MagicMock() + mock_vol_mount.name = "vol1" + mock_vol_mount.mountPath = "/data" + + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mock_vol_mount] + mock_container.containerId = "cri-o://abc123def4567890123456789012" + mock_pod.containers = [mock_container] + mock_pod.nodeName = "worker-1" + + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.get_pod_info.return_value = mock_pod + + mock_k8s.deploy_io_throttle_pod.return_value = "io-throttle-12345" + mock_k8s.exec_cmd_in_pod.side_effect = [ + "8:16\n", + "cgroup2fs\n", + "/sys/fs/cgroup/kubepods.slice/crio-abc.scope\n", + "", + "8:16 rbps=1048576 wbps=524288\n", + "", + ] + mock_st = MagicMock() + + # Avoid RollbackHandler persisting to disk during unit tests + self.plugin.rollback_handler.set_rollback_callable = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + + self.assertEqual(result, 0) + mock_k8s.deploy_io_throttle_pod.assert_called_once() + mock_k8s.create_pod.assert_not_called() + mock_k8s.delete_pod.assert_called_once() + self.assertEqual(mock_k8s.exec_cmd_in_pod.call_count, 6) + self.assertEqual(mock_sleep.call_count, 2) + + + @patch( + "krkn.scenario_plugins.storage_throttle." + "storage_throttle_scenario_plugin.time.sleep" + ) + def test_run_removes_throttle_on_wait_failure(self, mock_sleep): + """Throttle is removed even when _wait_with_progress raises.""" + mock_sleep.side_effect = RuntimeError("simulated failure") + + with tempfile.TemporaryDirectory() as temp_dir: + config = { + "storage_throttle_scenario": { + "namespace": "default", + "pod_name": "app-pod", + "throttle_type": "bandwidth", + "duration": 60, + } + } + path = self.create_scenario_file(config, temp_dir) + + mock_pod = MagicMock() + mock_volume = MagicMock() + mock_volume.pvcName = "pvc1" + mock_volume.name = "vol1" + mock_pod.volumes = [mock_volume] + + mock_vol_mount = MagicMock() + mock_vol_mount.name = "vol1" + mock_vol_mount.mountPath = "/data" + + mock_container = MagicMock() + mock_container.name = "app" + mock_container.volumeMounts = [mock_vol_mount] + mock_container.containerId = "cri-o://abc123def4567890123456789012" + mock_pod.containers = [mock_container] + mock_pod.nodeName = "worker-1" + + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock(spec=KrknKubernetes) + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.get_pod_info.return_value = mock_pod + + mock_k8s.deploy_io_throttle_pod.return_value = "io-throttle-12345" + mock_k8s.exec_cmd_in_pod.side_effect = [ + "8:16\n", # _get_device_maj_min + "cgroup2fs\n", # _detect_cgroup_version + "/sys/fs/cgroup/kubepods.slice/crio-abc.scope\n", # scope + "", # _apply_throttle echo + "8:16 rbps=1048576 wbps=524288\n", # _apply_throttle cat + "", # _remove_throttle (in finally cleanup) + ] + mock_st = MagicMock() + self.plugin.rollback_handler.set_rollback_callable = MagicMock() + + result = self.plugin.run( + run_uuid="uuid", scenario=path, + lib_telemetry=mock_telemetry, scenario_telemetry=mock_st, + ) + + self.assertEqual(result, 1) + mock_k8s.deploy_io_throttle_pod.assert_called_once() + mock_k8s.delete_pod.assert_called_once() + # 6 exec calls: maj_min, cgroup_ver, cgroup_path, apply echo, apply cat, remove in finally + self.assertEqual(mock_k8s.exec_cmd_in_pod.call_count, 6) + + +class TestRollbackThrottle(unittest.TestCase): + + def test_rollback_success_v2_stored_cgroup(self): + """Rollback uses stored cgroup_path (single chroot exec for v2 reset).""" + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock() + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + mock_k8s.exec_cmd_in_pod.return_value = "" + + rollback_data = { + "priv_pod_name": "io-throttle-12345", + "maj_min": "8:16", + "cgroup_path": "/kubepods.slice/crio-abc123.scope", + "cgroup_version": "v2", + } + encoded = base64.b64encode( + json.dumps(rollback_data).encode("utf-8") + ).decode("utf-8") + + content = RollbackContent( + namespace="default", resource_identifier=encoded + ) + + StorageThrottleScenarioPlugin.rollback_throttle( + content, mock_telemetry + ) + + mock_k8s.exec_cmd_in_pod.assert_called_once() + mock_k8s.delete_pod.assert_called_once_with( + "io-throttle-12345", "default" + ) + + def test_rollback_missing_cgroup_data_still_deletes_pod(self): + """Rollback without cgroup_path logs a warning but still deletes the pod.""" + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + mock_k8s = MagicMock() + mock_telemetry.get_lib_kubernetes.return_value = mock_k8s + + rollback_data = { + "priv_pod_name": "io-throttle-12345", + "maj_min": "8:16", + } + encoded = base64.b64encode( + json.dumps(rollback_data).encode("utf-8") + ).decode("utf-8") + + content = RollbackContent( + namespace="default", resource_identifier=encoded + ) + + StorageThrottleScenarioPlugin.rollback_throttle( + content, mock_telemetry + ) + + mock_k8s.exec_cmd_in_pod.assert_not_called() + mock_k8s.delete_pod.assert_called_once_with( + "io-throttle-12345", "default" + ) + + @patch( + "krkn.scenario_plugins.storage_throttle." + "storage_throttle_scenario_plugin.logging" + ) + def test_rollback_invalid_data(self, mock_logging): + mock_telemetry = MagicMock(spec=KrknTelemetryOpenshift) + + content = RollbackContent( + namespace="default", resource_identifier="bad-data!!!" + ) + + StorageThrottleScenarioPlugin.rollback_throttle( + content, mock_telemetry + ) + + mock_logging.error.assert_called_once() + error_msg = mock_logging.error.call_args[0][0] + self.assertIn("Failed to rollback", error_msg) + + +if __name__ == "__main__": + unittest.main()