diff --git a/lisa/sut_orchestrator/libvirt/ch_platform.py b/lisa/sut_orchestrator/libvirt/ch_platform.py index f67879777d..2427309069 100644 --- a/lisa/sut_orchestrator/libvirt/ch_platform.py +++ b/lisa/sut_orchestrator/libvirt/ch_platform.py @@ -319,6 +319,24 @@ def retry_start_domain_and_attach_logger() -> None: node_context=node_context, ) + def _destroy_domain(self, node_context: NodeContext, log: Logger) -> None: + try: + super()._destroy_domain(node_context, log) + except CloudHypervisorDomainStopError: + if self._reconcile_domain_absence(node_context, log): + node_context.domain_stop_failed = False + return + assert ( + node_context.domain + ), f"Missing domain handle for VM {node_context.vm_name} during cleanup" + if not node_context.domain.isActive(): + log.debug( + f"VM {node_context.vm_name} is already stopped; " + "continuing cleanup without reusing it." + ) + return + raise + def _stop_domain(self, node_context: NodeContext, log: Logger) -> None: if not node_context.passthrough_devices: super()._stop_domain(node_context, log) diff --git a/lisa/sut_orchestrator/libvirt/context.py b/lisa/sut_orchestrator/libvirt/context.py index a8001887e5..a9950def72 100644 --- a/lisa/sut_orchestrator/libvirt/context.py +++ b/lisa/sut_orchestrator/libvirt/context.py @@ -21,6 +21,7 @@ class DataDiskContext: @dataclass class EnvironmentContext: ssh_public_key: str = "" + vm_disks_dir: str = "" # Timeout for the OS to boot and acquire an IP address, in seconds. network_boot_timeout: float = 30.0 diff --git a/lisa/sut_orchestrator/libvirt/platform.py b/lisa/sut_orchestrator/libvirt/platform.py index 0c68fff3ad..88a804a5e9 100644 --- a/lisa/sut_orchestrator/libvirt/platform.py +++ b/lisa/sut_orchestrator/libvirt/platform.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import copy import faulthandler import fnmatch import io @@ -19,6 +20,7 @@ import libvirt import pycdlib import yaml +from retry import retry from lisa import feature, schema, search_space from lisa.environment import Environment @@ -118,6 +120,8 @@ def __init__(self, runbook: schema.Platform) -> None: self.platform_runbook: BaseLibvirtPlatformSchema self.host_node: Node self.vm_disks_dir: str + self._pending_environment_deletions: Dict[str, Environment] = {} + self._environment_deletion_lock = Lock() # used for port forwarding in case of Remote Host self._next_available_port: int @@ -211,6 +215,8 @@ def _initialize(self, *args: Any, **kwargs: Any) -> None: ) def _prepare_environment(self, environment: Environment, log: Logger) -> bool: + self._retry_pending_environment_deletions() + # Ensure environment log directory is created before connecting to any nodes. _ = environment.log_path @@ -249,31 +255,75 @@ def _prepare_environment(self, environment: Environment, log: Logger) -> bool: return self._configure_node_capabilities(environment, log) def _deploy_environment(self, environment: Environment, log: Logger) -> None: + self._retry_pending_environment_deletions() self._deploy_nodes(environment, log) + @retry(tries=3, delay=1) # type: ignore def _delete_environment(self, environment: Environment, log: Logger) -> None: - self._delete_nodes(environment, log) + with self._environment_deletion_lock: + if environment.id not in self._pending_environment_deletions: + # The runner can reset or discard the original environment on failure. + pending = copy.copy(environment) + pending.remove_context() + original_context = get_environment_context(environment) + pending_context = get_environment_context(pending) + pending_context.vm_disks_dir = original_context.vm_disks_dir + pending_context.port_forwarding_list = ( + original_context.port_forwarding_list.copy() + ) + self._pending_environment_deletions[environment.id] = pending + + self._delete_pending_environment( + self._pending_environment_deletions[environment.id], log + ) + def _delete_pending_environment( + self, environment: Environment, log: Logger + ) -> None: + self._delete_nodes(environment, log) if self.host_node.is_remote: self._stop_port_forwarding(environment, log) + del self._pending_environment_deletions[environment.id] + + @retry(tries=3, delay=1) # type: ignore + def _retry_pending_environment_deletions(self) -> None: + errors: List[str] = [] + with self._environment_deletion_lock: + for environment in list(self._pending_environment_deletions.values()): + try: + self._delete_pending_environment(environment, self._log) + except Exception as ex: + self._log.exception( + f"Failed to retry cleanup of environment '{environment.name}'" + ) + errors.append(f"{environment.name}: {ex}") + if errors: + raise LisaException( + f"Pending libvirt environment cleanup failed: {'; '.join(errors)}" + ) def _cleanup(self) -> None: try: - if self.platform_runbook.capture_libvirt_debug_logs: - self._disable_libvirt_debug_log() + self._retry_pending_environment_deletions() + finally: + try: + if self.platform_runbook.capture_libvirt_debug_logs: + self._disable_libvirt_debug_log() - self._disable_libvirt_keepalive() + self._disable_libvirt_keepalive() - self._capture_libvirt_logs() + self._capture_libvirt_logs() - if self.host_node.is_remote: - dmesg_output = self.host_node.tools[Dmesg].get_output(force_run=True) - dmesg_path = self.host_node.local_log_path / "dmesg.txt" - with open(str(dmesg_path), "w") as f: - f.write(dmesg_output) - finally: - if hasattr(self, "device_pool"): - self.device_pool.cleanup() + if self.host_node.is_remote: + dmesg_output = self.host_node.tools[Dmesg].get_output( + force_run=True + ) + dmesg_path = self.host_node.local_log_path / "dmesg.txt" + with open(str(dmesg_path), "w") as f: + f.write(dmesg_output) + finally: + if hasattr(self, "device_pool"): + self.device_pool.cleanup() def _configure_environment(self, environment: Environment, log: Logger) -> None: environment_context = get_environment_context(environment) @@ -448,6 +498,9 @@ def _configure_nodes(self, environment: Environment, log: Logger) -> None: self.vm_disks_dir = os.path.join( self.platform_runbook.hosts[0].lisa_working_dir, vm_name_prefix ) + environment_context = get_environment_context(environment) + environment_context.vm_disks_dir = self.vm_disks_dir + environment_context.port_forwarding_list.clear() assert environment.runbook.nodes_requirement for i, node_space in enumerate(environment.runbook.nodes_requirement): @@ -750,10 +803,12 @@ def _delete_nodes(self, environment: Environment, log: Logger) -> None: self._delete_node(node, log) # Delete VM disks directory. - try: - self.host_node.shell.remove(Path(self.vm_disks_dir), True) - except Exception as ex: - log.warning(f"Failed to delete VM files directory: {ex}") + vm_disks_dir = get_environment_context(environment).vm_disks_dir + if vm_disks_dir: + try: + self.host_node.shell.remove(Path(vm_disks_dir), True) + except Exception as ex: + log.warning(f"Failed to delete VM files directory: {ex}") def _delete_node_watchdog_callback(self) -> None: stderr = sys.__stderr__ @@ -770,7 +825,12 @@ def _destroy_domain(self, node_context: NodeContext, log: Logger) -> None: # In the libvirt API, "destroy" means "stop". self._stop_domain(node_context, log) except libvirt.libvirtError as ex: + if self._is_libvirt_domain_not_found_error(ex) and ( + self._reconcile_domain_absence(node_context, log) + ): + return log.info(f"VM stop failed for {node_context.vm_name}. {ex}") + raise def _undefine_domain(self, node_context: NodeContext, log: Logger) -> None: """Undefine the VM domain, reconnecting if the socket was closed.""" @@ -796,7 +856,28 @@ def retry_undefine_domain() -> None: log=log, ) except libvirt.libvirtError as ex: + if self._is_libvirt_domain_not_found_error(ex) and ( + self._reconcile_domain_absence(node_context, log) + ): + return log.info(f"VM delete failed for {node_context.vm_name}. {ex}") + raise + + def _reconcile_domain_absence(self, node_context: NodeContext, log: Logger) -> bool: + try: + node_context.domain = self._lookup_domain(node_context.vm_name, log) + except libvirt.libvirtError as ex: + if not self._is_libvirt_domain_not_found_error(ex): + raise + if self.libvirt_conn.isAlive() != 1: + raise LisaException( + f"Cannot confirm absence of VM '{node_context.vm_name}' " + "because the libvirt connection is not alive" + ) from ex + node_context.domain = None + log.info(f"Confirmed VM {node_context.vm_name} is already absent.") + return True + return False def _delete_node(self, node: Node, log: Logger) -> None: node_context = get_node_context(node) @@ -843,10 +924,27 @@ def _get_domain_undefine_flags(self) -> int: def _stop_port_forwarding(self, environment: Environment, log: Logger) -> None: log.debug(f"Clearing port forwarding rules for environment {environment.name}") environment_context = get_environment_context(environment) - for port, address in environment_context.port_forwarding_list: - self.host_node.tools[Iptables].stop_forwarding( - port, address, 22, force_run=True - ) + if not environment_context.port_forwarding_list: + return + iptables = self.host_node.tools[Iptables] + for port, address in environment_context.port_forwarding_list.copy(): + for command in ( + f"-D FORWARD -o virbr0 -p tcp -d {address} --dport 22 -j ACCEPT", + f"-t nat -D PREROUTING -p tcp --dport {port} " + f"-j DNAT --to {address}:22", + ): + result = iptables.run(command, force_run=True, sudo=True) + if result.exit_code == 1 and result.stderr.strip().endswith( + "Bad rule (does a matching rule exist in that chain?)." + ): + log.debug(f"Port forwarding rule is already absent: {command}") + continue + result.assert_exit_code( + 0, + message=f"Failed to remove port forwarding rule: {command}", + include_output=True, + ) + environment_context.port_forwarding_list.remove((port, address)) # Retrieve the VMs' dynamic properties (e.g. IP address). def _fill_nodes_metadata(self, environment: Environment, log: Logger) -> None: