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 <darjain@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com> Signed-off-by: ddjain <darjain@redhat.com> * adding need dco auto review (#1329) Signed-off-by: Paige Patton <prubenda@redhat.com> Signed-off-by: ddjain <darjain@redhat.com> * 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 <dahiyavarun2007@gmail.com> Co-authored-by: Paige Patton <64206430+paigerube14@users.noreply.github.com> Signed-off-by: ddjain <darjain@redhat.com> * adding links to issue of completed roadmap items (#1328) Signed-off-by: Paige Patton <prubenda@redhat.com> Signed-off-by: ddjain <darjain@redhat.com> * container scenario template image update (#1342) Signed-off-by: Paige Patton <prubenda@redhat.com> Signed-off-by: ddjain <darjain@redhat.com> --------- Signed-off-by: ddjain <darjain@redhat.com> Signed-off-by: Paige Patton <prubenda@redhat.com> Signed-off-by: v0idheaven <dahiyavarun2007@gmail.com> Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Paige Patton <64206430+paigerube14@users.noreply.github.com> Co-authored-by: Varun Yadav <dahiyavarun2007@gmail.com>
2026-08-25 09:27:36 +00:00 · 2026-05-19 00:24:55 +05:30
parent 3391ff2453
commit eb2efa84ea
13 changed files with 2311 additions and 1 deletions
@@ -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
@@ -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
@@ -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
@@ -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,
+        )
@@ -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:
@@ -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/"
+                                     "<blkio.throttle.*_device>" % (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)
@@ -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)
@@ -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
@@ -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
@@ -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
@@ -0,0 +1,12 @@
+storage_throttle_scenario:
+  pvc_name: <pvc_name>            # Target PVC name. If set, pod_name is auto-resolved from PVC.
+  pod_name: <pod_name>            # Target pod name. Ignored if pvc_name is set.
+  namespace: <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