feat: add label_selector support to kubevirt vm outage and virt health check (#1341)

Allows selecting VMIs by label selector as an alternative to vm_name regex,
making vm_name optional when label_selector is provided.

Signed-off-by: Paige Patton <prubenda@redhat.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

containers/krknctl-input.json

@@ -540,7 +540,7 @@
     "required": "false"
   },
   {
-    "name": "kubevirt-node-node",
+    "name": "kubevirt-node-name",
     "short_description": "KubeVirt node to filter vms on",
     "description": "Only track VMs in KubeVirt on given node name",
     "variable": "KUBE_VIRT_NODE_NAME",
@@ -548,6 +548,15 @@
     "default": "",
     "required": "false"
   },
+  {
+    "name": "kubevirt-label-selector",
+    "short_description": "KubeVirt label selector to filter vms",
+    "description": "Label selector to filter VMs in KubeVirt",
+    "variable": "KUBE_VIRT_LABEL_SELECTOR",
+    "type": "string",
+    "default": "",
+    "required": "false"
+  },
   {
     "name": "krkn-debug",
     "short_description": "Krkn debug mode",
@@ -584,5 +593,4 @@
     "required": "false",
     "mount_path": "/home/krkn/resiliency-file.yaml"
   }
-
 ]

krkn/health_checks/virt_health_check_plugin.py

@@ -21,7 +21,8 @@ Example configuration in config.yaml:
 kubevirt_checks:
     type: virt_health_check
     namespace: "default"
-    name: ".*"                    # VMI name regex pattern
+    name: ".*"                    # optional VMI name regex pattern; matches all if omitted
+    label_selector: ""            # optional label selector (e.g. "app=myvm"); if set, name is not required
     interval: 2                   # Check interval in seconds
     disconnected: false           # Use disconnected SSH access
     only_failures: false          # Only report failures
@@ -141,7 +142,8 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin):
         self.ssh_node = get_yaml_item_value(config, "ssh_node", "")
         self.node_names = get_yaml_item_value(config, "node_names", "")
         self.exit_on_failure = get_yaml_item_value(config, "exit_on_failure", False)
-        vmi_name_match = get_yaml_item_value(config, "name", ".*")
+        vmi_name_match = get_yaml_item_value(config, "name", None) or ".*"
+        label_selector = get_yaml_item_value(config, "label_selector", None) or None
 
         if self.namespace == "":
             logging.info("kubevirt checks config namespace is not defined, skipping them")
@@ -151,7 +153,7 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin):
             self.kube_vm_plugin = KubevirtVmOutageScenarioPlugin()
             self.kube_vm_plugin.init_clients(k8s_client=self.krkn_lib)
             self.vmis_list = self.kube_vm_plugin.k8s_client.get_vmis(
-                vmi_name_match, self.namespace
+                vmi_name_match, self.namespace, label_selector=label_selector
             )
         except Exception as e:
             logging.error(f"Virt Check init exception: {str(e)}")

krkn/scenario_plugins/kubevirt_vm_outage/kubevirt_vm_outage_scenario_plugin.py

@@ -95,23 +95,30 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
         self.vmis_status = VmisStatus()
         try:
             params = config.get("parameters", {})
-            vm_name = params.get("vm_name")
+            vm_name = params.get("vm_name") or None
+            label_selector = params.get("label_selector") or None
             namespace = params.get("namespace", "default")
             timeout = params.get("timeout", 60)
             kill_count = params.get("kill_count", 1)
             disable_auto_restart = params.get("disable_auto_restart", False)
 
-            if not vm_name:
-                logging.error("vm_name parameter is required")
+            if not vm_name and not label_selector:
+                logging.error("Either vm_name or label_selector parameter is required")
                 return self.vmis_status
             self.vmis_status = VmisStatus()
