diff --git a/sunbeam-python/sunbeam/core/juju.py b/sunbeam-python/sunbeam/core/juju.py index e1f095f8d..27430f1e0 100644 --- a/sunbeam-python/sunbeam/core/juju.py +++ b/sunbeam-python/sunbeam/core/juju.py @@ -759,22 +759,22 @@ def run_cmd_on_machine_unit_payload( ) -> "jubilant.Task": """Run a shell command on a machine unit. - Returns action results irrespective of the return-code - in action results. - :name: unit name :model: Name of the model where the application is located :cmd: Command to run :timeout: Timeout in seconds :returns: Command results - - Command execution failures are part of the results with - return-code, stdout, stderr. + :raises: ExecFailedException if command execution fails """ with self._model(model) as juju: try: task = juju.exec(cmd, unit=name, wait=timeout) - except jubilant.TaskError as e: + except ( + jubilant.TaskError, + jubilant.CLIError, + TimeoutError, + ValueError, + ) as e: raise ExecFailedException( f"Failed to run command {cmd!r} on unit" f" {name!r} in model {model!r}: {e}" @@ -825,7 +825,12 @@ def run_cmd_on_unit_payload( stdout, _ = juju._cli(*args, "--", *(cmd.split()), log=False) except jubilant.CLIError as e: stdout = e.stdout - return json.loads(stdout)[name]["results"] + try: + return json.loads(stdout)[name]["results"] + except (json.JSONDecodeError, KeyError, TypeError) as e: + raise ExecFailedException( + f"Failed to parse command result for unit {name!r}" + ) from e def run_action( self, diff --git a/sunbeam-python/sunbeam/provider/maas/client.py b/sunbeam-python/sunbeam/provider/maas/client.py index 824fc8300..75e138642 100644 --- a/sunbeam-python/sunbeam/provider/maas/client.py +++ b/sunbeam-python/sunbeam/provider/maas/client.py @@ -341,6 +341,7 @@ def _convert_raw_machine(machine_raw: dict, root_disk: dict | None) -> dict: machine = { "system_id": machine_raw["system_id"], "hostname": machine_raw["hostname"], + "fqdn": machine_raw["fqdn"], "roles": list(set(tag_names).intersection(RoleTags.values())), "zone": machine_raw["zone"]["name"], "status": machine_raw["status_name"], diff --git a/sunbeam-python/sunbeam/provider/maas/commands.py b/sunbeam-python/sunbeam/provider/maas/commands.py index b456051fa..10ac7ee9f 100644 --- a/sunbeam-python/sunbeam/provider/maas/commands.py +++ b/sunbeam-python/sunbeam/provider/maas/commands.py @@ -125,6 +125,8 @@ CheckCinderVolumeDistributionStep, DeployCinderVolumeApplicationStep, DestroyCinderVolumeApplicationStep, + DisableCinderVolumeServicesStep, + RemoveCinderVolumeServicesStep, RemoveCinderVolumeUnitsStep, ) from sunbeam.steps.clusterd import APPLICATION as CLUSTERD_APPLICATION @@ -137,6 +139,7 @@ DeployHypervisorApplicationStep, DestroyHypervisorApplicationStep, ReapplyHypervisorTerraformPlanStep, + RemoveHypervisorReferencesStep, RemoveHypervisorUnitStep, ) from sunbeam.steps.juju import ( @@ -172,6 +175,7 @@ CheckMicrocephDistributionStep, DeployMicrocephApplicationStep, DestroyMicrocephApplicationStep, + RemoveMicrocephOSDsStep, RemoveMicrocephUnitsStep, SetCephMgrPoolSizeStep, ) @@ -1693,6 +1697,9 @@ def remove_node(ctx: click.Context, name: str, force: bool, show_hints: bool) -> run_plan(check_plan, console, show_hints) + maas_client = MaasClient.from_deployment(deployment) + machine = get_machine(maas_client, name) + plan = [ MigrateK8SKubeconfigStep( client, name, jhelper, deployment.openstack_machines_model @@ -1701,20 +1708,46 @@ def remove_node(ctx: click.Context, name: str, force: bool, show_hints: bool) -> RemoveHypervisorUnitStep( client, jhelper, - deployment, + None, name, deployment.openstack_machines_model, force, ), + DisableCinderVolumeServicesStep( + jhelper, + deployment, + machine["hostname"], + machine["fqdn"], + ), RemoveCinderVolumeUnitsStep( client, name, jhelper, deployment.openstack_machines_model ), + RemoveCinderVolumeServicesStep( + jhelper, + deployment, + machine["hostname"], + machine["fqdn"], + ), + RemoveMicrocephOSDsStep( + client, + name, + jhelper, + deployment.openstack_machines_model, + force=force, + ), RemoveMicrocephUnitsStep( client, name, jhelper, deployment.openstack_machines_model ), RemoveMicroOVNUnitsStep( client, name, jhelper, deployment.openstack_machines_model ), + RemoveHypervisorReferencesStep( + jhelper, + deployment, + machine["hostname"], + machine["fqdn"], + force=force, + ), CordonK8SUnitStep(client, name, jhelper, deployment.openstack_machines_model), DrainK8SUnitStep( client, name, jhelper, deployment.openstack_machines_model, remove_pvc=True diff --git a/sunbeam-python/sunbeam/steps/cinder_volume.py b/sunbeam-python/sunbeam/steps/cinder_volume.py index 53ceed0fe..a2a60fcf0 100644 --- a/sunbeam-python/sunbeam/steps/cinder_volume.py +++ b/sunbeam-python/sunbeam/steps/cinder_volume.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import logging +import typing from typing import Any import sunbeam.steps.microceph as microceph @@ -10,19 +11,37 @@ from sunbeam.clusterd.service import ( NodeNotExistInClusterException, ) -from sunbeam.core.common import BaseStep, Result, ResultType, Role, StepContext +from sunbeam.core.common import ( + BaseStep, + Result, + ResultType, + Role, + StepContext, + SunbeamException, +) from sunbeam.core.deployment import Deployment, Networks from sunbeam.core.juju import ( ApplicationNotFoundException, + JujuException, JujuHelper, ) from sunbeam.core.manifest import CharmManifest, Manifest +from sunbeam.core.openstack import OPENSTACK_MODEL +from sunbeam.core.openstack_api import get_admin_connection from sunbeam.core.steps import ( DeployMachineApplicationStep, DestroyMachineApplicationStep, RemoveMachineUnitsStep, ) from sunbeam.core.terraform import TerraformException, TerraformHelper +from sunbeam.lazy import LazyImport + +if typing.TYPE_CHECKING: + import openstack + from keystoneauth1 import exceptions as keystoneauth_exceptions +else: + keystoneauth_exceptions = LazyImport("keystoneauth1.exceptions") + openstack = LazyImport("openstack") LOG = logging.getLogger(__name__) CONFIG_KEY = "TerraformVarsCinderVolumePlan" @@ -31,6 +50,27 @@ CINDER_VOLUME_UNIT_TIMEOUT = ( 1800 # 30 minutes, adding / removing units can take a long time ) +CINDER_APPLICATION = "cinder" +CINDER_API_CONTAINER = "cinder-api" +CINDER_VOLUME_BINARY = "cinder-volume" +CINDER_SERVICE_REMOVE_REASON = "Removing node from cluster" +CINDER_SERVICE_REMOVE_TIMEOUT = 1800 + + +def get_cinder_volume_services( + conn: "openstack.connection.Connection", + hostname: str, + fqdn: str, +) -> list[Any]: + """Return cinder-volume services for the exact node hostnames.""" + expected_hosts = {hostname, fqdn} + + services = [] + for service in conn.block_storage.services(binary=CINDER_VOLUME_BINARY): + if service.host.split("@", 1)[0] in expected_hosts: + services.append(service) + + return sorted(services, key=lambda service: service.host) def get_mandatory_control_plane_offers( @@ -231,6 +271,187 @@ def get_unit_timeout(self) -> int: return CINDER_VOLUME_UNIT_TIMEOUT +class _CinderVolumeServiceStep(BaseStep): + """Shared service discovery for Cinder cleanup steps.""" + + def __init__( + self, + name: str, + description: str, + jhelper: JujuHelper, + deployment: Deployment, + hostname: str, + fqdn: str, + ): + super().__init__(name, description) + self.jhelper = jhelper + self.deployment = deployment + self.hostname = hostname + self.fqdn = fqdn + self.connection: Any | None = None + self.services: list[Any] = [] + + def _discover_services(self) -> Result: + """Discover Cinder service records for the target node.""" + try: + self.connection = get_admin_connection(self.jhelper, self.deployment) + self.services = get_cinder_volume_services( + self.connection, self.hostname, self.fqdn + ) + except ( + openstack.exceptions.SDKException, + keystoneauth_exceptions.ClientException, + ) as e: + LOG.warning("Failed to discover Cinder volume services: %r", e) + return Result(ResultType.FAILED, str(e)) + + if not self.services: + return Result(ResultType.SKIPPED) + return Result(ResultType.COMPLETED) + + def _current_services(self) -> list[Any]: + """Return matching Cinder service records.""" + if self.connection is None: + raise SunbeamException("Cinder admin connection not found") + return get_cinder_volume_services(self.connection, self.hostname, self.fqdn) + + +class DisableCinderVolumeServicesStep(_CinderVolumeServiceStep): + """Disable matching Cinder volume services before unit removal.""" + + def __init__( + self, + jhelper: JujuHelper, + deployment: Deployment, + hostname: str, + fqdn: str, + ): + super().__init__( + "Disable Cinder Volume services", + "Disabling Cinder Volume services", + jhelper, + deployment, + hostname, + fqdn, + ) + + def is_skip(self, context: StepContext) -> Result: + """Determine whether matching Cinder services exist.""" + return self._discover_services() + + def run(self, context: StepContext) -> Result: + """Disable each enabled matching Cinder volume service.""" + if self.connection is None: + return Result(ResultType.FAILED, "Cinder admin connection not found") + try: + for service in self.services: + if service.status != "enabled": + continue + LOG.info("Disabling %s on %s", service.binary, service.host) + self.connection.block_storage.disable_service( + service, reason=CINDER_SERVICE_REMOVE_REASON + ) + except ( + openstack.exceptions.SDKException, + keystoneauth_exceptions.ClientException, + ) as e: + LOG.warning("Failed to disable Cinder volume service: %r", e) + return Result(ResultType.FAILED, str(e)) + + return Result(ResultType.COMPLETED) + + +class RemoveCinderVolumeServicesStep(_CinderVolumeServiceStep): + """Remove matching Cinder volume service records after unit removal.""" + + def __init__( + self, + jhelper: JujuHelper, + deployment: Deployment, + hostname: str, + fqdn: str, + ): + super().__init__( + "Remove Cinder Volume services", + "Removing Cinder Volume services", + jhelper, + deployment, + hostname, + fqdn, + ) + + def is_skip(self, context: StepContext) -> Result: + """Determine whether matching Cinder services exist.""" + return self._discover_services() + + def _healthy_units(self) -> list[str]: + """Return healthy Cinder API units with a leader first.""" + try: + application = self.jhelper.get_application( + CINDER_APPLICATION, OPENSTACK_MODEL + ) + except JujuException as e: + LOG.warning("Failed to find Cinder control-plane units: %r", e) + return [] + + units = [ + (name, unit) + for name, unit in application.units.items() + if unit.workload_status.current == "active" + and unit.juju_status.current == "idle" + ] + units.sort(key=lambda item: (not item[1].leader, item[0])) + return [name for name, _ in units] + + def _remove_service(self, unit: str, service: Any) -> None: + """Remove one Cinder service record through a healthy unit.""" + command = f"cinder-manage service remove {CINDER_VOLUME_BINARY} {service.host}" + result = self.jhelper.run_cmd_on_unit_payload( + unit, + OPENSTACK_MODEL, + command, + CINDER_API_CONTAINER, + timeout=CINDER_SERVICE_REMOVE_TIMEOUT, + ) + if result.get("return-code") != 0: + raise JujuException(f"Failed to remove Cinder service {service.host}") + + def run(self, context: StepContext) -> Result: + """Remove matching records and verify that none remain.""" + if self.connection is None: + return Result(ResultType.FAILED, "Cinder admin connection not found") + + try: + remaining = self.services + healthy_units = self._healthy_units() + if not healthy_units: + raise SunbeamException( + "No healthy Cinder control-plane units available" + ) + + for unit in healthy_units: + for service in remaining: + try: + self._remove_service(unit, service) + except JujuException as e: + LOG.warning( + "Failed to remove Cinder service on %s: %r", unit, e + ) + break + remaining = self._current_services() + if not remaining: + return Result(ResultType.COMPLETED) + + raise SunbeamException("Cinder service records remain after removal") + except ( + SunbeamException, + openstack.exceptions.SDKException, + keystoneauth_exceptions.ClientException, + ) as e: + LOG.warning("Failed to remove Cinder volume services: %r", e) + return Result(ResultType.FAILED, str(e)) + + class CheckCinderVolumeDistributionStep(BaseStep): _APPLICATION = APPLICATION diff --git a/sunbeam-python/sunbeam/steps/hypervisor.py b/sunbeam-python/sunbeam/steps/hypervisor.py index 45cae9b2e..6c7bac6b0 100644 --- a/sunbeam-python/sunbeam/steps/hypervisor.py +++ b/sunbeam-python/sunbeam/steps/hypervisor.py @@ -34,7 +34,12 @@ JujuStepHelper, ) from sunbeam.core.manifest import Manifest -from sunbeam.core.openstack_api import remove_hypervisor +from sunbeam.core.openstack_api import ( + get_admin_connection, + remove_compute_service, + remove_hypervisor, + remove_network_service, +) from sunbeam.core.steps import ( DeployMachineApplicationStep, DestroyMachineApplicationStep, @@ -49,7 +54,9 @@ if typing.TYPE_CHECKING: import openstack + from keystoneauth1 import exceptions as keystoneauth_exceptions else: + keystoneauth_exceptions = LazyImport("keystoneauth1.exceptions") openstack = LazyImport("openstack") LOG = logging.getLogger(__name__) @@ -60,6 +67,8 @@ HYPERVISOR_UNIT_TIMEOUT = ( 1800 # 30 minutes, adding / removing units can take a long time ) +HYPERVISOR_REFERENCES_TIMEOUT = 300 +HYPERVISOR_REFERENCES_POLL_INTERVAL = 10 class DeployHypervisorApplicationStep(DeployMachineApplicationStep): @@ -330,6 +339,76 @@ def run(self, context: StepContext) -> Result: return Result(ResultType.COMPLETED) +class RemoveHypervisorReferencesStep(BaseStep): + """Remove Nova and Neutron references to a hypervisor.""" + + def __init__( + self, + jhelper: JujuHelper, + deployment: Deployment, + hostname: str, + fqdn: str, + force: bool = False, + ): + super().__init__( + "Remove openstack-hypervisor references", + "Remove openstack-hypervisor references from the control plane", + ) + self.jhelper = jhelper + self.deployment = deployment + self.force = force + self._hostnames = tuple(dict.fromkeys((hostname, fqdn))) + + def _remove_references(self) -> None: + """Remove references and raise while records remain.""" + conn = get_admin_connection(self.jhelper, self.deployment) + for hostname in self._hostnames: + remove_compute_service(hostname, conn) + remove_network_service(hostname, conn) + remaining_hosts = [] + for hostname in self._hostnames: + compute_services = list(conn.compute.services(host=hostname)) + network_agents = list(conn.network.agents(host=hostname)) + if compute_services or network_agents: + remaining_hosts.append(hostname) + if remaining_hosts: + raise tenacity.TryAgain( + f"Hypervisor references remain for {', '.join(remaining_hosts)}" + ) + + def run(self, context: StepContext) -> Result: + """Remove references until Nova and Neutron report none remain.""" + client_exceptions = ( + openstack.exceptions.SDKException, + keystoneauth_exceptions.ClientException, + ) + retry_exceptions: tuple[type[BaseException], ...] = (tenacity.TryAgain,) + if not self.force: + retry_exceptions += client_exceptions + try: + for attempt in tenacity.Retrying( + stop=tenacity.stop_after_delay(HYPERVISOR_REFERENCES_TIMEOUT), + wait=tenacity.wait_fixed(HYPERVISOR_REFERENCES_POLL_INTERVAL), + retry=tenacity.retry_if_exception_type(retry_exceptions), + reraise=True, + ): + with attempt: + self._remove_references() + except client_exceptions as e: + LOG.error("Failed to remove hypervisor references from control plane") + if self.force: + LOG.warning( + "Force mode set, ignoring following exceptions", exc_info=True + ) + return Result(ResultType.COMPLETED) + return Result(ResultType.FAILED, str(e)) + except tenacity.TryAgain as e: + LOG.error("Failed to remove hypervisor references from control plane") + return Result(ResultType.FAILED, str(e)) + + return Result(ResultType.COMPLETED) + + class ReapplyHypervisorTerraformPlanStep(BaseStep): """Reapply openstack-hyervisor terraform plan.""" diff --git a/sunbeam-python/sunbeam/steps/microceph.py b/sunbeam-python/sunbeam/steps/microceph.py index e87aefdce..8957677d0 100644 --- a/sunbeam-python/sunbeam/steps/microceph.py +++ b/sunbeam-python/sunbeam/steps/microceph.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import ast +import json import logging from typing import Any @@ -24,6 +25,7 @@ from sunbeam.core.juju import ( ActionFailedException, ApplicationNotFoundException, + ExecFailedException, JujuHelper, LeaderNotFoundException, UnitNotFoundException, @@ -231,6 +233,273 @@ def get_unit_timeout(self) -> int: return MICROCEPH_UNIT_TIMEOUT +class RemoveMicrocephOSDsStep(BaseStep): + """Remove a node's MicroCeph OSDs before removing its unit.""" + + _OSD_REMOVE_TIMEOUT = 1800 + _COMMAND_TIMEOUT = _OSD_REMOVE_TIMEOUT + 60 + + def __init__( + self, + client: Client, + name: str, + jhelper: JujuHelper, + model: str, + force: bool = False, + ): + super().__init__( + "Remove MicroCeph OSDs", + "Removing MicroCeph OSDs", + ) + self.client = client + self.node = name + self.jhelper = jhelper + self.model = model + self.force = force + self.unit: str | None = None + self._units: list[str] = [] + + def _prepare(self) -> Result: + """Find a unit that can run cleanup before the target is removed.""" + self.unit = None + self._units = [] + try: + node_info = self.client.cluster.get_node_info(self.node) + except NodeNotExistInClusterException: + node_info = None + + machines = self.jhelper.get_machines(self.model) + target_machine_ids: set[str] = set() + if node_info is not None: + machine_id = node_info.get("machineid") + if machine_id is not None and str(machine_id) not in {"", "-1"}: + target_machine_ids.add(str(machine_id)) + target_machine_ids.update( + str(machine_id) + for machine_id, machine in machines.items() + if machine.hostname == self.node + ) + try: + app = self.jhelper.get_application(APPLICATION, self.model) + except ApplicationNotFoundException: + LOG.debug("Failed to get application", exc_info=True) + return Result( + ResultType.SKIPPED, + f"Application {APPLICATION} has not been deployed yet", + ) + + units = app.units + if not units: + if self.force: + return Result(ResultType.SKIPPED) + return Result( + ResultType.FAILED, + f"No MicroCeph unit is available to clean up {self.node}", + ) + + known_machine_ids = {str(machine_id) for machine_id in machines} + target_units: list[str] = [] + surviving_units: list[str] = [] + for unit_name, unit in units.items(): + machine_id = str(unit.machine) + if machine_id in target_machine_ids: + target_units.append(unit_name) + elif machine_id in known_machine_ids: + surviving_units.append(unit_name) + + if not (target_units or surviving_units): + return Result( + ResultType.FAILED, + f"Unable to identify a MicroCeph unit for {self.node}", + ) + + def unit_sort_key(unit_name: str) -> tuple[bool, bool, str]: + unit = units[unit_name] + healthy = ( + unit.workload_status.current == "active" + and unit.juju_status.current == "idle" + ) + return not healthy, unit_name in target_units, unit_name + + self._units = sorted(surviving_units + target_units, key=unit_sort_key) + self.unit = self._units[0] + + return Result(ResultType.COMPLETED) + + def is_skip(self, context: StepContext) -> Result: + """Determine whether cleanup can run and whether it is needed.""" + return self._prepare() + + def _run_command(self, command: str, *, allow_fallback: bool) -> str: + """Run a MicroCeph command and fail on transport or command errors.""" + if self.unit is None: + raise SunbeamException("MicroCeph cleanup unit is not available") + + candidates = [self.unit] + if allow_fallback: + candidates.extend(unit for unit in self._units if unit != self.unit) + for index, unit in enumerate(candidates): + try: + result = self.jhelper.run_cmd_on_machine_unit_payload( + unit, + self.model, + command, + timeout=self._COMMAND_TIMEOUT, + ) + except ExecFailedException as e: + if index < len(candidates) - 1: + LOG.warning( + "Failed to run MicroCeph command on %s: %r; " + "trying another unit", + unit, + e, + ) + continue + raise SunbeamException(f"Failed to run {command!r}: {e}") from e + self.unit = unit + break + + return result.stdout + + @staticmethod + def _parse_configured_disks(stdout: Any) -> list[dict[str, Any]]: + """Parse MicroCeph's configured disk records.""" + if not isinstance(stdout, str): + raise ValueError("Configured disk listing output is not text") + payload = json.loads(stdout) + if not isinstance(payload, dict) or "ConfiguredDisks" not in payload: + raise ValueError("Configured disk listing has no ConfiguredDisks") + configured_disks = payload["ConfiguredDisks"] + if not isinstance(configured_disks, list): + raise ValueError("ConfiguredDisks is not a list") + + required_fields = {"osd", "location"} + disks: list[dict[str, Any]] = [] + for disk in configured_disks: + if not isinstance(disk, dict) or not required_fields <= set(disk): + raise ValueError("Configured disk has an invalid schema") + if type(disk["osd"]) is not int or disk["osd"] < 0: + raise ValueError("Configured disk has an invalid OSD ID") + if not isinstance(disk["location"], str): + raise ValueError("Configured disk has an invalid location") + disks.append(disk) + return disks + + @staticmethod + def _parse_crush_tree(stdout: Any) -> list[dict[str, Any]]: + """Parse the JSON nodes returned by the Ceph OSD tree command.""" + if not isinstance(stdout, str): + raise ValueError("CRUSH tree output is not text") + tree = json.loads(stdout) + if not isinstance(tree, dict) or not isinstance(tree.get("nodes"), list): + raise ValueError("CRUSH tree output has no nodes list") + + nodes = tree["nodes"] + for node in nodes: + if not isinstance(node, dict): + raise ValueError("CRUSH tree node is not an object") + if not isinstance(node.get("name"), str) or not isinstance( + node.get("type"), str + ): + raise ValueError("CRUSH tree node has an invalid schema") + if node.get("type") == "host" and not isinstance( + node.get("children"), list + ): + raise ValueError("CRUSH host has no children list") + if "children" in node and not isinstance(node["children"], list): + raise ValueError("CRUSH tree node children is not a list") + return nodes + + def _list_configured_osd_ids(self) -> list[int]: + """Return target OSD IDs that still exist in the MicroCeph database.""" + try: + disks = self._parse_configured_disks( + self._run_command("microceph disk list --json", allow_fallback=True) + ) + except ValueError as e: + raise SunbeamException( + f"Failed to parse configured disk listing: {e}" + ) from e + return sorted({disk["osd"] for disk in disks if disk["location"] == self.node}) + + def _list_crush_osd_ids(self) -> list[int]: + """Return OSD IDs under the target CRUSH host.""" + try: + nodes = self._parse_crush_tree( + self._run_command( + "microceph.ceph osd tree --format json", allow_fallback=True + ) + ) + except ValueError as e: + raise SunbeamException(f"Failed to parse CRUSH tree: {e}") from e + + host = next( + ( + node + for node in nodes + if node["name"] == self.node and node["type"] == "host" + ), + None, + ) + if host is None: + return [] + children = host["children"] + if any(type(child) is not int or child < 0 for child in children): + raise SunbeamException("CRUSH host has an invalid OSD ID") + return sorted(set(children)) + + def _list_target_osds(self) -> tuple[list[int], list[int]]: + """Read both target OSD sources before changing either source.""" + return self._list_configured_osd_ids(), self._list_crush_osd_ids() + + def run(self, context: StepContext) -> Result: + """Remove DB-backed OSDs and verify both MicroCeph and CRUSH state.""" + if self.unit is None: + preparation = self._prepare() + if preparation.result_type != ResultType.COMPLETED: + return preparation + + try: + configured_osds, crush_osds = self._list_target_osds() + crush_only_osds = sorted(set(crush_osds) - set(configured_osds)) + if crush_only_osds: + return Result( + ResultType.FAILED, + f"CRUSH-only OSDs for {self.node}: {crush_only_osds}", + ) + + for osd_id in configured_osds: + command = ( + f"microceph disk remove osd.{osd_id} " + f"--timeout {self._OSD_REMOVE_TIMEOUT}" + ) + if self.force: + command += ( + " --confirm-failure-domain-downgrade --bypass-safety-checks" + ) + self._run_command(command, allow_fallback=False) + + if not configured_osds: + return Result(ResultType.COMPLETED) + + remaining_configured, remaining_crush = self._list_target_osds() + if remaining_configured: + return Result( + ResultType.FAILED, + f"Configured OSDs remain for {self.node}: {remaining_configured}", + ) + if remaining_crush: + return Result( + ResultType.FAILED, + f"CRUSH OSDs remain for {self.node}: {remaining_crush}", + ) + except SunbeamException as e: + LOG.debug("Failed to clean up MicroCeph OSDs", exc_info=True) + return Result(ResultType.FAILED, str(e)) + + return Result(ResultType.COMPLETED) + + class ConfigureMicrocephOSDStep(BaseStep): """Configure Microceph OSD disks.""" diff --git a/sunbeam-python/tests/unit/sunbeam/core/test_juju.py b/sunbeam-python/tests/unit/sunbeam/core/test_juju.py index 7a248e4e7..2a413f404 100644 --- a/sunbeam-python/tests/unit/sunbeam/core/test_juju.py +++ b/sunbeam-python/tests/unit/sunbeam/core/test_juju.py @@ -208,6 +208,21 @@ def test_run_cmd_on_machine_unit_payload_success(jhelper, juju): assert result.results["result"] == "ok" +@pytest.mark.parametrize( + "error", + [ + TimeoutError("timed out"), + jubilant.CLIError(1, ["exec"], "", "controller unavailable"), + ValueError("unit has no result"), + ], +) +def test_run_cmd_on_machine_unit_payload_normalizes_exec_errors(jhelper, juju, error): + juju.exec.side_effect = error + + with pytest.raises(jujulib.ExecFailedException): + jhelper.run_cmd_on_machine_unit_payload("app/0", "test-model", "ls") + + def test_run_action_success(jhelper, juju): juju.run = Mock(return_value=Mock(success=True, results={"app": "bar"})) @@ -238,6 +253,24 @@ def test_run_cmd_on_unit_payload_cli_error(jhelper, juju): assert result["err"] == "fail" +@pytest.mark.parametrize( + "stdout", + ["", "{}", json.dumps({"app/0": {}})], +) +def test_run_cmd_on_unit_payload_normalizes_invalid_result(jhelper, juju, stdout): + juju._cli.return_value = (stdout, "") + + with pytest.raises(jujulib.ExecFailedException): + jhelper.run_cmd_on_unit_payload("app/0", "test-model", "ls", "container") + + +def test_run_cmd_on_unit_payload_normalizes_invalid_cli_error(jhelper, juju): + juju._cli.side_effect = jubilant.CLIError(1, ["exec"], "", "unit unavailable") + + with pytest.raises(jujulib.ExecFailedException): + jhelper.run_cmd_on_unit_payload("app/0", "test-model", "ls", "container") + + def test_set_model_config(jhelper, juju): jhelper.set_model_config("test-model", {"app": "bar"}) juju.model_config.assert_called() diff --git a/sunbeam-python/tests/unit/sunbeam/provider/maas/test_maas.py b/sunbeam-python/tests/unit/sunbeam/provider/maas/test_maas.py index fbdcc42b8..9532595cf 100644 --- a/sunbeam-python/tests/unit/sunbeam/provider/maas/test_maas.py +++ b/sunbeam-python/tests/unit/sunbeam/provider/maas/test_maas.py @@ -17,6 +17,7 @@ from sunbeam.core.deployment import Networks from sunbeam.core.deployments import DeploymentsConfig from sunbeam.core.juju import ControllerNotFoundException +from sunbeam.provider.maas.client import _convert_raw_machine from sunbeam.provider.maas.commands import ( configure_cmd, remove_node, @@ -56,14 +57,46 @@ ZoneBalanceCheck, ZonesCheck, ) +from sunbeam.steps.cinder_volume import ( + DisableCinderVolumeServicesStep, + RemoveCinderVolumeServicesStep, + RemoveCinderVolumeUnitsStep, +) +from sunbeam.steps.hypervisor import ( + RemoveHypervisorReferencesStep, + RemoveHypervisorUnitStep, +) from sunbeam.steps.juju import RemoveJujuMachineStep -from sunbeam.steps.microovn import ReapplyMicroOVNTerraformPlanStep +from sunbeam.steps.microceph import RemoveMicrocephOSDsStep, RemoveMicrocephUnitsStep +from sunbeam.steps.microovn import ( + ReapplyMicroOVNTerraformPlanStep, + RemoveMicroOVNUnitsStep, +) from sunbeam.steps.role_distributor import ( ReapplyRoleDistributorApplicationStep, RemoveRoleDistributorUnitsStep, ) +class TestConvertRawMachine: + def test_preserves_fqdn(self): + machine_raw = { + "system_id": "sysid", + "hostname": "cloud-4", + "fqdn": "cloud-4.maas", + "blockdevice_set": [], + "interface_set": [], + "zone": {"name": "default"}, + "status_name": "Ready", + "cpu_count": 4, + "memory": 8192, + } + + machine = _convert_raw_machine(machine_raw, None) + + assert machine["fqdn"] == "cloud-4.maas" + + class TestMaasConfigureCommand: def test_network_agents_include_all_microovn_nodes( self, @@ -2484,7 +2517,98 @@ def test_storage_ippool_label_with_different_names(self): assert deployment.storage_ip_pool == expected_label -class TestRemoveNodeRoleDistributor: +class TestRemoveNode: + @pytest.fixture(autouse=True) + def mock_maas_machine(self, mocker): + maas_client = mocker.patch( + "sunbeam.provider.maas.commands.MaasClient.from_deployment" + ).return_value + get_machine = mocker.patch( + "sunbeam.provider.maas.commands.get_machine", + return_value={"hostname": "node-1", "fqdn": "node-1.maas"}, + ) + return maas_client, get_machine + + @patch("sunbeam.provider.maas.commands.JujuHelper") + @patch("sunbeam.provider.maas.commands.run_preflight_checks") + @patch("sunbeam.provider.maas.commands.run_plan") + @pytest.mark.parametrize( + ("cli_args", "force"), + [(["node-1"], False), (["--force", "node-1"], True)], + ) + def test_remove_orders_cleanup_steps( + self, + run_plan_cmd, + run_preflight, + juju_helper, + mock_maas_machine, + cli_args, + force, + ): + maas_client, get_machine = mock_maas_machine + deployment = Mock() + deployment.openstack_machines_model = "openstack-machines" + deployment.get_ovn_manager.return_value.get_machines.return_value = [] + + result = CliRunner().invoke(remove_node, cli_args, obj=deployment) + + assert result.exit_code == 0, result.output + get_machine.assert_called_once_with(maas_client, "node-1") + plan = run_plan_cmd.call_args_list[1][0][0] + hypervisor_unit_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveHypervisorUnitStep) + ) + microovn_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveMicroOVNUnitsStep) + ) + references_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveHypervisorReferencesStep) + ) + hypervisor_unit = plan[hypervisor_unit_index] + references = plan[references_index] + + assert hypervisor_unit.deployment is None + assert references._hostnames == ("node-1", "node-1.maas") + assert references.force is force + assert microovn_index < references_index + + disable_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, DisableCinderVolumeServicesStep) + ) + cinder_units_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveCinderVolumeUnitsStep) + ) + cinder_services_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveCinderVolumeServicesStep) + ) + assert disable_index < cinder_units_index < cinder_services_index + + osd_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveMicrocephOSDsStep) + ) + unit_index = next( + i + for i, step in enumerate(plan) + if isinstance(step, RemoveMicrocephUnitsStep) + ) + assert osd_index < unit_index + assert plan[osd_index].node == "node-1" + assert plan[osd_index].force is force + @patch("sunbeam.provider.maas.commands.JujuHelper") @patch("sunbeam.provider.maas.commands.run_preflight_checks") @patch("sunbeam.provider.maas.commands.run_plan") diff --git a/sunbeam-python/tests/unit/sunbeam/steps/test_cinder_volume.py b/sunbeam-python/tests/unit/sunbeam/steps/test_cinder_volume.py index 930c10ad9..f7a2f82c3 100644 --- a/sunbeam-python/tests/unit/sunbeam/steps/test_cinder_volume.py +++ b/sunbeam-python/tests/unit/sunbeam/steps/test_cinder_volume.py @@ -4,12 +4,19 @@ from unittest.mock import MagicMock, Mock, patch import pytest +from keystoneauth1.exceptions.catalog import EndpointNotFound +from keystoneauth1.exceptions.connection import ConnectFailure +from sunbeam.core.common import ResultType +from sunbeam.core.juju import ExecFailedException, JujuException from sunbeam.steps.cinder_volume import ( CINDER_VOLUME_APP_TIMEOUT, CINDER_VOLUME_UNIT_TIMEOUT, DeployCinderVolumeApplicationStep, + DisableCinderVolumeServicesStep, + RemoveCinderVolumeServicesStep, RemoveCinderVolumeUnitsStep, + get_cinder_volume_services, ) @@ -361,3 +368,344 @@ def test_get_unit_timeout(self, remove_cinder_volume_units_step): remove_cinder_volume_units_step.get_unit_timeout() == CINDER_VOLUME_UNIT_TIMEOUT ) + + +class TestCinderVolumeServiceCleanup: + @pytest.fixture + def connection(self): + return Mock() + + @pytest.fixture + def services(self): + return [ + Mock(binary="cinder-volume", host="cloud-4@backend-a", status="enabled"), + Mock( + binary="cinder-volume", host="cloud-4.maas@backend-b", status="disabled" + ), + ] + + def test_get_cinder_volume_services_matches_exact_hosts(self): + matching_short = Mock(binary="cinder-volume", host="cloud-4@backend-a") + matching_fqdn = Mock(binary="cinder-volume", host="cloud-4.maas@backend-b") + unrelated = Mock(binary="cinder-volume", host="cloud-40@backend-a") + conn = Mock() + conn.block_storage.services.return_value = [ + unrelated, + matching_fqdn, + matching_short, + ] + + services = get_cinder_volume_services(conn, "cloud-4", "cloud-4.maas") + + conn.block_storage.services.assert_called_once_with(binary="cinder-volume") + assert services == [matching_fqdn, matching_short] + + def test_disable_enabled_services( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + return_value=services, + ) + step = DisableCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + connection.block_storage.disable_service.assert_called_once_with( + services[0], reason="Removing node from cluster" + ) + + def test_discovery_fails_on_keystoneauth_error( + self, + mocker, + basic_jhelper, + basic_deployment, + step_context, + ): + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + side_effect=EndpointNotFound("volume endpoint missing"), + ) + step = DisableCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.FAILED + + def test_disable_fails_on_keystoneauth_error( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + connection.block_storage.disable_service.side_effect = ConnectFailure( + "volume endpoint unavailable" + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + return_value=services, + ) + step = DisableCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + + def test_remove_prefers_healthy_nonleader_when_leader_unhealthy( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + unhealthy_leader = Mock( + workload_status=Mock(current="blocked"), + juju_status=Mock(current="idle"), + leader=True, + ) + healthy_unit = Mock( + workload_status=Mock(current="active"), + juju_status=Mock(current="idle"), + leader=False, + ) + application = Mock( + units={"cinder/0": unhealthy_leader, "cinder/1": healthy_unit} + ) + basic_jhelper.get_application.return_value = application + basic_jhelper.run_cmd_on_unit_payload.return_value = {"return-code": 0} + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + side_effect=[services, []], + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + assert all( + call.args[0] == "cinder/1" + for call in basic_jhelper.run_cmd_on_unit_payload.call_args_list + ) + + def test_remove_succeeds_when_ambiguous_command_error_has_no_records( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + unit = Mock( + workload_status=Mock(current="active"), + juju_status=Mock(current="idle"), + leader=True, + ) + basic_jhelper.get_application.return_value = Mock(units={"cinder/0": unit}) + basic_jhelper.run_cmd_on_unit_payload.side_effect = JujuException("unknown") + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + side_effect=[services, []], + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + basic_jhelper.run_cmd_on_unit_payload.assert_called_once() + + def test_remove_falls_back_when_first_unit_disappears( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + healthy_units = { + name: Mock( + workload_status=Mock(current="active"), + juju_status=Mock(current="idle"), + leader=False, + ) + for name in ("cinder/0", "cinder/1") + } + service = services[0] + basic_jhelper.get_application.return_value = Mock(units=healthy_units) + basic_jhelper.run_cmd_on_unit_payload.side_effect = [ + ExecFailedException("unit disappeared"), + {"return-code": 0}, + ] + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + side_effect=[[service], [service], []], + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + assert [ + call.args[0] + for call in basic_jhelper.run_cmd_on_unit_payload.call_args_list + ] == ["cinder/0", "cinder/1"] + + def test_remove_fails_on_keystoneauth_error( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + unit = Mock( + workload_status=Mock(current="active"), + juju_status=Mock(current="idle"), + leader=True, + ) + basic_jhelper.get_application.return_value = Mock(units={"cinder/0": unit}) + basic_jhelper.run_cmd_on_unit_payload.return_value = {"return-code": 0} + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + side_effect=[ + services, + ConnectFailure("volume endpoint unavailable"), + ], + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + + def test_remove_fails_without_healthy_units( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + unhealthy = Mock( + workload_status=Mock(current="blocked"), + juju_status=Mock(current="idle"), + leader=True, + ) + basic_jhelper.get_application.return_value = Mock(units={"cinder/0": unhealthy}) + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + return_value=services, + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + basic_jhelper.run_cmd_on_unit_payload.assert_not_called() + + def test_remove_fails_after_healthy_candidates_leave_records( + self, + mocker, + basic_jhelper, + basic_deployment, + connection, + services, + step_context, + ): + healthy_units = { + name: Mock( + workload_status=Mock(current="active"), + juju_status=Mock(current="idle"), + leader=False, + ) + for name in ("cinder/0", "cinder/1") + } + basic_jhelper.get_application.return_value = Mock(units=healthy_units) + basic_jhelper.run_cmd_on_unit_payload.return_value = {"return-code": 1} + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", + return_value=services, + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + assert [ + call.args[0] + for call in basic_jhelper.run_cmd_on_unit_payload.call_args_list + ] == ["cinder/0", "cinder/1"] + + def test_remove_skips_when_no_records( + self, mocker, basic_jhelper, basic_deployment, connection, step_context + ): + mocker.patch( + "sunbeam.steps.cinder_volume.get_admin_connection", + return_value=connection, + ) + mocker.patch( + "sunbeam.steps.cinder_volume.get_cinder_volume_services", return_value=[] + ) + step = RemoveCinderVolumeServicesStep( + basic_jhelper, basic_deployment, "cloud-4", "cloud-4.maas" + ) + + assert step.is_skip(step_context).result_type == ResultType.SKIPPED + basic_jhelper.get_application.assert_not_called() diff --git a/sunbeam-python/tests/unit/sunbeam/steps/test_hypervisor.py b/sunbeam-python/tests/unit/sunbeam/steps/test_hypervisor.py index ae06ce217..257d0b192 100644 --- a/sunbeam-python/tests/unit/sunbeam/steps/test_hypervisor.py +++ b/sunbeam-python/tests/unit/sunbeam/steps/test_hypervisor.py @@ -2,9 +2,12 @@ # SPDX-License-Identifier: Apache-2.0 import json -from unittest.mock import Mock, patch +from unittest.mock import Mock, call, patch import pytest +from keystoneauth1.exceptions.catalog import EndpointNotFound +from keystoneauth1.exceptions.connection import ConnectFailure +from openstack.exceptions import SDKException from sunbeam.clusterd.service import NodeNotExistInClusterException from sunbeam.core.common import ResultType @@ -13,6 +16,7 @@ from sunbeam.steps.hypervisor import ( ReapplyHypervisorOptionalIntegrationsStep, ReapplyHypervisorTerraformPlanStep, + RemoveHypervisorReferencesStep, RemoveHypervisorUnitStep, ) @@ -311,6 +315,198 @@ def test_run_timeout( assert result.message == "timed out" +class TestRemoveHypervisorReferencesStep: + @patch("sunbeam.steps.hypervisor.get_admin_connection") + @patch("sunbeam.steps.hypervisor.remove_network_service") + @patch("sunbeam.steps.hypervisor.remove_compute_service") + def test_run_removes_short_and_fqdn_references( + self, + remove_compute_service, + remove_network_service, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + ): + conn = Mock() + conn.compute.services.return_value = [] + conn.network.agents.return_value = [] + get_admin_connection.return_value = conn + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4.maas", + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert remove_compute_service.call_args_list == [ + call("cloud-4", conn), + call("cloud-4.maas", conn), + ] + assert remove_network_service.call_args_list == [ + call("cloud-4", conn), + call("cloud-4.maas", conn), + ] + assert conn.compute.services.call_args_list == [ + call(host="cloud-4"), + call(host="cloud-4.maas"), + ] + assert conn.network.agents.call_args_list == [ + call(host="cloud-4"), + call(host="cloud-4.maas"), + ] + + @patch("sunbeam.steps.hypervisor.HYPERVISOR_REFERENCES_POLL_INTERVAL", 0) + @patch("sunbeam.steps.hypervisor.get_admin_connection") + @patch("sunbeam.steps.hypervisor.remove_network_service") + @patch("sunbeam.steps.hypervisor.remove_compute_service") + def test_run_retries_until_references_are_gone( + self, + remove_compute_service, + remove_network_service, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + ): + conn = Mock() + conn.compute.services.side_effect = [[Mock()], [], [], []] + conn.network.agents.return_value = [] + get_admin_connection.return_value = conn + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4.maas", + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert remove_compute_service.call_count == 4 + assert remove_network_service.call_count == 4 + + @patch("sunbeam.steps.hypervisor.get_admin_connection") + @patch("sunbeam.steps.hypervisor.remove_network_service") + @patch("sunbeam.steps.hypervisor.remove_compute_service") + def test_run_deduplicates_equal_hostnames( + self, + remove_compute_service, + remove_network_service, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + ): + conn = Mock() + conn.compute.services.return_value = [] + conn.network.agents.return_value = [] + get_admin_connection.return_value = conn + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4", + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + remove_compute_service.assert_called_once_with("cloud-4", conn) + remove_network_service.assert_called_once_with("cloud-4", conn) + + @pytest.mark.parametrize( + "error", + [ + SDKException("control plane unavailable"), + EndpointNotFound("control plane unavailable"), + ], + ) + @patch("sunbeam.steps.hypervisor.get_admin_connection") + def test_run_force_ignores_client_error( + self, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + error, + ): + get_admin_connection.side_effect = error + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4.maas", + force=True, + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + get_admin_connection.assert_called_once_with(basic_jhelper, basic_deployment) + + @patch("sunbeam.steps.hypervisor.HYPERVISOR_REFERENCES_POLL_INTERVAL", 0) + @patch("sunbeam.steps.hypervisor.get_admin_connection") + def test_run_retries_keystoneauth_error( + self, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + ): + conn = Mock() + conn.compute.services.return_value = [] + conn.network.agents.return_value = [] + get_admin_connection.side_effect = [ + ConnectFailure("control plane unavailable"), + conn, + ] + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4.maas", + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert get_admin_connection.call_count == 2 + + @patch("sunbeam.steps.hypervisor.HYPERVISOR_REFERENCES_TIMEOUT", 0) + @patch("sunbeam.steps.hypervisor.get_admin_connection") + @patch("sunbeam.steps.hypervisor.remove_network_service") + @patch("sunbeam.steps.hypervisor.remove_compute_service") + def test_run_fails_when_references_persist( + self, + remove_compute_service, + remove_network_service, + get_admin_connection, + basic_jhelper, + basic_deployment, + step_context, + ): + conn = Mock() + conn.compute.services.return_value = [Mock()] + conn.network.agents.return_value = [Mock()] + get_admin_connection.return_value = conn + step = RemoveHypervisorReferencesStep( + basic_jhelper, + basic_deployment, + "cloud-4", + "cloud-4.maas", + ) + + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + assert remove_compute_service.called + assert remove_network_service.called + + class TestReapplyHypervisorTerraformPlanStep: @pytest.fixture def get_network_config_patch(self): diff --git a/sunbeam-python/tests/unit/sunbeam/steps/test_microceph.py b/sunbeam-python/tests/unit/sunbeam/steps/test_microceph.py index 45710175c..15e4efc29 100644 --- a/sunbeam-python/tests/unit/sunbeam/steps/test_microceph.py +++ b/sunbeam-python/tests/unit/sunbeam/steps/test_microceph.py @@ -1,11 +1,81 @@ # SPDX-FileCopyrightText: 2023 - Canonical Ltd # SPDX-License-Identifier: Apache-2.0 -from unittest.mock import Mock +import json +from unittest.mock import MagicMock, Mock -from sunbeam.core.common import ResultType -from sunbeam.core.juju import ActionFailedException -from sunbeam.steps.microceph import ConfigureMicrocephOSDStep, SetCephMgrPoolSizeStep +from sunbeam.clusterd.service import NodeNotExistInClusterException +from sunbeam.core.common import Result, ResultType, run_plan +from sunbeam.core.juju import ( + ActionFailedException, + ApplicationNotFoundException, + ExecFailedException, +) +from sunbeam.steps.microceph import ( + ConfigureMicrocephOSDStep, + RemoveMicrocephOSDsStep, + SetCephMgrPoolSizeStep, +) + + +def _command_result(stdout=""): + return Mock(stdout=stdout) + + +def _configured_disks(disks): + return json.dumps({"ConfiguredDisks": disks}) + + +def _crush_tree(children=None): + nodes = ( + [] + if children is None + else [{"name": "node-1", "type": "host", "children": children}] + ) + return json.dumps({"nodes": nodes}) + + +def _unit(machine, workload="active", agent="idle"): + return Mock( + machine=machine, + workload_status=Mock(current=workload), + juju_status=Mock(current=agent), + ) + + +def _cleanup_step(cclient, jhelper, force=False): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = { + "1": Mock(hostname="node-1"), + "2": Mock(hostname="node-2"), + } + jhelper.get_application.return_value = Mock( + units={"microceph/0": _unit("1"), "microceph/1": _unit("2")} + ) + return RemoveMicrocephOSDsStep( + cclient, + "node-1", + jhelper, + "test-model", + force=force, + ) + + +def _run_cleanup_plan(step): + console = MagicMock() + follow_up = Mock() + follow_up.name = "Follow up" + follow_up.status = "Following up ..." + follow_up.has_prompts.return_value = False + follow_up.is_skip.return_value = Result(ResultType.COMPLETED) + follow_up.run.return_value = Result(ResultType.COMPLETED) + + run_plan([step, follow_up], console) + + return follow_up class TestConfigureMicrocephOSDStep: @@ -88,6 +158,498 @@ def test_run_with_wipe_false(self, cclient, jhelper, step_context): assert result.result_type == ResultType.COMPLETED +class TestRemoveMicrocephOSDsStep: + def test_prefers_healthy_surviving_unit(self, cclient, jhelper, step_context): + step = _cleanup_step(cclient, jhelper) + jhelper.get_machines.return_value["3"] = Mock(hostname="node-3") + jhelper.get_application.return_value = Mock( + units={ + "microceph/0": _unit("1"), + "microceph/1": _unit("2", workload="blocked"), + "microceph/2": _unit("3"), + } + ) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.COMPLETED + assert step.unit == "microceph/2" + + def test_prefers_healthy_target_over_unhealthy_survivor( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.get_application.return_value = Mock( + units={ + "microceph/0": _unit("1"), + "microceph/1": _unit("2", workload="blocked"), + } + ) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.COMPLETED + assert step.unit == "microceph/0" + + def test_read_falls_back_after_exec_failure(self, cclient, jhelper, step_context): + step = _cleanup_step(cclient, jhelper) + exec_error = ExecFailedException("exec failed") + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + exec_error, + _command_result(stdout=_configured_disks([])), + _command_result(stdout=_crush_tree()), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert step.unit == "microceph/0" + assert [ + call.args[0] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] == ["microceph/1", "microceph/0", "microceph/0"] + + def test_remove_does_not_fall_back_after_exec_failure( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + exec_error = ExecFailedException("exec failed") + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result(stdout=_crush_tree([2])), + exec_error, + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + assert [ + call.args[0] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] == ["microceph/1", "microceph/1", "microceph/1"] + + def test_configured_disk_uses_only_cleanup_fields(self): + disks = RemoveMicrocephOSDsStep._parse_configured_disks( + _configured_disks( + [ + { + "osd": 2, + "location": "node-1", + "serial": "disk-2", + } + ] + ) + ) + + assert disks[0]["osd"] == 2 + + def test_removed_target_reaches_follow_up_step(self, cclient, jhelper): + cclient.cluster.get_node_info.side_effect = NodeNotExistInClusterException( + "node removed" + ) + jhelper.get_machines.return_value = {"2": Mock(hostname="node-2")} + jhelper.get_application.return_value = Mock(units={"microceph/1": _unit("2")}) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result(stdout=_configured_disks([])), + _command_result(stdout=json.dumps({"nodes": []})), + ] + step = RemoveMicrocephOSDsStep(cclient, "node-1", jhelper, "test-model") + + follow_up = _run_cleanup_plan(step) + + follow_up.run.assert_called_once() + assert step.unit == "microceph/1" + + def test_forced_last_unit_retry_reaches_follow_up_step(self, cclient, jhelper): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = {"1": Mock(hostname="node-1")} + jhelper.get_application.return_value = Mock(units={}) + step = RemoveMicrocephOSDsStep( + cclient, "node-1", jhelper, "test-model", force=True + ) + + follow_up = _run_cleanup_plan(step) + + follow_up.run.assert_called_once() + jhelper.run_cmd_on_machine_unit_payload.assert_not_called() + + def test_missing_application_is_skipped(self, cclient, jhelper, step_context): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = {"1": Mock(hostname="node-1")} + jhelper.get_application.side_effect = ApplicationNotFoundException( + "application removed" + ) + step = RemoveMicrocephOSDsStep( + cclient, "node-1", jhelper, "test-model", force=True + ) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.SKIPPED + jhelper.run_cmd_on_machine_unit_payload.assert_not_called() + + def test_non_storage_role_still_removes_actual_osd( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + cclient.cluster.get_node_info.return_value["role"] = "compute" + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result(stdout=_crush_tree([2])), + _command_result(), + _command_result(stdout=_configured_disks([])), + _command_result(stdout=_crush_tree()), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + commands = [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] + assert "microceph disk remove osd.2 --timeout 1800" in commands + + def test_run_removes_sorted_db_osds_and_verifies_both_sources( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [ + {"osd": 5, "location": "node-1", "path": "/dev/sdc"}, + {"osd": 2, "location": "node-1", "path": "/dev/sdb"}, + ] + ) + ), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": [5, 2]}]} + ) + ), + _command_result(), + _command_result(), + _command_result(stdout=_configured_disks([])), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": []}]} + ) + ), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + assert [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] == [ + "microceph disk list --json", + "microceph.ceph osd tree --format json", + "microceph disk remove osd.2 --timeout 1800", + "microceph disk remove osd.5 --timeout 1800", + "microceph disk list --json", + "microceph.ceph osd tree --format json", + ] + + def test_crush_only_osd_fails_before_any_removal( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result(stdout=_configured_disks([])), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": [7]}]} + ) + ), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + assert jhelper.run_cmd_on_machine_unit_payload.call_count == 2 + + def test_db_and_crush_mismatch_fails_before_any_removal( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": [2, 7]}]} + ) + ), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + assert jhelper.run_cmd_on_machine_unit_payload.call_count == 2 + + def test_force_keeps_safety_flags_only_on_remove_command( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper, force=True) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": [2]}]} + ) + ), + _command_result(), + _command_result(stdout=_configured_disks([])), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": []}]} + ) + ), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + command = jhelper.run_cmd_on_machine_unit_payload.call_args_list[2].args[2] + assert command == ( + "microceph disk remove osd.2 --timeout 1800 " + "--confirm-failure-domain-downgrade --bypass-safety-checks" + ) + + def test_force_does_not_ignore_exec_failure(self, cclient, jhelper, step_context): + step = _cleanup_step(cclient, jhelper, force=True) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result( + stdout=json.dumps( + {"nodes": [{"name": "node-1", "type": "host", "children": [2]}]} + ) + ), + ExecFailedException("transport failed"), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + + def test_last_target_unit_is_allowed_as_cleanup_fallback( + self, cclient, jhelper, step_context + ): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = {"1": Mock(hostname="node-1")} + jhelper.get_application.return_value = Mock(units={"microceph/0": _unit("1")}) + step = RemoveMicrocephOSDsStep(cclient, "node-1", jhelper, "test-model") + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.unit == "microceph/0" + + def test_no_unit_fails_closed(self, cclient, jhelper, step_context): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = {"1": Mock(hostname="node-1")} + jhelper.get_application.return_value = Mock(units={}) + step = RemoveMicrocephOSDsStep(cclient, "node-1", jhelper, "test-model") + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.FAILED + + def test_unknown_machine_fails_closed(self, cclient, jhelper, step_context): + cclient.cluster.get_node_info.return_value = { + "machineid": "1", + "role": "storage", + } + jhelper.get_machines.return_value = {"1": Mock(hostname="node-1")} + jhelper.get_application.return_value = Mock( + units={"microceph/0": _unit("unknown")} + ) + step = RemoveMicrocephOSDsStep(cclient, "node-1", jhelper, "test-model") + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.FAILED + + def test_unknown_machine_does_not_block_known_surviving_unit( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.get_application.return_value = Mock( + units={ + "microceph/0": _unit("1"), + "microceph/1": _unit("2"), + "microceph/2": _unit("unknown"), + } + ) + + result = step.is_skip(step_context) + + assert result.result_type == ResultType.COMPLETED + assert step.unit == "microceph/1" + + def test_db_only_osd_is_removed_when_crush_host_is_absent( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + ), + _command_result(stdout=_crush_tree()), + _command_result(), + _command_result(stdout=_configured_disks([])), + _command_result(stdout=_crush_tree()), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.COMPLETED + commands = [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] + assert "microceph disk remove osd.2 --timeout 1800" in commands + assert not any( + "purge" in command or "crush remove" in command for command in commands + ) + + def test_malformed_json_fails_before_removal(self, cclient, jhelper, step_context): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.return_value = _command_result( + stdout="not-json" + ) + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + assert jhelper.run_cmd_on_machine_unit_payload.call_count == 1 + + def test_remaining_db_and_crush_osds_fail_after_removal( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + configured = _configured_disks( + [{"osd": 2, "location": "node-1", "path": "/dev/sdb"}] + ) + tree = _crush_tree([2]) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result(stdout=configured), + _command_result(stdout=tree), + _command_result(), + _command_result(stdout=configured), + _command_result(stdout=tree), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + result = step.run(step_context) + + assert result.result_type == ResultType.FAILED + commands = [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] + assert commands.count("microceph disk remove osd.2 --timeout 1800") == 1 + + def test_partial_retry_removes_only_remaining_db_osd( + self, cclient, jhelper, step_context + ): + step = _cleanup_step(cclient, jhelper) + configured = _configured_disks( + [ + {"osd": 2, "location": "node-1", "path": "/dev/sdb"}, + {"osd": 5, "location": "node-1", "path": "/dev/sdc"}, + ] + ) + tree = _crush_tree([2, 5]) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result(stdout=configured), + _command_result(stdout=tree), + _command_result(), + ExecFailedException("busy"), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.FAILED + + retry = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.reset_mock() + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result( + stdout=_configured_disks( + [{"osd": 5, "location": "node-1", "path": "/dev/sdc"}] + ) + ), + _command_result(stdout=_crush_tree([5])), + _command_result(), + _command_result(stdout=_configured_disks([])), + _command_result(stdout=_crush_tree()), + ] + + assert retry.is_skip(step_context).result_type == ResultType.COMPLETED + assert retry.run(step_context).result_type == ResultType.COMPLETED + commands = [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] + assert commands.count("microceph disk remove osd.5 --timeout 1800") == 1 + assert "microceph disk remove osd.2 --timeout 1800" not in commands + + def test_clean_state_is_idempotent(self, cclient, jhelper, step_context): + step = _cleanup_step(cclient, jhelper) + jhelper.run_cmd_on_machine_unit_payload.side_effect = [ + _command_result(stdout=_configured_disks([])), + _command_result(stdout=json.dumps({"nodes": []})), + ] + + assert step.is_skip(step_context).result_type == ResultType.COMPLETED + assert step.run(step_context).result_type == ResultType.COMPLETED + assert [ + call.args[2] + for call in jhelper.run_cmd_on_machine_unit_payload.call_args_list + ] == [ + "microceph disk list --json", + "microceph.ceph osd tree --format json", + ] + + class TestSetCephMgrPoolSizeStep: def test_is_skip(self, cclient, jhelper, step_context): cclient.cluster.list_nodes_by_role.return_value = []