Filter hypervisors.

- do the nova-compute instance-count to find empty hypervisors
- integrate the force flag to include non-empty hypervisors

cou/apps/base.py

@@ -49,7 +49,7 @@
 DEFAULT_WAITING_TIMEOUT = 5 * 60  # 5 min
 
 
-@dataclass
+@dataclass(frozen=True)
 class ApplicationUnit:
     """Representation of a single unit of application."""
 
@@ -58,6 +58,14 @@ class ApplicationUnit:
     machine: Machine
     workload_version: str = ""
 
+    def __repr__(self) -> str:
+        """Representation of the application unit.
+
+        :return: Representation of the application unit
+        :rtype: str
+        """
+        return f"Unit[{self.name}]-Machine[{self.machine.machine_id}]"
+
 
 @dataclass
 class OpenStackApplication:
@@ -322,6 +330,7 @@ def current_os_release(self) -> OpenStackRelease:
             f"'{openstack_release.codename}': {units}"
             for openstack_release, units in os_versions.items()
         ]
+
         raise MismatchedOpenStackVersions(
             f"Units of application {self.name} are running mismatched OpenStack versions: "
             f"{', '.join(mismatched_repr)}. This is not currently handled."

cou/steps/plan.py

@@ -30,7 +30,7 @@
     OpenStackAuxiliarySubordinateApplication,
     OvnSubordinateApplication,
 )
-from cou.apps.base import OpenStackApplication
+from cou.apps.base import ApplicationUnit, OpenStackApplication
 from cou.apps.channel_based import OpenStackChannelBasedApplication  # noqa: F401
 from cou.apps.core import Keystone, Octavia  # noqa: F401
 from cou.apps.subordinate import (  # noqa: F401
@@ -49,6 +49,7 @@
 from cou.steps import PreUpgradeStep, UpgradePlan
 from cou.steps.analyze import Analysis
 from cou.steps.backup import backup
+from cou.utils.app_utils import _get_empty_hypervisors
 from cou.utils.juju_utils import DEFAULT_TIMEOUT
 from cou.utils.openstack import LTS_TO_OS_RELEASE, OpenStackRelease
 
@@ -345,6 +346,8 @@ async def generate_plan(analysis_result: Analysis, args: CLIargs) -> UpgradePlan
     :rtype: UpgradePlan
     """
     pre_plan_sanity_checks(args, analysis_result)
+    hypervisors = await filter_hypervisors_units(args, analysis_result)
+    logger.info("Hypervisors selected: %s", hypervisors)
     target = determine_upgrade_target(analysis_result)
 
     plan = UpgradePlan(
@@ -393,6 +396,77 @@ async def generate_plan(analysis_result: Analysis, args: CLIargs) -> UpgradePlan
     return plan
 
 
+async def filter_hypervisors_units(
+    args: CLIargs, analysis_result: Analysis
+) -> set[ApplicationUnit]:
+    """Filter the hypervisors to generate plan and upgrade.
+
+    :param args: CLI arguments
+    :type args: CLIargs
+    :param analysis_result: Analysis result
+    :type analysis_result: Analysis
+    :return: hypervisors filtered to generate plan and upgrade.
+    :rtype: list[ApplicationUnit]
+    """
+    hypervisors_units = await _get_hypervisors_possible_to_upgrade(args.force, analysis_result)
+
+    if cli_machines := args.machines:
+        return {
+            unit
+            for unit in hypervisors_units
+            if unit.machine.machine_id in parametrize_cli_inputs(cli_machines)
+        }
+
+    if cli_hostnames := args.hostnames:
+        return {
+            unit
+            for unit in hypervisors_units
+            if unit.machine.hostname in parametrize_cli_inputs(cli_hostnames)
+        }
+
+    if cli_azs := args.availability_zones:
+        return {
+            unit
+            for unit in hypervisors_units
+            if unit.machine.az in parametrize_cli_inputs(cli_azs)
+        }
+
+    return hypervisors_units
+
+
+async def _get_hypervisors_possible_to_upgrade(
+    cli_force: bool, analysis_result: Analysis
+) -> set[ApplicationUnit]:
+    """Get the hypervisors that are possible to upgrade.
+
+    :param cli_force: If force is used, it gets all hypervisors, otherwise just the empty ones
+    :type cli_force: bool
+    :param analysis_result: Analysis result
+    :type analysis_result: Analysis
+    :return: Set of nova-compute units to upgrade
+    :rtype: set[ApplicationUnit]
+    """
+    hypervisors_units = _get_hypervisors_units(analysis_result.apps_data_plane)
+
+    if cli_force:
+        return hypervisors_units
+
+    hypervisors_units_name = [unit.name for unit in hypervisors_units]
+    empty_hypervisors = await _get_empty_hypervisors(hypervisors_units_name, analysis_result.model)
+    return {unit for unit in hypervisors_units if unit.name in empty_hypervisors}
+
+
+def _get_hypervisors_units(data_plane_apps: list[OpenStackApplication]) -> set[ApplicationUnit]:
+    """Get all the hypervisors (nova-compute) units.
+
+    :param data_plane_apps: list of the data-plane applications
+    :type data_plane_apps: list[OpenStackApplication]
+    :return: set of all nova-compute units in the model.
+    :rtype: set[ApplicationUnit]
+    """
+    return {unit for app in data_plane_apps for unit in app.units if app.charm == "nova-compute"}
+
+
 async def create_upgrade_group(
     apps: list[OpenStackApplication],
     target: OpenStackRelease,

cou/utils/app_utils.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 """Application utilities."""
+import asyncio
 import json
 import logging
 from collections.abc import Iterable
@@ -50,6 +51,22 @@ async def upgrade_packages(
         await model.run_on_unit(unit_name=unit, command=command, timeout=600)
 
 
+async def _get_empty_hypervisors(units: list[str], model: COUModel) -> set[str]:
+    """Get the empty hypervisors in the model.
+
+    :param units: all nova-compute units.
+    :type units: list[str]
+    :param model: COUModel object
+    :type model: COUModel
+    :return: Set with just the empty hypervisors
+    :rtype: set[str]
+    """
+    tasks = [get_instance_count(unit, model) for unit in units]
+    instances = await asyncio.gather(*tasks)
+    units_instances = zip(units, instances)
+    return {unit for unit, instances in units_instances if instances == 0}
+
+
 async def get_instance_count(unit: str, model: COUModel) -> int:
     """Get instance count on a nova-compute unit.