-            self.vmis_list = self.k8s_client.get_vmis(vm_name,namespace)
+            name_regex = vm_name or ".*"
+            self.vmis_list = self.k8s_client.get_vmis(name_regex, namespace, label_selector=label_selector)
+            if not self.vmis_list:
+                target = f"label_selector={label_selector}" if label_selector else f"vm_name={vm_name}"
+                logging.error(f"No VMIs found matching {target} in namespace {namespace}")
+                return self.vmis_status
             for _ in range(kill_count):
-                
+
                 rand_int = random.randint(0, len(self.vmis_list) - 1)
                 vmi = self.vmis_list[rand_int]
-                    
-                logging.info(f"Starting KubeVirt VM outage scenario for VM: {vm_name} in namespace: {namespace}")
+
+                target = f"label_selector={label_selector}" if label_selector else f"vm_name={vm_name}"
+                logging.info(f"Starting KubeVirt VM outage scenario for {target} in namespace: {namespace}")
                 vmi_name = vmi.get("metadata").get("name")
                 vmi_namespace = vmi.get("metadata").get("namespace")
 
@@ -127,12 +134,12 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
 
                 vmi = self.k8s_client.get_vmi(vmi_name, vmi_namespace)
                 if not vmi:
-                    logging.error(f"VMI {vm_name} not found in namespace {namespace}")
+                    logging.error(f"VMI {vmi_name} not found in namespace {vmi_namespace}")
                     self.vmis_status.unrecovered.append(self.affected_vmi)
                     continue
-                    
+
                 self.original_vmi = vmi
-                logging.info(f"Captured initial state of VMI: {vm_name}")
+                logging.info(f"Captured initial state of VMI: {vmi_name}")
                 result = self.delete_vmi(vmi_name, vmi_namespace, disable_auto_restart)
                 if result != 0:
                     self.vmis_status.unrecovered.append(self.affected_vmi)
@@ -149,7 +156,7 @@ class KubevirtVmOutageScenarioPlugin(AbstractScenarioPlugin):
                 )
 
                 self.vmis_status.recovered.append(self.affected_vmi)
-                logging.info(f"Successfully completed KubeVirt VM outage scenario for VM: {vm_name}")
+                logging.info(f"Successfully completed KubeVirt VM outage scenario for VM: {vmi_name}")
             
             return self.vmis_status
             

scenarios/kubevirt/kubevirt-vm-outage.yaml

@@ -2,6 +2,7 @@ scenarios:
   - name: "kubevirt outage test"
     scenario: kubevirt_vm_outage
     parameters:
-      vm_name: <vm-name>  
-      namespace: <namespace>  
+      vm_name: <vm-name>        # optional if label_selector is set
+      label_selector: ""        # optional; if set, vm_name is not required
+      namespace: <namespace>
       timeout: 60

tests/test_kubevirt_vm_outage.py

@@ -644,6 +644,126 @@ class TestKubevirtVmOutageScenarioPlugin(unittest.TestCase):
         self.assertEqual(result, ["kubevirt_vm_outage"])
         self.assertEqual(len(result), 1)
 
