diff --git a/krkn/health_checks/virt_health_check_plugin.py b/krkn/health_checks/virt_health_check_plugin.py index e984e639..b44354d3 100644 --- a/krkn/health_checks/virt_health_check_plugin.py +++ b/krkn/health_checks/virt_health_check_plugin.py @@ -175,20 +175,7 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): ip_address = interfaces[0].get("ipAddress") namespace = vmi.get("metadata", {}).get("namespace") - # Filter by node names if specified - if len(node_name_list) > 0 and node_name in node_name_list: - self.vm_list.append( - VirtCheck( - { - "vm_name": vmi_name, - "ip_address": ip_address, - "namespace": namespace, - "node_name": node_name, - "new_ip_address": "", - } - ) - ) - elif len(node_name_list) == 0: + if not node_name_list or node_name in node_name_list: self.vm_list.append( VirtCheck( { @@ -271,6 +258,54 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): return False, None, None + def _get_ssh_status(self, vm) -> bool: + """ + Check SSH accessibility for a VM, updating vm.new_ip_address and vm.node_name + in-place if the VM has migrated. + + :param vm: VirtCheck object representing the VM + :return: True if SSH access succeeded, False otherwise + """ + if not self.disconnected: + return self.get_vm_access(vm.vm_name, vm.namespace) + + ip = vm.new_ip_address or vm.ip_address + vm_status, new_ip_address, new_node_name = self.check_disconnected_access( + ip, vm.node_name, vm.vm_name + ) + # Only update tracked addresses on first migration discovery + if not vm.new_ip_address: + if new_ip_address and vm.ip_address != new_ip_address: + vm.new_ip_address = new_ip_address + if new_node_name and vm.node_name != new_node_name: + vm.node_name = new_node_name + return vm_status + + def check_vmi_ready(self, vmi_name: str, namespace: str) -> bool: + """ + Check if a VMI is in Running phase with a Ready=True condition. + + :param vmi_name: VMI name + :param namespace: namespace + :return: True if VMI is ready, False otherwise + """ + try: + vmi = self.krkn_lib.get_vmi(vmi_name, namespace) + if vmi is None: + return False + phase = vmi.get("status", {}).get("phase", "") + if phase != "Running": + logging.debug(f"VMI {vmi_name} phase is '{phase}', not Running") + return False + for cond in vmi.get("status", {}).get("conditions", []): + if cond.get("type") == "Ready" and cond.get("status") == "True": + return True + logging.debug(f"VMI {vmi_name} has no Ready=True condition") + return False + except Exception: + logging.exception(f"Exception checking VMI ready state for {vmi_name}") + return False + def get_vm_access(self, vm_name: str = "", namespace: str = "") -> bool: """ Check VM accessibility using virtctl protocol. @@ -290,6 +325,45 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): return True return False + @staticmethod + def _compute_check_type(ssh_status: bool, vmi_ready: bool) -> str: + """ + Derive the check_type discriminator from individual check results. + + :param ssh_status: result of the SSH access check + :param vmi_ready: result of the VMI readiness check + :return: 'both', 'ssh_access', 'vmi_ready', or 'healthy' + """ + if not ssh_status and not vmi_ready: + return "both" + if not ssh_status: + return "ssh_access" + if not vmi_ready: + return "vmi_ready" + return "healthy" + + def _make_tracker_entry(self, vm, ssh_status: bool, vmi_ready: bool) -> dict: + """ + Build a fresh tracker entry dict for a VM with the current check results. + + :param vm: VirtCheck object representing the VM + :param ssh_status: result of the SSH access check + :param vmi_ready: result of the VMI readiness check + :return: tracker entry dict + """ + return { + "vm_name": vm.vm_name, + "ip_address": vm.ip_address, + "namespace": vm.namespace, + "node_name": vm.node_name, + "ssh_status": ssh_status, + "vmi_ready": vmi_ready, + "status": ssh_status and vmi_ready, + "check_type": self._compute_check_type(ssh_status, vmi_ready), + "start_timestamp": datetime.now(), + "new_ip_address": vm.new_ip_address, + } + def thread_join(self): """Join all worker threads.""" for thread in self.threads: @@ -303,14 +377,10 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): """ if self.batch_size > 0: for i in range(0, len(self.vm_list), self.batch_size): - if i + self.batch_size > len(self.vm_list): - sub_list = self.vm_list[i:] - else: - sub_list = self.vm_list[i : i + self.batch_size] - index = i + sub_list = self.vm_list[i : i + self.batch_size] t = threading.Thread( target=self._run_virt_check_batch, - name=str(index), + name=str(i), args=(sub_list, telemetry_queue), ) self.threads.append(t) @@ -322,111 +392,95 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): """ Run health checks for a batch of VMs (executed in worker thread). + Each VM gets a single combined tracker entry that carries both ssh_status + and vmi_ready. An entry is closed and a new one started whenever either + check changes state. + :param vm_list_batch: list of VMs to check :param virt_check_telemetry_queue: queue for telemetry """ virt_check_telemetry = [] - virt_check_tracker = {} + vm_tracker = {} while True: - # Thread-safe read of current_iterations with self.iteration_lock: current = self.current_iterations if current >= self.iterations or self._stop_event.is_set(): break for vm in vm_list_batch: - start_time = datetime.now() try: - if not self.disconnected: - vm_status = self.get_vm_access(vm.vm_name, vm.namespace) - else: - # Use new IP if available - if vm.new_ip_address: - vm_status, new_ip_address, new_node_name = ( - self.check_disconnected_access( - vm.new_ip_address, vm.node_name, vm.vm_name - ) - ) - else: - vm_status, new_ip_address, new_node_name = ( - self.check_disconnected_access( - vm.ip_address, vm.node_name, vm.vm_name - ) - ) - if new_ip_address and vm.ip_address != new_ip_address: - vm.new_ip_address = new_ip_address - if new_node_name and vm.node_name != new_node_name: - vm.node_name = new_node_name + ssh_status = self._get_ssh_status(vm) except Exception: logging.exception("Exception in get vm status") - vm_status = False + ssh_status = False - if vm.vm_name not in virt_check_tracker: - start_timestamp = datetime.now() - virt_check_tracker[vm.vm_name] = { - "vm_name": vm.vm_name, - "ip_address": vm.ip_address, - "namespace": vm.namespace, - "node_name": vm.node_name, - "status": vm_status, - "start_timestamp": start_timestamp, - "new_ip_address": vm.new_ip_address, - } - else: - if vm_status != virt_check_tracker[vm.vm_name]["status"]: - end_timestamp = datetime.now() - start_timestamp = virt_check_tracker[vm.vm_name][ - "start_timestamp" - ] - duration = (end_timestamp - start_timestamp).total_seconds() - virt_check_tracker[vm.vm_name][ - "end_timestamp" - ] = end_timestamp.isoformat() - virt_check_tracker[vm.vm_name]["duration"] = duration - virt_check_tracker[vm.vm_name][ - "start_timestamp" - ] = start_timestamp.isoformat() - if vm.new_ip_address: - virt_check_tracker[vm.vm_name][ - "new_ip_address" - ] = vm.new_ip_address + vmi_ready = self.check_vmi_ready(vm.vm_name, vm.namespace) - if self.only_failures: - if not virt_check_tracker[vm.vm_name]["status"]: - virt_check_telemetry.append( - VirtCheck(virt_check_tracker[vm.vm_name]) - ) - else: - virt_check_telemetry.append( - VirtCheck(virt_check_tracker[vm.vm_name]) - ) - del virt_check_tracker[vm.vm_name] + if vm.vm_name not in vm_tracker: + vm_tracker[vm.vm_name] = self._make_tracker_entry( + vm, ssh_status, vmi_ready + ) + elif ( + ssh_status != vm_tracker[vm.vm_name]["ssh_status"] + or vmi_ready != vm_tracker[vm.vm_name]["vmi_ready"] + ): + if not vmi_ready and vm_tracker[vm.vm_name]["vmi_ready"]: + logging.warning( + f"VMI {vm.vm_name} in namespace {vm.namespace} transitioned to not-ready" + ) + if vm.new_ip_address: + vm_tracker[vm.vm_name]["new_ip_address"] = vm.new_ip_address + self._close_tracker_entry( + vm_tracker, vm.vm_name, virt_check_telemetry + ) + vm_tracker[vm.vm_name] = self._make_tracker_entry( + vm, ssh_status, vmi_ready + ) time.sleep(self.interval) - # Record final status - virt_check_end_time_stamp = datetime.now() - for vm in virt_check_tracker.keys(): - final_start_timestamp = virt_check_tracker[vm]["start_timestamp"] - final_duration = ( - virt_check_end_time_stamp - final_start_timestamp - ).total_seconds() - virt_check_tracker[vm]["end_timestamp"] = virt_check_end_time_stamp.isoformat() - virt_check_tracker[vm]["duration"] = final_duration - virt_check_tracker[vm]["start_timestamp"] = final_start_timestamp.isoformat() - - if self.only_failures: - if not virt_check_tracker[vm]["status"]: - virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm])) - else: - virt_check_telemetry.append(VirtCheck(virt_check_tracker[vm])) + # Record final status for all open tracker entries + end_timestamp = datetime.now() + for vm_name in vm_tracker: + self._close_tracker_entry( + vm_tracker, vm_name, virt_check_telemetry, end_timestamp, delete=False + ) try: virt_check_telemetry_queue.put(virt_check_telemetry) except Exception as e: logging.error(f"Put queue error: {str(e)}") + def _close_tracker_entry( + self, + tracker: dict, + vm_name: str, + telemetry: list, + end_timestamp: datetime = None, + delete: bool = True, + ) -> None: + """ + Finalize a tracker entry: stamp timestamps/duration, conditionally append + to telemetry, and optionally remove from the tracker. + + :param tracker: the vm_tracker dict + :param vm_name: key into tracker + :param telemetry: list to append VirtCheck to + :param end_timestamp: override end time; defaults to now + :param delete: whether to remove the entry from tracker after closing + """ + if end_timestamp is None: + end_timestamp = datetime.now() + start = tracker[vm_name]["start_timestamp"] + tracker[vm_name]["end_timestamp"] = end_timestamp.isoformat() + tracker[vm_name]["duration"] = (end_timestamp - start).total_seconds() + tracker[vm_name]["start_timestamp"] = start.isoformat() + if not self.only_failures or not tracker[vm_name]["status"]: + telemetry.append(VirtCheck(tracker[vm_name])) + if delete: + del tracker[vm_name] + def gather_post_virt_checks(self, kubevirt_check_telem): """ Gather final post-run VM health check status. @@ -440,10 +494,9 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): if self.batch_size > 0: for i in range(0, len(self.vm_list), self.batch_size): sub_list = self.vm_list[i : i + self.batch_size] - index = i t = threading.Thread( target=self._run_post_virt_check, - name=str(index), + name=str(i), args=(sub_list, kubevirt_check_telem, post_kubevirt_check_queue), ) post_threads.append(t) @@ -467,46 +520,48 @@ class VirtHealthCheckPlugin(AbstractHealthCheckPlugin): post_virt_check_queue: queue.SimpleQueue, ): """ - Run post-chaos VM health check for a batch. + Run post-chaos VM health check for a batch. Emits one combined VirtCheck + entry per VM containing both ssh_status and vmi_ready. :param vm_list_batch: list of VMs to check :param virt_check_telemetry: telemetry data :param post_virt_check_queue: queue for results """ virt_check_telemetry = [] - virt_check_tracker = {} start_timestamp = datetime.now() for vm in vm_list_batch: try: - if not self.disconnected: - vm_status = self.get_vm_access(vm.vm_name, vm.namespace) - else: - vm_status, new_ip_address, new_node_name = ( - self.check_disconnected_access( - vm.ip_address, vm.node_name, vm.vm_name - ) - ) - if new_ip_address and vm.ip_address != new_ip_address: - vm.new_ip_address = new_ip_address - if new_node_name and vm.node_name != new_node_name: - vm.node_name = new_node_name + ssh_status = self._get_ssh_status(vm) except Exception: - vm_status = False + ssh_status = False - if not vm_status: - virt_check_tracker = { - "vm_name": vm.vm_name, - "ip_address": vm.ip_address, - "namespace": vm.namespace, - "node_name": vm.node_name, - "status": vm_status, - "start_timestamp": start_timestamp.isoformat(), - "new_ip_address": vm.new_ip_address, - "duration": 0, - "end_timestamp": start_timestamp.isoformat(), - } - virt_check_telemetry.append(VirtCheck(virt_check_tracker)) + vmi_ready = self.check_vmi_ready(vm.vm_name, vm.namespace) + combined_status = ssh_status and vmi_ready + + if not combined_status: + if not vmi_ready: + logging.warning( + f"Post-check: VMI {vm.vm_name} in namespace {vm.namespace} is not ready" + ) + virt_check_telemetry.append( + VirtCheck( + { + "vm_name": vm.vm_name, + "ip_address": vm.ip_address, + "namespace": vm.namespace, + "node_name": vm.node_name, + "ssh_status": ssh_status, + "vmi_ready": vmi_ready, + "status": combined_status, + "check_type": self._compute_check_type(ssh_status, vmi_ready), + "start_timestamp": start_timestamp.isoformat(), + "new_ip_address": vm.new_ip_address, + "duration": 0, + "end_timestamp": start_timestamp.isoformat(), + } + ) + ) post_virt_check_queue.put(virt_check_telemetry) diff --git a/tests/test_virt_health_check_plugin.py b/tests/test_virt_health_check_plugin.py index b74cb9ca..ba2b19af 100644 --- a/tests/test_virt_health_check_plugin.py +++ b/tests/test_virt_health_check_plugin.py @@ -17,29 +17,13 @@ How to run: cd /path/to/kraken python3 tests/test_virt_health_check_plugin.py - # Run with pytest - pytest tests/test_virt_health_check_plugin.py -v - # Run with unittest python3 -m unittest tests/test_virt_health_check_plugin.py -v # Run specific test python3 -m unittest tests.test_virt_health_check_plugin.TestVirtHealthCheckPlugin.test_plugin_creation -v - # Run with coverage - coverage run -m pytest tests/test_virt_health_check_plugin.py -v - coverage report - -Requirements: - - krkn_lib library (pip install krkn-lib) - - All scenario plugin dependencies - - All dependencies in requirements.txt - -Note: - - Tests will be skipped if virt_health_check plugin fails to load - - Plugin may fail to load if 'krkn_lib' module is not installed - - Use a virtual environment with all dependencies installed - - Some tests mock KubeVirt components for unit testing + Migrated from test_virt_checker.py to use the plugin architecture. """ @@ -684,10 +668,60 @@ class TestVirtHealthCheckPluginCoverage(unittest.TestCase): self.assertIn("10.0.0.5", check_calls) + @patch("krkn.health_checks.virt_health_check_plugin.time.sleep") + def test_run_virt_check_batch_vmi_ready_to_not_ready_transition(self, mock_sleep): + """Test batch state machine records two segments with correct check_type on vmi_ready→not-ready flip""" + call_count = [0] + def stop_after_two(*_): + call_count[0] += 1 + if call_count[0] >= 2: + self.plugin.current_iterations = self.plugin.iterations + + mock_sleep.side_effect = stop_after_two + + mock_vm = MagicMock() + mock_vm.vm_name = "vm1" + mock_vm.namespace = "default" + mock_vm.ip_address = "10.0.0.1" + mock_vm.node_name = "worker-1" + mock_vm.new_ip_address = "" + self.plugin.disconnected = False + self.plugin.only_failures = False + + # vmi_ready: True on first check, False on second; ssh stays True throughout + vmi_seq = [True, False] + vmi_idx = [0] + def next_vmi_status(*_): + val = vmi_seq[vmi_idx[0] % len(vmi_seq)] + vmi_idx[0] += 1 + return val + + with patch.object(self.plugin, "get_vm_access", return_value=True), \ + patch.object(self.plugin, "check_vmi_ready", side_effect=next_vmi_status): + telemetry_queue = queue.SimpleQueue() + self.plugin._run_virt_check_batch([mock_vm], telemetry_queue) + + result = telemetry_queue.get_nowait() + # Transition closes the first segment and opens a second, so two VirtCheck entries + self.assertGreaterEqual(len(result), 2) + + healthy = [r for r in result if r.vmi_ready] + not_ready = [r for r in result if not r.vmi_ready] + + self.assertGreater(len(healthy), 0) + self.assertTrue(healthy[0].ssh_status) + self.assertTrue(healthy[0].vmi_ready) + self.assertEqual(healthy[0].check_type, "healthy") + + self.assertGreater(len(not_ready), 0) + self.assertTrue(not_ready[0].ssh_status) + self.assertFalse(not_ready[0].vmi_ready) + self.assertEqual(not_ready[0].check_type, "vmi_ready") + # --- _run_post_virt_check --- def test_run_post_virt_check_failed_vm_added_to_telemetry(self): - """Test _run_post_virt_check adds failing VMs to telemetry""" + """Test _run_post_virt_check adds failing VMs to telemetry with correct discriminator fields""" mock_vm = MagicMock() mock_vm.vm_name = "vm-fail" mock_vm.namespace = "default" @@ -696,16 +730,20 @@ class TestVirtHealthCheckPluginCoverage(unittest.TestCase): mock_vm.new_ip_address = "" self.plugin.disconnected = False - with patch.object(self.plugin, "get_vm_access", return_value=False): + with patch.object(self.plugin, "get_vm_access", return_value=False), \ + patch.object(self.plugin, "check_vmi_ready", return_value=False): result_queue = queue.SimpleQueue() self.plugin._run_post_virt_check([mock_vm], [], result_queue) result = result_queue.get_nowait() self.assertEqual(len(result), 1) self.assertEqual(result[0].vm_name, "vm-fail") + self.assertFalse(result[0].ssh_status) + self.assertFalse(result[0].vmi_ready) + self.assertEqual(result[0].check_type, "both") def test_run_post_virt_check_healthy_vm_not_in_telemetry(self): - """Test _run_post_virt_check skips healthy VMs""" + """Test _run_post_virt_check skips healthy VMs (ssh up and VMI ready)""" mock_vm = MagicMock() mock_vm.vm_name = "vm-ok" mock_vm.namespace = "default" @@ -714,12 +752,15 @@ class TestVirtHealthCheckPluginCoverage(unittest.TestCase): mock_vm.new_ip_address = "" self.plugin.disconnected = False - with patch.object(self.plugin, "get_vm_access", return_value=True): + with patch.object(self.plugin, "get_vm_access", return_value=True), \ + patch.object(self.plugin, "check_vmi_ready", return_value=True) as mock_vmi_ready: result_queue = queue.SimpleQueue() self.plugin._run_post_virt_check([mock_vm], [], result_queue) result = result_queue.get_nowait() self.assertEqual(len(result), 0) + # Both checks were invoked; their True results combined to skip this VM from telemetry + mock_vmi_ready.assert_called_once_with("vm-ok", "default") def test_run_post_virt_check_disconnected_mode(self): """Test _run_post_virt_check in disconnected mode""" @@ -740,6 +781,69 @@ class TestVirtHealthCheckPluginCoverage(unittest.TestCase): self.assertEqual(mock_vm.new_ip_address, "10.0.0.2") self.assertEqual(mock_vm.node_name, "worker-2") + def test_run_post_virt_check_check_type_both_when_ssh_and_vmi_fail(self): + """Test VirtCheck.check_type is 'both' when ssh_status and vmi_ready are both False""" + mock_vm = MagicMock() + mock_vm.vm_name = "vm-both-fail" + mock_vm.namespace = "default" + mock_vm.ip_address = "10.0.0.1" + mock_vm.node_name = "worker-1" + mock_vm.new_ip_address = "" + self.plugin.disconnected = False + + with patch.object(self.plugin, "get_vm_access", return_value=False), \ + patch.object(self.plugin, "check_vmi_ready", return_value=False): + result_queue = queue.SimpleQueue() + self.plugin._run_post_virt_check([mock_vm], [], result_queue) + + result = result_queue.get_nowait() + self.assertEqual(len(result), 1) + self.assertFalse(result[0].ssh_status) + self.assertFalse(result[0].vmi_ready) + self.assertEqual(result[0].check_type, "both") + + def test_run_post_virt_check_check_type_ssh_access_when_only_ssh_fails(self): + """Test VirtCheck.check_type is 'ssh_access' when ssh_status is False but vmi_ready is True""" + mock_vm = MagicMock() + mock_vm.vm_name = "vm-ssh-fail" + mock_vm.namespace = "default" + mock_vm.ip_address = "10.0.0.1" + mock_vm.node_name = "worker-1" + mock_vm.new_ip_address = "" + self.plugin.disconnected = False + + with patch.object(self.plugin, "get_vm_access", return_value=False), \ + patch.object(self.plugin, "check_vmi_ready", return_value=True): + result_queue = queue.SimpleQueue() + self.plugin._run_post_virt_check([mock_vm], [], result_queue) + + result = result_queue.get_nowait() + self.assertEqual(len(result), 1) + self.assertFalse(result[0].ssh_status) + self.assertTrue(result[0].vmi_ready) + self.assertEqual(result[0].check_type, "ssh_access") + + def test_run_post_virt_check_check_type_vmi_ready_when_only_vmi_not_ready(self): + """Test VirtCheck.check_type is 'vmi_ready' when ssh_status is True but vmi_ready is False""" + mock_vm = MagicMock() + mock_vm.vm_name = "vm-vmi-fail" + mock_vm.namespace = "default" + mock_vm.ip_address = "10.0.0.1" + mock_vm.node_name = "worker-1" + mock_vm.new_ip_address = "" + self.plugin.disconnected = False + + with patch.object(self.plugin, "get_vm_access", return_value=True), \ + patch.object(self.plugin, "check_vmi_ready", return_value=False): + result_queue = queue.SimpleQueue() + self.plugin._run_post_virt_check([mock_vm], [], result_queue) + + result = result_queue.get_nowait() + self.assertEqual(len(result), 1) + self.assertTrue(result[0].ssh_status) + self.assertFalse(result[0].vmi_ready) + self.assertEqual(result[0].check_type, "vmi_ready") + # --- gather_post_virt_checks --- def test_gather_post_virt_checks_exit_on_failure(self):