+    # ==================== label_selector Tests ====================
+
+    def test_execute_scenario_missing_both_vm_name_and_label_selector(self):
+        """
+        Test execute_scenario returns empty status when neither vm_name nor label_selector is set
+        """
+        config = {"parameters": {"namespace": "default"}}
+
+        result = self.plugin.execute_scenario(config, self.scenario_telemetry)
+
+        self.assertIsInstance(result, VmisStatus)
+        self.assertEqual(len(result.recovered), 0)
+        self.assertEqual(len(result.unrecovered), 0)
+        self.k8s_client.get_vmis.assert_not_called()
+
+    def test_execute_scenario_with_label_selector_only(self):
+        """
+        Test execute_scenario succeeds using label_selector without vm_name
+        """
+        config = {
+            "parameters": {
+                "label_selector": "app=test-vm",
+                "namespace": "default",
+            }
+        }
+
+        self.k8s_client.get_vmis.return_value = [self.mock_vmi]
+        self.k8s_client.get_vms.return_value = [{"metadata": {"name": "test-vm"}}]
+
+        new_vmi = copy.deepcopy(self.mock_vmi)
+        new_vmi["metadata"]["creationTimestamp"] = "2023-01-01T00:05:00Z"
+
+        self.k8s_client.get_vmi.side_effect = [
+            self.mock_vmi,  # validate_environment
+            self.mock_vmi,  # execute_scenario
+            new_vmi,        # delete_vmi - recreated
+            new_vmi,        # wait_for_running
+        ]
+        self.k8s_client.delete_vmi.return_value = None
+
+        result = self.plugin.execute_scenario(config, self.scenario_telemetry)
+
+        self.assertIsInstance(result, VmisStatus)
+        self.assertEqual(len(result.recovered), 1)
+        self.k8s_client.get_vmis.assert_called_once_with(
+            ".*", "default", label_selector="app=test-vm"
+        )
+
+    def test_execute_scenario_with_both_vm_name_and_label_selector(self):
+        """
+        Test execute_scenario passes both vm_name and label_selector to get_vmis
+        """
+        config = {
+            "parameters": {
+                "vm_name": "test-vm",
+                "label_selector": "app=test-vm",
+                "namespace": "default",
+            }
+        }
+
+        self.k8s_client.get_vmis.return_value = [self.mock_vmi]
+        self.k8s_client.get_vms.return_value = [{"metadata": {"name": "test-vm"}}]
+
+        new_vmi = copy.deepcopy(self.mock_vmi)
+        new_vmi["metadata"]["creationTimestamp"] = "2023-01-01T00:05:00Z"
+
+        self.k8s_client.get_vmi.side_effect = [
+            self.mock_vmi,
+            self.mock_vmi,
+            new_vmi,
+            new_vmi,
+        ]
+        self.k8s_client.delete_vmi.return_value = None
+
+        result = self.plugin.execute_scenario(config, self.scenario_telemetry)
+
+        self.assertIsInstance(result, VmisStatus)
+        self.assertEqual(len(result.recovered), 1)
+        self.k8s_client.get_vmis.assert_called_once_with(
+            "test-vm", "default", label_selector="app=test-vm"
+        )
+
+    def test_execute_scenario_empty_vmis_list_returns_early(self):
+        """
+        Test execute_scenario returns empty status without crashing when no VMIs match
+        """
+        config = {
+            "parameters": {
+                "vm_name": "nonexistent-vm",
+                "namespace": "default",
+            }
+        }
+        self.k8s_client.get_vmis.return_value = []
+
+        result = self.plugin.execute_scenario(config, self.scenario_telemetry)
+
+        self.assertIsInstance(result, VmisStatus)
+        self.assertEqual(len(result.recovered), 0)
+        self.assertEqual(len(result.unrecovered), 0)
+        self.k8s_client.delete_vmi.assert_not_called()
+
+    def test_execute_scenario_empty_label_selector_treated_as_none(self):
+        """
+        Test that an empty string label_selector is treated the same as not set,
+        so vm_name is still required
+        """
+        config = {
+            "parameters": {
+                "label_selector": "",
+                "namespace": "default",
+            }
+        }
+
+        result = self.plugin.execute_scenario(config, self.scenario_telemetry)
+
+        self.assertIsInstance(result, VmisStatus)
+        self.assertEqual(len(result.recovered), 0)
+        self.assertEqual(len(result.unrecovered), 0)
+        self.k8s_client.get_vmis.assert_not_called()
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_virt_health_check_plugin.py

@@ -837,5 +837,101 @@ class TestVirtHealthCheckPluginFactory(unittest.TestCase):
         self.assertEqual(plugin3.__class__.__name__, "VirtHealthCheckPlugin")
 
 
+class TestVirtHealthCheckPluginLabelSelector(unittest.TestCase):
+    """Tests for label_selector support in VirtHealthCheckPlugin._initialize_from_config"""
+
+    def setUp(self):
+        self.factory = HealthCheckFactory()
+        if "virt_health_check" not in self.factory.loaded_plugins:
+            self.skipTest("Virt health check plugin not loaded (missing dependencies)")
+        self.mock_kubecli = MagicMock()
+        self.plugin = self.factory.create_plugin(
+            "virt_health_check", iterations=1, krkn_lib=self.mock_kubecli
+        )
+
+    def _mock_vmi(self, name, ip="10.0.0.1", node="worker-1", namespace="default"):
+        return {
+            "metadata": {"name": name, "namespace": namespace},
+            "status": {
+                "nodeName": node,
+                "interfaces": [{"ipAddress": ip}],
+            },
+        }
+
+    @patch("krkn.health_checks.virt_health_check_plugin.KubevirtVmOutageScenarioPlugin")
+    def test_initialize_with_label_selector_only(self, mock_plugin_class):
+        """Test _initialize_from_config filters VMIs by label_selector when name is not set"""
+        mock_plugin = MagicMock()
+        mock_plugin_class.return_value = mock_plugin
+        mock_plugin.k8s_client = self.mock_kubecli
+        self.mock_kubecli.get_vmis.return_value = [self._mock_vmi("labeled-vm")]
+
+        config = {"namespace": "default", "label_selector": "app=myvm", "interval": 2}
+
+        result = self.plugin._initialize_from_config(config)
+
+        self.assertTrue(result)
+        self.assertEqual(len(self.plugin.vm_list), 1)
+        self.mock_kubecli.get_vmis.assert_called_once_with(
+            ".*", "default", label_selector="app=myvm"
+        )
+
+    @patch("krkn.health_checks.virt_health_check_plugin.KubevirtVmOutageScenarioPlugin")
+    def test_initialize_with_name_and_label_selector(self, mock_plugin_class):
+        """Test _initialize_from_config passes both name and label_selector to get_vmis"""
+        mock_plugin = MagicMock()
+        mock_plugin_class.return_value = mock_plugin
+        mock_plugin.k8s_client = self.mock_kubecli
+        self.mock_kubecli.get_vmis.return_value = [self._mock_vmi("test-vm")]
+
+        config = {
+            "namespace": "default",
+            "name": "test-vm",
+            "label_selector": "env=chaos",
+            "interval": 2,
+        }
+
+        result = self.plugin._initialize_from_config(config)
+
+        self.assertTrue(result)
+        self.mock_kubecli.get_vmis.assert_called_once_with(
+            "test-vm", "default", label_selector="env=chaos"
+        )
+
+    @patch("krkn.health_checks.virt_health_check_plugin.KubevirtVmOutageScenarioPlugin")
+    def test_initialize_no_name_no_label_selector_defaults_to_match_all(self, mock_plugin_class):
+        """Test _initialize_from_config falls back to '.*' when neither name nor label_selector is set"""
+        mock_plugin = MagicMock()
+        mock_plugin_class.return_value = mock_plugin
+        mock_plugin.k8s_client = self.mock_kubecli
+        self.mock_kubecli.get_vmis.return_value = []
+
+        config = {"namespace": "default", "interval": 2}
+
+        result = self.plugin._initialize_from_config(config)
+
+        self.assertTrue(result)
+        self.mock_kubecli.get_vmis.assert_called_once_with(
+            ".*", "default", label_selector=None
+        )
+
+    @patch("krkn.health_checks.virt_health_check_plugin.KubevirtVmOutageScenarioPlugin")
+    def test_initialize_empty_label_selector_treated_as_none(self, mock_plugin_class):
+        """Test that an empty string label_selector is treated as not set, falling back to '.*'"""
+        mock_plugin = MagicMock()
+        mock_plugin_class.return_value = mock_plugin
+        mock_plugin.k8s_client = self.mock_kubecli
+        self.mock_kubecli.get_vmis.return_value = []
+
+        config = {"namespace": "default", "label_selector": "", "interval": 2}
+
+        result = self.plugin._initialize_from_config(config)
+
+        self.assertTrue(result)
+        self.mock_kubecli.get_vmis.assert_called_once_with(
+            ".*", "default", label_selector=None
+        )
+
+
 if __name__ == "__main__":
     unittest.main()