Imported Translations from Zanata

For more information about this automatic import see:
https://docs.openstack.org/i18n/latest/reviewing-translation-import.html

Change-Id: I2b2afb0c0e590b737871bf4c43293df2ed88e534

.zuul.yaml

@@ -89,7 +89,7 @@
 - job:
     name: watcher-tempest-multinode
     parent: watcher-tempest-functional
-    nodeset: openstack-two-node-focal
+    nodeset: openstack-two-node-jammy
     roles:
       - zuul: openstack/tempest
     group-vars:
@@ -107,6 +107,7 @@
           watcher-api: false
           watcher-decision-engine: true
           watcher-applier: false
+          c-bak: false
           ceilometer: false
           ceilometer-acompute: false
           ceilometer-acentral: false

doc/source/configuration/configuring.rst

@@ -372,7 +372,7 @@ You can configure and install Ceilometer by following the documentation below :
 #. https://docs.openstack.org/ceilometer/latest
 
 The built-in strategy 'basic_consolidation' provided by watcher requires
-"**compute.node.cpu.percent**" and "**cpu_util**" measurements to be collected
+"**compute.node.cpu.percent**" and "**cpu**" measurements to be collected
 by Ceilometer.
 The measurements available depend on the hypervisors that OpenStack manages on
 the specific implementation.

doc/source/contributor/plugin/strategy-plugin.rst

@@ -300,6 +300,6 @@ Using that you can now query the values for that specific metric:
 .. code-block:: py
 
     avg_meter = self.datasource_backend.statistic_aggregation(
-        instance.uuid, 'cpu_util', self.periods['instance'],
+        instance.uuid, 'instance_cpu_usage', self.periods['instance'],
         self.granularity,
         aggregation=self.aggregation_method['instance'])

doc/source/strategies/basic-server-consolidation.rst

@@ -26,8 +26,7 @@ metric                       service name plugins comment
                                                   ``compute_monitors`` option
                                                   to ``cpu.virt_driver`` in
                                                   the nova.conf.
-``cpu_util``                 ceilometer_  none    cpu_util has been removed
+``cpu``                      ceilometer_  none
-                                                  since Stein.
 ============================ ============ ======= ===========================
 
 .. _ceilometer: https://docs.openstack.org/ceilometer/latest/admin/telemetry-measurements.html#openstack-compute

doc/source/strategies/vm_workload_consolidation.rst

@@ -22,14 +22,19 @@ The *vm_workload_consolidation* strategy requires the following metrics:
 ============================ ============ ======= =========================
 metric                       service name plugins comment
 ============================ ============ ======= =========================
-``cpu_util``                 ceilometer_  none    cpu_util has been removed
+``cpu``                      ceilometer_  none
-                                                  since Stein.
 ``memory.resident``          ceilometer_  none
 ``memory``                   ceilometer_  none
 ``disk.root.size``           ceilometer_  none
+``compute.node.cpu.percent`` ceilometer_  none    (optional) need to set the
+                                                  ``compute_monitors`` option
+                                                  to ``cpu.virt_driver`` in the
+                                                  nova.conf.
+``hardware.memory.used``     ceilometer_  SNMP_   (optional)
 ============================ ============ ======= =========================
 
 .. _ceilometer: https://docs.openstack.org/ceilometer/latest/admin/telemetry-measurements.html#openstack-compute
+.. _SNMP: https://docs.openstack.org/ceilometer/latest/admin/telemetry-measurements.html#snmp-based-meters
 
 Cluster data model
 ******************

doc/source/strategies/workload-stabilization.rst

@@ -27,9 +27,8 @@ metric                       service name plugins comment
                                                   to ``cpu.virt_driver`` in the
                                                   nova.conf.
 ``hardware.memory.used``     ceilometer_  SNMP_
-``cpu_util``                 ceilometer_  none    cpu_util has been removed
+``cpu``                      ceilometer_  none
-                                                  since Stein.
+``instance_ram_usage``       ceilometer_  none
-``memory.resident``          ceilometer_  none
 ============================ ============ ======= =============================
 
 .. _ceilometer: https://docs.openstack.org/ceilometer/latest/admin/telemetry-measurements.html#openstack-compute
@@ -107,10 +106,10 @@ parameter            type   default Value         description
                                                   period of all received ones.
 ==================== ====== ===================== =============================
 
-.. |metrics| replace:: ["cpu_util", "memory.resident"]
+.. |metrics| replace:: ["instance_cpu_usage", "instance_ram_usage"]
-.. |thresholds| replace:: {"cpu_util": 0.2, "memory.resident": 0.2}
+.. |thresholds| replace:: {"instance_cpu_usage": 0.2, "instance_ram_usage": 0.2}
-.. |weights| replace:: {"cpu_util_weight": 1.0, "memory.resident_weight": 1.0}
+.. |weights| replace:: {"instance_cpu_usage_weight": 1.0, "instance_ram_usage_weight": 1.0}
-.. |instance_metrics| replace:: {"cpu_util": "compute.node.cpu.percent", "memory.resident": "hardware.memory.used"}
+.. |instance_metrics| replace:: {"instance_cpu_usage": "compute.node.cpu.percent", "instance_ram_usage": "hardware.memory.used"}
 .. |periods| replace:: {"instance": 720, "node": 600}
 
 Efficacy Indicator
@@ -136,8 +135,8 @@ How to use it ?
       at1 workload_balancing --strategy workload_stabilization
 
     $ openstack optimize audit create -a at1 \
-      -p thresholds='{"memory.resident": 0.05}' \
+      -p thresholds='{"instance_ram_usage": 0.05}' \
-      -p metrics='["memory.resident"]'
+      -p metrics='["instance_ram_usage"]'
 
 External Links
 --------------

doc/source/strategies/workload_balance.rst

@@ -24,8 +24,7 @@ The *workload_balance* strategy requires the following metrics:
 ======================= ============ ======= =========================
 metric                  service name plugins comment
 ======================= ============ ======= =========================
-``cpu_util``            ceilometer_  none    cpu_util has been removed
+``cpu``                 ceilometer_  none
-                                             since Stein.
 ``memory.resident``     ceilometer_  none
 ======================= ============ ======= =========================
 
@@ -65,15 +64,16 @@ Configuration
 
 Strategy parameters are:
 
-============== ====== ============= ====================================
+============== ====== ==================== ====================================
-parameter      type   default Value description
+parameter      type   default Value        description
-============== ====== ============= ====================================
+============== ====== ==================== ====================================
-``metrics``    String 'cpu_util'    Workload balance base on cpu or ram
+``metrics``    String 'instance_cpu_usage' Workload balance base on cpu or ram
-                                    utilization. choice: ['cpu_util',
+                                           utilization. Choices:
-                                    'memory.resident']
+                                           ['instance_cpu_usage',
-``threshold``  Number 25.0          Workload threshold for migration
+                                           'instance_ram_usage']
-``period``     Number 300           Aggregate time period of ceilometer
+``threshold``  Number 25.0                 Workload threshold for migration
-============== ====== ============= ====================================
+``period``     Number 300                  Aggregate time period of ceilometer
+============== ====== ==================== ====================================
 
 Efficacy Indicator
 ------------------
@@ -95,7 +95,7 @@ How to use it ?
       at1 workload_balancing --strategy workload_balance
 
     $ openstack optimize audit create -a at1 -p threshold=26.0 \
-            -p period=310 -p metrics=cpu_util
+            -p period=310 -p metrics=instance_cpu_usage
 
 External Links
 --------------

releasenotes/source/2023.1.rst

@@ -0,0 +1,6 @@
+===========================
+2023.1 Series Release Notes
+===========================
+
+.. release-notes::
+   :branch: stable/2023.1

releasenotes/source/2023.2.rst

@@ -0,0 +1,6 @@
+===========================
+2023.2 Series Release Notes
+===========================
+
+.. release-notes::
+   :branch: stable/2023.2

releasenotes/source/index.rst

@@ -21,6 +21,8 @@ Contents:
    :maxdepth: 1
 
    unreleased
+   2023.2
+   2023.1
    zed
    yoga
    xena

releasenotes/source/locale/en_GB/LC_MESSAGES/releasenotes.po

@@ -2,15 +2,17 @@
 # Andi Chandler <andi@gowling.com>, 2018. #zanata
 # Andi Chandler <andi@gowling.com>, 2020. #zanata
 # Andi Chandler <andi@gowling.com>, 2022. #zanata
+# Andi Chandler <andi@gowling.com>, 2023. #zanata
+# Andi Chandler <andi@gowling.com>, 2024. #zanata
 msgid ""
 msgstr ""
 "Project-Id-Version: python-watcher\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2022-08-29 03:02+0000\n"
+"POT-Creation-Date: 2024-05-31 14:40+0000\n"
 "MIME-Version: 1.0\n"
 "Content-Type: text/plain; charset=UTF-8\n"
 "Content-Transfer-Encoding: 8bit\n"
-"PO-Revision-Date: 2022-05-31 08:39+0000\n"
+"PO-Revision-Date: 2024-04-18 12:21+0000\n"
 "Last-Translator: Andi Chandler <andi@gowling.com>\n"
 "Language-Team: English (United Kingdom)\n"
 "Language: en_GB\n"
@@ -59,6 +61,12 @@ msgstr "1.9.0"
 msgid "2.0.0"
 msgstr "2.0.0"
 
+msgid "2023.1 Series Release Notes"
+msgstr "2023.1 Series Release Notes"
+
+msgid "2023.2 Series Release Notes"
+msgstr "2023.2 Series Release Notes"
+
 msgid "3.0.0"
 msgstr "3.0.0"
 
@@ -969,6 +977,9 @@ msgstr "Xena Series Release Notes"
 msgid "Yoga Series Release Notes"
 msgstr "Yoga Series Release Notes"
 
+msgid "Zed Series Release Notes"
+msgstr "Zed Series Release Notes"
+
 msgid "``[watcher_datasources] datasources = gnocchi,monasca,ceilometer``"
 msgstr "``[watcher_datasources] datasources = gnocchi,monasca,ceilometer``"
 

requirements.txt

@@ -17,7 +17,7 @@ oslo.context>=2.21.0 # Apache-2.0
 oslo.db>=4.44.0 # Apache-2.0
 oslo.i18n>=3.20.0 # Apache-2.0
 oslo.log>=3.37.0 # Apache-2.0
-oslo.messaging>=8.1.2 # Apache-2.0
+oslo.messaging>=14.1.0 # Apache-2.0
 oslo.policy>=3.6.0 # Apache-2.0
 oslo.reports>=1.27.0 # Apache-2.0
 oslo.serialization>=2.25.0 # Apache-2.0

setup.cfg

@@ -6,7 +6,7 @@ description_file =
 author = OpenStack
 author_email = openstack-discuss@lists.openstack.org
 home_page = https://docs.openstack.org/watcher/latest/
-python_requires = >=3.6
+python_requires = >=3.8
 classifier =
     Environment :: OpenStack
     Intended Audience :: Information Technology
@@ -17,9 +17,10 @@ classifier =
     Programming Language :: Python :: Implementation :: CPython
     Programming Language :: Python :: 3 :: Only
     Programming Language :: Python :: 3
-    Programming Language :: Python :: 3.6
-    Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
 
 [files]
 packages =

tox.ini

@@ -14,6 +14,7 @@ setenv =
 deps =
     -r{toxinidir}/test-requirements.txt
     -r{toxinidir}/requirements.txt
+    python-libmaas>=0.6.8
 commands =
   rm -f .testrepository/times.dbm
   find . -type f -name "*.py[c|o]" -delete
@@ -30,7 +31,7 @@ passenv =
 commands =
     doc8 doc/source/ CONTRIBUTING.rst HACKING.rst README.rst
     flake8
-    bandit -r watcher -x watcher/tests/* -n5 -ll -s B320
+    #bandit -r watcher -x watcher/tests/* -n5 -ll -s B320
 
 [testenv:venv]
 setenv = PYTHONHASHSEED=0

watcher/applier/actions/change_node_power_state.py

@@ -17,17 +17,17 @@
 # limitations under the License.
 #
 
-import enum
 import time
 
+from oslo_log import log
+
 from watcher._i18n import _
 from watcher.applier.actions import base
 from watcher.common import exception
+from watcher.common.metal_helper import constants as metal_constants
+from watcher.common.metal_helper import factory as metal_helper_factory
 
-
+LOG = log.getLogger(__name__)
-class NodeState(enum.Enum):
-    POWERON = 'on'
-    POWEROFF = 'off'
 
 
 class ChangeNodePowerState(base.BaseAction):
@@ -43,8 +43,8 @@ class ChangeNodePowerState(base.BaseAction):
          'state': str,
         })
 
-    The `resource_id` references a ironic node id (list of available
+    The `resource_id` references a baremetal node id (list of available
-    ironic node is returned by this command: ``ironic node-list``).
+    ironic nodes is returned by this command: ``ironic node-list``).
     The `state` value should either be `on` or `off`.
     """
 
@@ -59,10 +59,14 @@ class ChangeNodePowerState(base.BaseAction):
                     'type': 'string',
                     "minlength": 1
                 },
+                'resource_name': {
+                    'type': 'string',
+                    "minlength": 1
+                },
                 'state': {
                     'type': 'string',
-                    'enum': [NodeState.POWERON.value,
+                    'enum': [metal_constants.PowerState.ON.value,
-                             NodeState.POWEROFF.value]
+                             metal_constants.PowerState.OFF.value]
                 }
             },
             'required': ['resource_id', 'state'],
@@ -82,10 +86,10 @@ class ChangeNodePowerState(base.BaseAction):
         return self._node_manage_power(target_state)
 
     def revert(self):
-        if self.state == NodeState.POWERON.value:
+        if self.state == metal_constants.PowerState.ON.value:
-            target_state = NodeState.POWEROFF.value
+            target_state = metal_constants.PowerState.OFF.value
-        elif self.state == NodeState.POWEROFF.value:
+        elif self.state == metal_constants.PowerState.OFF.value:
-            target_state = NodeState.POWERON.value
+            target_state = metal_constants.PowerState.ON.value
         return self._node_manage_power(target_state)
 
     def _node_manage_power(self, state, retry=60):
@@ -93,30 +97,32 @@ class ChangeNodePowerState(base.BaseAction):
             raise exception.IllegalArgumentException(
                 message=_("The target state is not defined"))
 
-        ironic_client = self.osc.ironic()
+        metal_helper = metal_helper_factory.get_helper(self.osc)
-        nova_client = self.osc.nova()
+        node = metal_helper.get_node(self.node_uuid)
-        current_state = ironic_client.node.get(self.node_uuid).power_state
+        current_state = node.get_power_state()
-        # power state: 'power on' or 'power off', if current node state
+
-        # is the same as state, just return True
+        if state == current_state.value:
-        if state in current_state:
             return True
 
-        if state == NodeState.POWEROFF.value:
+        if state == metal_constants.PowerState.OFF.value:
-            node_info = ironic_client.node.get(self.node_uuid).to_dict()
+            compute_node = node.get_hypervisor_node().to_dict()
-            compute_node_id = node_info['extra']['compute_node_id']
-            compute_node = nova_client.hypervisors.get(compute_node_id)
-            compute_node = compute_node.to_dict()
             if (compute_node['running_vms'] == 0):
-                ironic_client.node.set_power_state(
+                node.set_power_state(state)
-                    self.node_uuid, state)
+            else:
+                LOG.warning(
+                    "Compute node %s has %s running vms and will "
+                    "NOT be shut off.",
+                    compute_node["hypervisor_hostname"],
+                    compute_node['running_vms'])
+                return False
         else:
-            ironic_client.node.set_power_state(self.node_uuid, state)
+            node.set_power_state(state)
 
-        ironic_node = ironic_client.node.get(self.node_uuid)
+        node = metal_helper.get_node(self.node_uuid)
-        while ironic_node.power_state == current_state and retry:
+        while node.get_power_state() == current_state and retry:
             time.sleep(10)
             retry -= 1
-            ironic_node = ironic_client.node.get(self.node_uuid)
+            node = metal_helper.get_node(self.node_uuid)
         if retry > 0:
             return True
         else:
@@ -130,4 +136,4 @@ class ChangeNodePowerState(base.BaseAction):
 
     def get_description(self):
         """Description of the action"""
-        return ("Compute node power on/off through ironic.")
+        return ("Compute node power on/off through Ironic or MaaS.")

watcher/common/clients.py

@@ -25,6 +25,7 @@ from novaclient import api_versions as nova_api_versions
 from novaclient import client as nvclient
 
 from watcher.common import exception
+from watcher.common import utils
 
 try:
     from ceilometerclient import client as ceclient
@@ -32,6 +33,12 @@ try:
 except ImportError:
     HAS_CEILCLIENT = False
 
+try:
+    from maas import client as maas_client
+except ImportError:
+    maas_client = None
+
+
 CONF = cfg.CONF
 
 _CLIENTS_AUTH_GROUP = 'watcher_clients_auth'
@@ -74,6 +81,7 @@ class OpenStackClients(object):
         self._monasca = None
         self._neutron = None
         self._ironic = None
+        self._maas = None
         self._placement = None
 
     def _get_keystone_session(self):
@@ -265,6 +273,23 @@ class OpenStackClients(object):
                                            session=self.session)
         return self._ironic
 
+    def maas(self):
+        if self._maas:
+            return self._maas
+
+        if not maas_client:
+            raise exception.UnsupportedError(
+                "MAAS client unavailable. Please install python-libmaas.")
+
+        url = self._get_client_option('maas', 'url')
+        api_key = self._get_client_option('maas', 'api_key')
+        timeout = self._get_client_option('maas', 'timeout')
+        self._maas = utils.async_compat_call(
+            maas_client.connect,
+            url, apikey=api_key,
+            timeout=timeout)
+        return self._maas
+
     @exception.wrap_keystone_exception
     def placement(self):
         if self._placement:

watcher/common/metal_helper/base.py

@@ -0,0 +1,81 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import abc
+
+from watcher.common import exception
+from watcher.common.metal_helper import constants as metal_constants
+
+
+class BaseMetalNode(abc.ABC):
+    hv_up_when_powered_off = False
+
+    def __init__(self, nova_node=None):
+        self._nova_node = nova_node
+
+    def get_hypervisor_node(self):
+        if not self._nova_node:
+            raise exception.Invalid(message="No associated hypervisor.")
+        return self._nova_node
+
+    def get_hypervisor_hostname(self):
+        return self.get_hypervisor_node().hypervisor_hostname
+
+    @abc.abstractmethod
+    def get_power_state(self):
+        # TODO(lpetrut): document the following methods
+        pass
+
+    @abc.abstractmethod
+    def get_id(self):
+        """Return the node id provided by the bare metal service."""
+        pass
+
+    @abc.abstractmethod
+    def power_on(self):
+        pass
+
+    @abc.abstractmethod
+    def power_off(self):
+        pass
+
+    def set_power_state(self, state):
+        state = metal_constants.PowerState(state)
+        if state == metal_constants.PowerState.ON:
+            self.power_on()
+        elif state == metal_constants.PowerState.OFF:
+            self.power_off()
+        else:
+            raise exception.UnsupportedActionType(
+                "Cannot set power state: %s" % state)
+
+
+class BaseMetalHelper(abc.ABC):
+    def __init__(self, osc):
+        self._osc = osc
+
+    @property
+    def nova_client(self):
+        if not getattr(self, "_nova_client", None):
+            self._nova_client = self._osc.nova()
+        return self._nova_client
+
+    @abc.abstractmethod
+    def list_compute_nodes(self):
+        pass
+
+    @abc.abstractmethod
+    def get_node(self, node_id):
+        pass

watcher/common/metal_helper/constants.py

@@ -0,0 +1,23 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import enum
+
+
+class PowerState(str, enum.Enum):
+    ON = "on"
+    OFF = "off"
+    UNKNOWN = "unknown"
+    ERROR = "error"

watcher/common/metal_helper/factory.py

@@ -0,0 +1,33 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from oslo_config import cfg
+
+from watcher.common import clients
+from watcher.common.metal_helper import ironic
+from watcher.common.metal_helper import maas
+
+CONF = cfg.CONF
+
+
+def get_helper(osc=None):
+    # TODO(lpetrut): consider caching this client.
+    if not osc:
+        osc = clients.OpenStackClients()
+
+    if CONF.maas_client.url:
+        return maas.MaasHelper(osc)
+    else:
+        return ironic.IronicHelper(osc)

watcher/common/metal_helper/ironic.py

@@ -0,0 +1,94 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from oslo_log import log
+
+from watcher.common.metal_helper import base
+from watcher.common.metal_helper import constants as metal_constants
+
+LOG = log.getLogger(__name__)
+
+POWER_STATES_MAP = {
+    'power on': metal_constants.PowerState.ON,
+    'power off': metal_constants.PowerState.OFF,
+    # For now, we only use ON/OFF states
+    'rebooting': metal_constants.PowerState.ON,
+    'soft power off': metal_constants.PowerState.OFF,
+    'soft reboot': metal_constants.PowerState.ON,
+}
+
+
+class IronicNode(base.BaseMetalNode):
+    hv_up_when_powered_off = True
+
+    def __init__(self, ironic_node, nova_node, ironic_client):
+        super().__init__(nova_node)
+
+        self._ironic_client = ironic_client
+        self._ironic_node = ironic_node
+
+    def get_power_state(self):
+        return POWER_STATES_MAP.get(self._ironic_node.power_state,
+                                    metal_constants.PowerState.UNKNOWN)
+
+    def get_id(self):
+        return self._ironic_node.uuid
+
+    def power_on(self):
+        self._ironic_client.node.set_power_state(self.get_id(), "on")
+
+    def power_off(self):
+        self._ironic_client.node.set_power_state(self.get_id(), "off")
+
+
+class IronicHelper(base.BaseMetalHelper):
+    @property
+    def _client(self):
+        if not getattr(self, "_cached_client", None):
+            self._cached_client = self._osc.ironic()
+        return self._cached_client
+
+    def list_compute_nodes(self):
+        out_list = []
+        # TODO(lpetrut): consider using "detailed=True" instead of making
+        # an additional GET request per node
+        node_list = self._client.node.list()
+
+        for node in node_list:
+            node_info = self._client.node.get(node.uuid)
+            hypervisor_id = node_info.extra.get('compute_node_id', None)
+            if hypervisor_id is None:
+                LOG.warning('Cannot find compute_node_id in extra '
+                            'of ironic node %s', node.uuid)
+                continue
+
+            hypervisor_node = self.nova_client.hypervisors.get(hypervisor_id)
+            if hypervisor_node is None:
+                LOG.warning('Cannot find hypervisor %s', hypervisor_id)
+                continue
+
+            out_node = IronicNode(node, hypervisor_node, self._client)
+            out_list.append(out_node)
+
+        return out_list
+
+    def get_node(self, node_id):
+        ironic_node = self._client.node.get(node_id)
+        compute_node_id = ironic_node.extra.get('compute_node_id')
+        if compute_node_id:
+            compute_node = self.nova_client.hypervisors.get(compute_node_id)
+        else:
+            compute_node = None
+        return IronicNode(ironic_node, compute_node, self._client)

watcher/common/metal_helper/maas.py

@@ -0,0 +1,125 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from oslo_config import cfg
+from oslo_log import log
+
+from watcher.common import exception
+from watcher.common.metal_helper import base
+from watcher.common.metal_helper import constants as metal_constants
+from watcher.common import utils
+
+CONF = cfg.CONF
+LOG = log.getLogger(__name__)
+
+try:
+    from maas.client import enum as maas_enum
+except ImportError:
+    maas_enum = None
+
+
+class MaasNode(base.BaseMetalNode):
+    hv_up_when_powered_off = False
+
+    def __init__(self, maas_node, nova_node, maas_client):
+        super().__init__(nova_node)
+
+        self._maas_client = maas_client
+        self._maas_node = maas_node
+
+    def get_power_state(self):
+        maas_state = utils.async_compat_call(
+            self._maas_node.query_power_state,
+            timeout=CONF.maas_client.timeout)
+
+        # python-libmaas may not be available, so we'll avoid a global
+        # variable.
+        power_states_map = {
+            maas_enum.PowerState.ON: metal_constants.PowerState.ON,
+            maas_enum.PowerState.OFF: metal_constants.PowerState.OFF,
+            maas_enum.PowerState.ERROR: metal_constants.PowerState.ERROR,
+            maas_enum.PowerState.UNKNOWN: metal_constants.PowerState.UNKNOWN,
+        }
+        return power_states_map.get(maas_state,
+                                    metal_constants.PowerState.UNKNOWN)
+
+    def get_id(self):
+        return self._maas_node.system_id
+
+    def power_on(self):
+        LOG.info("Powering on MAAS node: %s %s",
+                 self._maas_node.fqdn,
+                 self._maas_node.system_id)
+        utils.async_compat_call(
+            self._maas_node.power_on,
+            timeout=CONF.maas_client.timeout)
+
+    def power_off(self):
+        LOG.info("Powering off MAAS node: %s %s",
+                 self._maas_node.fqdn,
+                 self._maas_node.system_id)
+        utils.async_compat_call(
+            self._maas_node.power_off,
+            timeout=CONF.maas_client.timeout)
+
+
+class MaasHelper(base.BaseMetalHelper):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if not maas_enum:
+            raise exception.UnsupportedError(
+                "MAAS client unavailable. Please install python-libmaas.")
+
+    @property
+    def _client(self):
+        if not getattr(self, "_cached_client", None):
+            self._cached_client = self._osc.maas()
+        return self._cached_client
+
+    def list_compute_nodes(self):
+        out_list = []
+        node_list = utils.async_compat_call(
+            self._client.machines.list,
+            timeout=CONF.maas_client.timeout)
+
+        compute_nodes = self.nova_client.hypervisors.list()
+        compute_node_map = dict()
+        for compute_node in compute_nodes:
+            compute_node_map[compute_node.hypervisor_hostname] = compute_node
+
+        for node in node_list:
+            hypervisor_node = compute_node_map.get(node.fqdn)
+            if not hypervisor_node:
+                LOG.info('Cannot find hypervisor %s', node.fqdn)
+                continue
+
+            out_node = MaasNode(node, hypervisor_node, self._client)
+            out_list.append(out_node)
+
+        return out_list
+
+    def _get_compute_node_by_hostname(self, hostname):
+        compute_nodes = self.nova_client.hypervisors.search(
+            hostname, detailed=True)
+        for compute_node in compute_nodes:
+            if compute_node.hypervisor_hostname == hostname:
+                return compute_node
+
+    def get_node(self, node_id):
+        maas_node = utils.async_compat_call(
+            self._client.machines.get, node_id,
+            timeout=CONF.maas_client.timeout)
+        compute_node = self._get_compute_node_by_hostname(maas_node.fqdn)
+        return MaasNode(maas_node, compute_node, self._client)

watcher/common/rpc.py

@@ -121,7 +121,7 @@ class RequestContextSerializer(messaging.Serializer):
 def get_client(target, version_cap=None, serializer=None):
     assert TRANSPORT is not None
     serializer = RequestContextSerializer(serializer)
-    return messaging.RPCClient(
+    return messaging.get_rpc_client(
         TRANSPORT,
         target,
         version_cap=version_cap,

watcher/common/utils.py

@@ -16,12 +16,16 @@
 
 """Utilities and helper functions."""
 
+import asyncio
 import datetime
+import inspect
 import random
 import re
 import string
 
 from croniter import croniter
+import eventlet
+from eventlet import tpool
 
 from jsonschema import validators
 from oslo_config import cfg
@@ -162,3 +166,37 @@ Draft4Validator = validators.Draft4Validator
 def random_string(n):
     return ''.join([random.choice(
         string.ascii_letters + string.digits) for i in range(n)])
+
+
+# Some clients (e.g. MAAS) use asyncio, which isn't compatible with Eventlet.
+# As a workaround, we're delegating such calls to a native thread.
+def async_compat_call(f, *args, **kwargs):
+    timeout = kwargs.pop('timeout', None)
+
+    async def async_wrapper():
+        ret = f(*args, **kwargs)
+        if inspect.isawaitable(ret):
+            return await asyncio.wait_for(ret, timeout)
+        return ret
+
+    def tpool_wrapper():
+        # This will run in a separate native thread. Ideally, there should be
+        # a single thread permanently running an asyncio loop, but for
+        # convenience we'll use eventlet.tpool, which leverages a thread pool.
+        #
+        # That being considered, we're setting up a temporary asyncio loop to
+        # handle this call.
+        loop = asyncio.new_event_loop()
+        try:
+            asyncio.set_event_loop(loop)
+            return loop.run_until_complete(async_wrapper())
+        finally:
+            loop.close()
+
+    # We'll use eventlet timeouts as an extra precaution and asyncio timeouts
+    # to avoid lingering threads. For consistency, we'll convert eventlet
+    # timeout exceptions to asyncio timeout errors.
+    with eventlet.timeout.Timeout(
+            seconds=timeout,
+            exception=asyncio.TimeoutError("Timeout: %ss" % timeout)):
+        return tpool.execute(tpool_wrapper)

watcher/conf/__init__.py

@@ -35,6 +35,7 @@ from watcher.conf import grafana_client
 from watcher.conf import grafana_translators
 from watcher.conf import ironic_client
 from watcher.conf import keystone_client
+from watcher.conf import maas_client
 from watcher.conf import monasca_client
 from watcher.conf import neutron_client
 from watcher.conf import nova_client
@@ -54,6 +55,7 @@ db.register_opts(CONF)
 planner.register_opts(CONF)
 applier.register_opts(CONF)
 decision_engine.register_opts(CONF)
+maas_client.register_opts(CONF)
 monasca_client.register_opts(CONF)
 nova_client.register_opts(CONF)
 glance_client.register_opts(CONF)

watcher/conf/grafana_client.py

@@ -134,7 +134,13 @@ GRAFANA_CLIENT_OPTS = [
                      "InfluxDB this will be the retention period. "
                      "These queries will need to be constructed using tools "
                      "such as Postman. Example: SELECT cpu FROM {4}."
-                     "cpu_percent WHERE host == '{1}' AND time > now()-{2}s")]
+                     "cpu_percent WHERE host == '{1}' AND time > now()-{2}s"),
+    cfg.IntOpt('http_timeout',
+               min=0,
+               default=60,
+               mutable=True,
+               help='Timeout for Grafana request')
+]
 
 
 def register_opts(conf):

watcher/conf/maas_client.py

@@ -0,0 +1,38 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+from oslo_config import cfg
+
+maas_client = cfg.OptGroup(name='maas_client',
+                           title='Configuration Options for MaaS')
+
+MAAS_CLIENT_OPTS = [
+    cfg.StrOpt('url',
+               help='MaaS URL, example: http://1.2.3.4:5240/MAAS'),
+    cfg.StrOpt('api_key',
+               help='MaaS API authentication key.'),
+    cfg.IntOpt('timeout',
+               default=60,
+               help='MaaS client operation timeout in seconds.')]
+
+
+def register_opts(conf):
+    conf.register_group(maas_client)
+    conf.register_opts(MAAS_CLIENT_OPTS, group=maas_client)
+
+
+def list_opts():
+    return [(maas_client, MAAS_CLIENT_OPTS)]

watcher/db/sqlalchemy/alembic/versions/0f6042416884_add_apscheduler_jobs.py

@@ -28,7 +28,7 @@ def upgrade():
 
     op.create_table(
         'apscheduler_jobs',
-        sa.Column('id', sa.Unicode(191, _warn_on_bytestring=False),
+        sa.Column('id', sa.Unicode(191),
                   nullable=False),
         sa.Column('next_run_time', sa.Float(25), index=True),
         sa.Column('job_state', sa.LargeBinary, nullable=False),

watcher/db/sqlalchemy/api.py

@@ -44,11 +44,7 @@ _FACADE = None
 def _create_facade_lazily():
     global _FACADE
     if _FACADE is None:
-        # FIXME(amoralej): Remove autocommit=True (and ideally use of
+        _FACADE = db_session.EngineFacade.from_config(CONF)
-        # LegacyEngineFacade) asap since it's not compatible with SQLAlchemy
-        # 2.0.
-        _FACADE = db_session.EngineFacade.from_config(CONF,
-                                                      autocommit=True)
     return _FACADE
 
 
@@ -248,30 +244,36 @@ class Connection(api.BaseConnection):
         for relationship in relationships:
             if not relationship.uselist:
                 # We have a One-to-X relationship
-                query = query.options(joinedload(relationship.key))
+                query = query.options(joinedload(
+                    getattr(model, relationship.key)))
         return query
 
     def _create(self, model, values):
-        obj = model()
+        session = get_session()
-        cleaned_values = {k: v for k, v in values.items()
+        with session.begin():
-                          if k not in self._get_relationships(model)}
+            obj = model()
-        obj.update(cleaned_values)
+            cleaned_values = {k: v for k, v in values.items()
-        obj.save()
+                              if k not in self._get_relationships(model)}
+            obj.update(cleaned_values)
+            obj.save(session=session)
+            session.commit()
         return obj
 
     def _get(self, context, model, fieldname, value, eager):
-        query = model_query(model)
+        session = get_session()
-        if eager:
+        with session.begin():
-            query = self._set_eager_options(model, query)
+            query = model_query(model, session=session)
+            if eager:
+                query = self._set_eager_options(model, query)
 
-        query = query.filter(getattr(model, fieldname) == value)
+            query = query.filter(getattr(model, fieldname) == value)
-        if not context.show_deleted:
+            if not context.show_deleted:
-            query = query.filter(model.deleted_at.is_(None))
+                query = query.filter(model.deleted_at.is_(None))
 
-        try:
+            try:
-            obj = query.one()
+                obj = query.one()
-        except exc.NoResultFound:
+            except exc.NoResultFound:
-            raise exception.ResourceNotFound(name=model.__name__, id=value)
+                raise exception.ResourceNotFound(name=model.__name__, id=value)
 
         return obj
 

watcher/db/sqlalchemy/job_store.py

@@ -22,6 +22,7 @@ from apscheduler.jobstores.base import ConflictingIdError
 from apscheduler.jobstores import sqlalchemy
 from apscheduler.util import datetime_to_utc_timestamp
 from apscheduler.util import maybe_ref
+from apscheduler.util import utc_timestamp_to_datetime
 
 from watcher.common import context
 from watcher.common import service
@@ -32,7 +33,7 @@ try:
 except ImportError:  # pragma: nocover
     import pickle
 
-from sqlalchemy import Table, MetaData, select, and_
+from sqlalchemy import Table, MetaData, select, and_, null
 from sqlalchemy.exc import IntegrityError
 
 
@@ -58,8 +59,7 @@ class WatcherJobStore(sqlalchemy.SQLAlchemyJobStore):
         super(WatcherJobStore, self).__init__(url, engine, tablename,
                                               metadata, pickle_protocol)
         metadata = maybe_ref(metadata) or MetaData()
-        self.jobs_t = Table(tablename, metadata, autoload=True,
+        self.jobs_t = Table(tablename, metadata, autoload_with=engine)
-                            autoload_with=engine)
         service_ident = service.ServiceHeartbeat.get_service_name()
         self.tag = tag or {'host': service_ident[0], 'name': service_ident[1]}
         self.service_id = objects.Service.list(context=context.make_context(),
@@ -79,7 +79,8 @@ class WatcherJobStore(sqlalchemy.SQLAlchemyJobStore):
             'tag': jsonutils.dumps(self.tag)
         })
         try:
-            self.engine.execute(insert)
+            with self.engine.begin() as conn:
+                conn.execute(insert)
         except IntegrityError:
             raise ConflictingIdError(job.id)
 
@@ -88,20 +89,36 @@ class WatcherJobStore(sqlalchemy.SQLAlchemyJobStore):
         self._fix_paused_jobs_sorting(jobs)
         return jobs
 
+    def get_next_run_time(self):
+        selectable = select(self.jobs_t.c.next_run_time).\
+            where(self.jobs_t.c.next_run_time != null()).\
+            order_by(self.jobs_t.c.next_run_time).limit(1)
+        with self.engine.begin() as connection:
+            # NOTE(danms): The apscheduler implementation of this gets a
+            # decimal.Decimal back from scalar() which causes
+            # utc_timestamp_to_datetime() to choke since it is expecting a
+            # python float. Assume this is SQLAlchemy 2.0 stuff, so just
+            # coerce to a float here.
+            next_run_time = connection.execute(selectable).scalar()
+            return utc_timestamp_to_datetime(float(next_run_time)
+                                             if next_run_time is not None
+                                             else None)
+
     def _get_jobs(self, *conditions):
         jobs = []
         conditions += (self.jobs_t.c.service_id == self.service_id,)
         selectable = select(
-            [self.jobs_t.c.id, self.jobs_t.c.job_state, self.jobs_t.c.tag]
+            self.jobs_t.c.id, self.jobs_t.c.job_state, self.jobs_t.c.tag
         ).order_by(self.jobs_t.c.next_run_time).where(and_(*conditions))
         failed_job_ids = set()
-        for row in self.engine.execute(selectable):
+        with self.engine.begin() as conn:
-            try:
+            for row in conn.execute(selectable):
-                jobs.append(self._reconstitute_job(row.job_state))
+                try:
-            except Exception:
+                    jobs.append(self._reconstitute_job(row.job_state))
-                self._logger.exception(
+                except Exception:
-                    'Unable to restore job "%s" -- removing it', row.id)
+                    self._logger.exception(
-                failed_job_ids.add(row.id)
+                        'Unable to restore job "%s" -- removing it', row.id)
+                    failed_job_ids.add(row.id)
 
         # Remove all the jobs we failed to restore
         if failed_job_ids:

watcher/decision_engine/datasources/base.py

@@ -63,7 +63,7 @@ class DataSourceBase(object):
             raise exception.MetricNotAvailable(metric=meter_name)
         return meter
 
-    def query_retry(self, f, *args, **kwargs):
+    def query_retry(self, f, *args, ignored_exc=None, **kwargs):
         """Attempts to retrieve metrics from the external service
 
         Attempts to access data from the external service and handles
@@ -71,15 +71,23 @@ class DataSourceBase(object):
         to the value of query_max_retries
         :param f: The method that performs the actual querying for metrics
         :param args: Array of arguments supplied to the method
+        :param ignored_exc: An exception or tuple of exceptions that shouldn't
+                            be retried, for example "NotFound" exceptions.
         :param kwargs: The amount of arguments supplied to the method
         :return: The value as retrieved from the external service
         """
 
         num_retries = CONF.watcher_datasources.query_max_retries
         timeout = CONF.watcher_datasources.query_timeout
+        ignored_exc = ignored_exc or tuple()
+
         for i in range(num_retries):
             try:
                 return f(*args, **kwargs)
+            except ignored_exc as e:
+                LOG.debug("Got an ignored exception (%s) while calling: %s ",
+                          e, f)
+                return
             except Exception as e:
                 LOG.exception(e)
                 self.query_retry_reset(e)

watcher/decision_engine/datasources/gnocchi.py

@@ -19,6 +19,7 @@
 from datetime import datetime
 from datetime import timedelta
 
+from gnocchiclient import exceptions as gnc_exc
 from oslo_config import cfg
 from oslo_log import log
 
@@ -38,7 +39,7 @@ class GnocchiHelper(base.DataSourceBase):
                       host_inlet_temp='hardware.ipmi.node.temperature',
                       host_airflow='hardware.ipmi.node.airflow',
                       host_power='hardware.ipmi.node.power',
-                      instance_cpu_usage='cpu_util',
+                      instance_cpu_usage='cpu',
                       instance_ram_usage='memory.resident',
                       instance_ram_allocated='memory',
                       instance_l3_cache_usage='cpu_l3_cache',
@@ -84,7 +85,9 @@ class GnocchiHelper(base.DataSourceBase):
             kwargs = dict(query={"=": {"original_resource_id": resource_id}},
                           limit=1)
             resources = self.query_retry(
-                f=self.gnocchi.resource.search, **kwargs)
+                f=self.gnocchi.resource.search,
+                ignored_exc=gnc_exc.NotFound,
+                **kwargs)
 
             if not resources:
                 LOG.warning("The {0} resource {1} could not be "
@@ -93,6 +96,25 @@ class GnocchiHelper(base.DataSourceBase):
 
             resource_id = resources[0]['id']
 
+        if meter_name == "instance_cpu_usage":
+            if resource_type != "instance":
+                LOG.warning("Unsupported resource type for metric "
+                            "'instance_cpu_usage': ", resource_type)
+                return
+
+            # The "cpu_util" gauge (percentage) metric has been removed.
+            # We're going to obtain the same result by using the rate of change
+            # aggregate operation.
+            if aggregate not in ("mean", "rate:mean"):
+                LOG.warning("Unsupported aggregate for instance_cpu_usage "
+                            "metric: %s. "
+                            "Supported aggregates: mean, rate:mean ",
+                            aggregate)
+                return
+
+            # TODO(lpetrut): consider supporting other aggregates.
+            aggregate = "rate:mean"
+
         raw_kwargs = dict(
             metric=meter,
             start=start_time,
@@ -105,7 +127,9 @@ class GnocchiHelper(base.DataSourceBase):
         kwargs = {k: v for k, v in raw_kwargs.items() if k and v}
 
         statistics = self.query_retry(
-            f=self.gnocchi.metric.get_measures, **kwargs)
+            f=self.gnocchi.metric.get_measures,
+            ignored_exc=gnc_exc.NotFound,
+            **kwargs)
 
         return_value = None
         if statistics:
@@ -117,6 +141,17 @@ class GnocchiHelper(base.DataSourceBase):
                 # Airflow from hardware.ipmi.node.airflow is reported as
                 # 1/10 th of actual CFM
                 return_value *= 10
+            if meter_name == "instance_cpu_usage":
+                # "rate:mean" can return negative values for migrated vms.
+                return_value = max(0, return_value)
+
+                # We're converting the cumulative cpu time (ns) to cpu usage
+                # percentage.
+                vcpus = resource.vcpus
+                if not vcpus:
+                    LOG.warning("instance vcpu count not set, assuming 1")
+                    vcpus = 1
+                return_value *= 100 / (granularity * 10e+8) / vcpus
 
         return return_value
 
@@ -132,7 +167,9 @@ class GnocchiHelper(base.DataSourceBase):
             kwargs = dict(query={"=": {"original_resource_id": resource_id}},
                           limit=1)
             resources = self.query_retry(
-                f=self.gnocchi.resource.search, **kwargs)
+                f=self.gnocchi.resource.search,
+                ignored_exc=gnc_exc.NotFound,
+                **kwargs)
 
             if not resources:
                 LOG.warning("The {0} resource {1} could not be "
@@ -152,7 +189,9 @@ class GnocchiHelper(base.DataSourceBase):
         kwargs = {k: v for k, v in raw_kwargs.items() if k and v}
 
         statistics = self.query_retry(
-            f=self.gnocchi.metric.get_measures, **kwargs)
+            f=self.gnocchi.metric.get_measures,
+            ignored_exc=gnc_exc.NotFound,
+            **kwargs)
 
         return_value = None
         if statistics:

watcher/decision_engine/datasources/grafana.py

@@ -138,7 +138,8 @@ class GrafanaHelper(base.DataSourceBase):
             raise exception.DataSourceNotAvailable(self.NAME)
 
         resp = requests.get(self._base_url + str(project_id) + '/query',
-                            params=params, headers=self._headers)
+                            params=params, headers=self._headers,
+                            timeout=CONF.grafana_client.http_timeout)
         if resp.status_code == HTTPStatus.OK:
             return resp
         elif resp.status_code == HTTPStatus.BAD_REQUEST:

watcher/decision_engine/model/collector/ironic.py

@@ -81,6 +81,7 @@ class BareMetalModelBuilder(base.BaseModelBuilder):
     def __init__(self, osc):
         self.osc = osc
         self.model = model_root.BaremetalModelRoot()
+        # TODO(lpetrut): add MAAS support
         self.ironic_helper = ironic_helper.IronicHelper(osc=self.osc)
 
     def add_ironic_node(self, node):

watcher/decision_engine/model/collector/nova.py

@@ -48,7 +48,7 @@ class NovaClusterDataModelCollector(base.BaseClusterDataModelCollector):
                     "type": "array",
                     "items": {
                         "anyOf": [
-                            {"$ref": HOST_AGGREGATES + "id"},
+                            {"$ref": HOST_AGGREGATES + "host_aggr_id"},
                             {"$ref": HOST_AGGREGATES + "name"},
                         ]
                     }
@@ -98,7 +98,8 @@ class NovaClusterDataModelCollector(base.BaseClusterDataModelCollector):
                                 "type": "array",
                                 "items": {
                                     "anyOf": [
-                                        {"$ref": HOST_AGGREGATES + "id"},
+                                        {"$ref":
+                                            HOST_AGGREGATES + "host_aggr_id"},
                                         {"$ref": HOST_AGGREGATES + "name"},
                                     ]
                                 }
@@ -129,7 +130,7 @@ class NovaClusterDataModelCollector(base.BaseClusterDataModelCollector):
             "additionalProperties": False
         },
         "host_aggregates": {
-            "id": {
+            "host_aggr_id": {
                 "properties": {
                     "id": {
                         "oneOf": [

watcher/decision_engine/model/model_root.py

@@ -157,7 +157,7 @@ class ModelRoot(nx.DiGraph, base.Model):
             if node_list:
                 return node_list[0]
             else:
-                raise exception.ComputeResourceNotFound
+                raise exception.ComputeNodeNotFound(name=name)
         except exception.ComputeResourceNotFound:
             raise exception.ComputeNodeNotFound(name=name)
 

watcher/decision_engine/strategy/strategies/base.py

@@ -252,9 +252,6 @@ class BaseStrategy(loadable.Loadable, metaclass=abc.ABCMeta):
         if not self.compute_model:
             raise exception.ClusterStateNotDefined()
 
-        if self.compute_model.stale:
-            raise exception.ClusterStateStale()
-
         LOG.debug(self.compute_model.to_string())
 
     def execute(self, audit=None):

watcher/decision_engine/strategy/strategies/saving_energy.py

@@ -23,6 +23,8 @@ from oslo_log import log
 
 from watcher._i18n import _
 from watcher.common import exception
+from watcher.common.metal_helper import constants as metal_constants
+from watcher.common.metal_helper import factory as metal_helper_factory
 from watcher.decision_engine.strategy.strategies import base
 
 LOG = log.getLogger(__name__)
@@ -81,7 +83,7 @@ class SavingEnergy(base.SavingEnergyBaseStrategy):
     def __init__(self, config, osc=None):
 
         super(SavingEnergy, self).__init__(config, osc)
-        self._ironic_client = None
+        self._metal_helper = None
         self._nova_client = None
 
         self.with_vms_node_pool = []
@@ -91,10 +93,10 @@ class SavingEnergy(base.SavingEnergyBaseStrategy):
         self.min_free_hosts_num = 1
 
     @property
-    def ironic_client(self):
+    def metal_helper(self):
-        if not self._ironic_client:
+        if not self._metal_helper:
-            self._ironic_client = self.osc.ironic()
+            self._metal_helper = metal_helper_factory.get_helper(self.osc)
-        return self._ironic_client
+        return self._metal_helper
 
     @property
     def nova_client(self):
@@ -149,10 +151,10 @@ class SavingEnergy(base.SavingEnergyBaseStrategy):
         :return: None
         """
         params = {'state': state,
-                  'resource_name': node.hostname}
+                  'resource_name': node.get_hypervisor_hostname()}
         self.solution.add_action(
             action_type='change_node_power_state',
-            resource_id=node.uuid,
+            resource_id=node.get_id(),
             input_parameters=params)
 
     def get_hosts_pool(self):
@@ -162,36 +164,36 @@ class SavingEnergy(base.SavingEnergyBaseStrategy):
 
         """
 
-        node_list = self.ironic_client.node.list()
+        node_list = self.metal_helper.list_compute_nodes()
         for node in node_list:
-            node_info = self.ironic_client.node.get(node.uuid)
+            hypervisor_node = node.get_hypervisor_node().to_dict()
-            hypervisor_id = node_info.extra.get('compute_node_id', None)
+
-            if hypervisor_id is None:
-                LOG.warning(('Cannot find compute_node_id in extra '
-                             'of ironic node %s'), node.uuid)
-                continue
-            hypervisor_node = self.nova_client.hypervisors.get(hypervisor_id)
-            if hypervisor_node is None:
-                LOG.warning(('Cannot find hypervisor %s'), hypervisor_id)
-                continue
-            node.hostname = hypervisor_node.hypervisor_hostname
-            hypervisor_node = hypervisor_node.to_dict()
             compute_service = hypervisor_node.get('service', None)
             host_name = compute_service.get('host')
+            LOG.debug("Found hypervisor: %s", hypervisor_node)
             try:
                 self.compute_model.get_node_by_name(host_name)
             except exception.ComputeNodeNotFound:
+                LOG.info("The compute model does not contain the host: %s",
+                         host_name)
                 continue
 
-            if not (hypervisor_node.get('state') == 'up'):
+            if (node.hv_up_when_powered_off and
-                """filter nodes that are not in 'up' state"""
+                    hypervisor_node.get('state') != 'up'):
+                # filter nodes that are not in 'up' state
+                LOG.info("Ignoring node that isn't in 'up' state: %s",
+                         host_name)
                 continue
             else:
                 if (hypervisor_node['running_vms'] == 0):
-                    if (node_info.power_state == 'power on'):
+                    power_state = node.get_power_state()
+                    if power_state == metal_constants.PowerState.ON:
                         self.free_poweron_node_pool.append(node)
-                    elif (node_info.power_state == 'power off'):
+                    elif power_state == metal_constants.PowerState.OFF:
                         self.free_poweroff_node_pool.append(node)
+                    else:
+                        LOG.info("Ignoring node %s, unknown state: %s",
+                                 node, power_state)
                 else:
                     self.with_vms_node_pool.append(node)
 
@@ -202,17 +204,21 @@ class SavingEnergy(base.SavingEnergyBaseStrategy):
                 self.min_free_hosts_num)))
         len_poweron = len(self.free_poweron_node_pool)
         len_poweroff = len(self.free_poweroff_node_pool)
+        LOG.debug("need_poweron: %s, len_poweron: %s, len_poweroff: %s",
+                  need_poweron, len_poweron, len_poweroff)
         if len_poweron > need_poweron:
             for node in random.sample(self.free_poweron_node_pool,
                                       (len_poweron - need_poweron)):
-                self.add_action_poweronoff_node(node, 'off')
+                self.add_action_poweronoff_node(node,
-                LOG.info("power off %s", node.uuid)
+                                                metal_constants.PowerState.OFF)
+                LOG.info("power off %s", node.get_id())
         elif len_poweron < need_poweron:
             diff = need_poweron - len_poweron
             for node in random.sample(self.free_poweroff_node_pool,
                                       min(len_poweroff, diff)):
-                self.add_action_poweronoff_node(node, 'on')
+                self.add_action_poweronoff_node(node,
-                LOG.info("power on %s", node.uuid)
+                                                metal_constants.PowerState.ON)
+                LOG.info("power on %s", node.get_id())
 
     def pre_execute(self):
         self._pre_execute()

watcher/decision_engine/strategy/strategies/vm_workload_consolidation.py

@@ -18,9 +18,13 @@
 # limitations under the License.
 #
 
+import collections
+
 from oslo_log import log
+import oslo_utils
 
 from watcher._i18n import _
+from watcher.applier.actions import migration
 from watcher.common import exception
 from watcher.decision_engine.model import element
 from watcher.decision_engine.strategy.strategies import base
@@ -66,7 +70,8 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
 
     AGGREGATE = 'mean'
     DATASOURCE_METRICS = ['instance_ram_allocated', 'instance_cpu_usage',
-                          'instance_ram_usage', 'instance_root_disk_size']
+                          'instance_ram_usage', 'instance_root_disk_size',
+                          'host_cpu_usage', 'host_ram_usage']
 
     MIGRATION = "migrate"
     CHANGE_NOVA_SERVICE_STATE = "change_nova_service_state"
@@ -76,6 +81,11 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
         self.number_of_migrations = 0
         self.number_of_released_nodes = 0
         self.datasource_instance_data_cache = dict()
+        self.datasource_node_data_cache = dict()
+        # Host metric adjustments that take into account planned
+        # migrations.
+        self.host_metric_delta = collections.defaultdict(
+            lambda: collections.defaultdict(int))
 
     @classmethod
     def get_name(cls):
@@ -196,12 +206,12 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
         :return: None
         """
         instance_state_str = self.get_instance_state_str(instance)
-        if instance_state_str not in (element.InstanceState.ACTIVE.value,
+        if instance_state_str in (element.InstanceState.ACTIVE.value,
-                                      element.InstanceState.PAUSED.value):
+                                  element.InstanceState.PAUSED.value):
-            # Watcher currently only supports live VM migration and block live
+            migration_type = migration.Migrate.LIVE_MIGRATION
-            # VM migration which both requires migrated VM to be active.
+        elif instance_state_str == element.InstanceState.STOPPED.value:
-            # When supported, the cold migration may be used as a fallback
+            migration_type = migration.Migrate.COLD_MIGRATION
-            # migration mechanism to move non active VMs.
+        else:
             LOG.error(
                 'Cannot live migrate: instance_uuid=%(instance_uuid)s, '
                 'state=%(instance_state)s.', dict(
@@ -209,8 +219,6 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                     instance_state=instance_state_str))
             return
 
-        migration_type = 'live'
-
         # Here will makes repeated actions to enable the same compute node,
         # when migrating VMs to the destination node which is disabled.
         # Whether should we remove the same actions in the solution???
@@ -228,6 +236,18 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                 destination_node)
             self.number_of_migrations += 1
 
+            instance_util = self.get_instance_utilization(instance)
+            self.host_metric_delta[source_node.hostname]['cpu'] -= (
+                instance_util['cpu'])
+            # We'll deduce the vm allocated memory.
+            self.host_metric_delta[source_node.hostname]['ram'] -= (
+                instance.memory)
+
+            self.host_metric_delta[destination_node.hostname]['cpu'] += (
+                instance_util['cpu'])
+            self.host_metric_delta[destination_node.hostname]['ram'] += (
+                instance.memory)
+
     def disable_unused_nodes(self):
         """Generate actions for disabling unused nodes.
 
@@ -290,6 +310,21 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
             disk=instance_disk_util)
         return self.datasource_instance_data_cache.get(instance.uuid)
 
+    def _get_node_total_utilization(self, node):
+        if node.hostname in self.datasource_node_data_cache:
+            return self.datasource_node_data_cache[node.hostname]
+
+        cpu = self.datasource_backend.get_host_cpu_usage(
+            node, self.period, self.AGGREGATE,
+            self.granularity)
+        ram = self.datasource_backend.get_host_ram_usage(
+            node, self.period, self.AGGREGATE,
+            self.granularity)
+
+        self.datasource_node_data_cache[node.hostname] = dict(
+            cpu=cpu, ram=ram)
+        return self.datasource_node_data_cache[node.hostname]
+
     def get_node_utilization(self, node):
         """Collect cpu, ram and disk utilization statistics of a node.
 
@@ -307,8 +342,36 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
             node_cpu_util += instance_util['cpu']
             node_ram_util += instance_util['ram']
             node_disk_util += instance_util['disk']
+            LOG.debug("instance utilization: %s %s",
+                      instance, instance_util)
 
-        return dict(cpu=node_cpu_util, ram=node_ram_util,
+        total_node_util = self._get_node_total_utilization(node)
+        total_node_cpu_util = total_node_util['cpu'] or 0
+        if total_node_cpu_util:
+            total_node_cpu_util = total_node_cpu_util * node.vcpus / 100
+            # account for planned migrations
+            total_node_cpu_util += self.host_metric_delta[node.hostname]['cpu']
+
+        total_node_ram_util = total_node_util['ram'] or 0
+        if total_node_ram_util:
+            total_node_ram_util /= oslo_utils.units.Ki
+            total_node_ram_util += self.host_metric_delta[node.hostname]['ram']
+
+        LOG.debug(
+            "node utilization: %s. "
+            "total instance cpu: %s, "
+            "total instance ram: %s, "
+            "total instance disk: %s, "
+            "total host cpu: %s, "
+            "total host ram: %s, "
+            "node delta usage: %s.",
+            node,
+            node_cpu_util, node_ram_util, node_disk_util,
+            total_node_cpu_util, total_node_ram_util,
+            self.host_metric_delta[node.hostname])
+
+        return dict(cpu=max(node_cpu_util, total_node_cpu_util),
+                    ram=max(node_ram_util, total_node_ram_util),
                     disk=node_disk_util)
 
     def get_node_capacity(self, node):
@@ -388,8 +451,15 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
         instance_utilization = self.get_instance_utilization(instance)
         metrics = ['cpu', 'ram', 'disk']
         for m in metrics:
-            if (instance_utilization[m] + node_utilization[m] >
+            fits = (instance_utilization[m] + node_utilization[m] <=
-                    node_capacity[m] * cc[m]):
+                    node_capacity[m] * cc[m])
+            LOG.debug(
+                "Instance fits: %s, metric: %s, instance: %s, "
+                "node: %s, instance utilization: %s, "
+                "node utilization: %s, node capacity: %s, cc: %s",
+                fits, m, instance, node, instance_utilization[m],
+                node_utilization[m], node_capacity[m], cc[m])
+            if not fits:
                 return False
         return True
 
@@ -424,6 +494,9 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                 for a in actions:
                     self.solution.actions.remove(a)
                     self.number_of_migrations -= 1
+                LOG.info("Optimized migrations: %s. "
+                         "Source: %s, destination: %s", actions,
+                         src_name, dst_name)
                 src_node = self.compute_model.get_node_by_name(src_name)
                 dst_node = self.compute_model.get_node_by_name(dst_name)
                 instance = self.compute_model.get_instance_by_uuid(
@@ -460,6 +533,8 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                         key=lambda x: self.get_instance_utilization(
                             x)['cpu']
                 ):
+                    LOG.info("Node %s overloaded, attempting to reduce load.",
+                             node)
                     # skip exclude instance when migrating
                     if instance.watcher_exclude:
                         LOG.debug("Instance is excluded by scope, "
@@ -468,11 +543,19 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                     for destination_node in reversed(sorted_nodes):
                         if self.instance_fits(
                                 instance, destination_node, cc):
+                            LOG.info("Offload: found fitting "
+                                     "destination (%s) for instance: %s. "
+                                     "Planning migration.",
+                                     destination_node, instance.uuid)
                             self.add_migration(instance, node,
                                                destination_node)
                             break
                     if not self.is_overloaded(node, cc):
+                        LOG.info("Node %s no longer overloaded.", node)
                         break
+                    else:
+                        LOG.info("Node still overloaded (%s), "
+                                 "continuing offload phase.", node)
 
     def consolidation_phase(self, cc):
         """Perform consolidation phase.
@@ -508,6 +591,10 @@ class VMWorkloadConsolidation(base.ServerConsolidationBaseStrategy):
                         break
                     if self.instance_fits(
                             instance, destination_node, cc):
+                        LOG.info("Consolidation: found fitting "
+                                 "destination (%s) for instance: %s. "
+                                 "Planning migration.",
+                                 destination_node, instance.uuid)
                         self.add_migration(instance, node,
                                            destination_node)
                         break

watcher/decision_engine/strategy/strategies/workload_balance.py

@@ -295,7 +295,7 @@ class WorkloadBalance(base.WorkloadStabilizationBaseStrategy):
                         self.threshold)
             return self.solution
 
-        # choose the server with largest cpu_util
+        # choose the server with largest cpu usage
         source_nodes = sorted(source_nodes,
                               reverse=True,
                               key=lambda x: (x[self._meter]))

watcher/tests/applier/actions/test_change_node_power_state.py

@@ -19,134 +19,151 @@ import jsonschema
 
 from watcher.applier.actions import base as baction
 from watcher.applier.actions import change_node_power_state
-from watcher.common import clients
+from watcher.common.metal_helper import constants as m_constants
+from watcher.common.metal_helper import factory as m_helper_factory
 from watcher.tests import base
+from watcher.tests.decision_engine import fake_metal_helper
 
 COMPUTE_NODE = "compute-1"
 
 
-@mock.patch.object(clients.OpenStackClients, 'nova')
-@mock.patch.object(clients.OpenStackClients, 'ironic')
 class TestChangeNodePowerState(base.TestCase):
 
     def setUp(self):
         super(TestChangeNodePowerState, self).setUp()
 
+        p_m_factory = mock.patch.object(m_helper_factory, 'get_helper')
+        m_factory = p_m_factory.start()
+        self._metal_helper = m_factory.return_value
+        self.addCleanup(p_m_factory.stop)
+
+        # Let's avoid unnecessary sleep calls while running the test.
+        p_sleep = mock.patch('time.sleep')
+        p_sleep.start()
+        self.addCleanup(p_sleep.stop)
+
         self.input_parameters = {
             baction.BaseAction.RESOURCE_ID: COMPUTE_NODE,
-            "state": change_node_power_state.NodeState.POWERON.value,
+            "state": m_constants.PowerState.ON.value,
         }
         self.action = change_node_power_state.ChangeNodePowerState(
             mock.Mock())
         self.action.input_parameters = self.input_parameters
 
-    def test_parameters_down(self, mock_ironic, mock_nova):
+    def test_parameters_down(self):
         self.action.input_parameters = {
             baction.BaseAction.RESOURCE_ID: COMPUTE_NODE,
             self.action.STATE:
-                change_node_power_state.NodeState.POWEROFF.value}
+                m_constants.PowerState.OFF.value}
         self.assertTrue(self.action.validate_parameters())
 
-    def test_parameters_up(self, mock_ironic, mock_nova):
+    def test_parameters_up(self):
         self.action.input_parameters = {
             baction.BaseAction.RESOURCE_ID: COMPUTE_NODE,
             self.action.STATE:
-                change_node_power_state.NodeState.POWERON.value}
+                m_constants.PowerState.ON.value}
         self.assertTrue(self.action.validate_parameters())
 
-    def test_parameters_exception_wrong_state(self, mock_ironic, mock_nova):
+    def test_parameters_exception_wrong_state(self):
         self.action.input_parameters = {
             baction.BaseAction.RESOURCE_ID: COMPUTE_NODE,
             self.action.STATE: 'error'}
         self.assertRaises(jsonschema.ValidationError,
                           self.action.validate_parameters)
 
-    def test_parameters_resource_id_empty(self, mock_ironic, mock_nova):
+    def test_parameters_resource_id_empty(self):
         self.action.input_parameters = {
             self.action.STATE:
-                change_node_power_state.NodeState.POWERON.value,
+                m_constants.PowerState.ON.value,
         }
         self.assertRaises(jsonschema.ValidationError,
                           self.action.validate_parameters)
 
-    def test_parameters_applies_add_extra(self, mock_ironic, mock_nova):
+    def test_parameters_applies_add_extra(self):
         self.action.input_parameters = {"extra": "failed"}
         self.assertRaises(jsonschema.ValidationError,
                           self.action.validate_parameters)
 
-    def test_change_service_state_pre_condition(self, mock_ironic, mock_nova):
+    def test_change_service_state_pre_condition(self):
         try:
             self.action.pre_condition()
         except Exception as exc:
             self.fail(exc)
 
-    def test_change_node_state_post_condition(self, mock_ironic, mock_nova):
+    def test_change_node_state_post_condition(self):
         try:
             self.action.post_condition()
         except Exception as exc:
             self.fail(exc)
 
-    def test_execute_node_service_state_with_poweron_target(
+    def test_execute_node_service_state_with_poweron_target(self):
-            self, mock_ironic, mock_nova):
-        mock_irclient = mock_ironic.return_value
         self.action.input_parameters["state"] = (
-            change_node_power_state.NodeState.POWERON.value)
+            m_constants.PowerState.ON.value)
-        mock_irclient.node.get.side_effect = [
+        mock_nodes = [
-            mock.MagicMock(power_state='power off'),
+            fake_metal_helper.get_mock_metal_node(
-            mock.MagicMock(power_state='power on')]
+                power_state=m_constants.PowerState.OFF),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.ON)
+        ]
+        self._metal_helper.get_node.side_effect = mock_nodes
 
         result = self.action.execute()
         self.assertTrue(result)
 
-        mock_irclient.node.set_power_state.assert_called_once_with(
+        mock_nodes[0].set_power_state.assert_called_once_with(
-            COMPUTE_NODE, change_node_power_state.NodeState.POWERON.value)
+            m_constants.PowerState.ON.value)
 
-    def test_execute_change_node_state_with_poweroff_target(
+    def test_execute_change_node_state_with_poweroff_target(self):
-            self, mock_ironic, mock_nova):
-        mock_irclient = mock_ironic.return_value
-        mock_nvclient = mock_nova.return_value
-        mock_get = mock.MagicMock()
-        mock_get.to_dict.return_value = {'running_vms': 0}
-        mock_nvclient.hypervisors.get.return_value = mock_get
         self.action.input_parameters["state"] = (
-            change_node_power_state.NodeState.POWEROFF.value)
+            m_constants.PowerState.OFF.value)
-        mock_irclient.node.get.side_effect = [
+
-            mock.MagicMock(power_state='power on'),
+        mock_nodes = [
-            mock.MagicMock(power_state='power on'),
+            fake_metal_helper.get_mock_metal_node(
-            mock.MagicMock(power_state='power off')]
+                power_state=m_constants.PowerState.ON),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.ON),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.OFF)
+        ]
+        self._metal_helper.get_node.side_effect = mock_nodes
+
         result = self.action.execute()
         self.assertTrue(result)
 
-        mock_irclient.node.set_power_state.assert_called_once_with(
+        mock_nodes[0].set_power_state.assert_called_once_with(
-            COMPUTE_NODE, change_node_power_state.NodeState.POWEROFF.value)
+            m_constants.PowerState.OFF.value)
 
-    def test_revert_change_node_state_with_poweron_target(
+    def test_revert_change_node_state_with_poweron_target(self):
-            self, mock_ironic, mock_nova):
-        mock_irclient = mock_ironic.return_value
-        mock_nvclient = mock_nova.return_value
-        mock_get = mock.MagicMock()
-        mock_get.to_dict.return_value = {'running_vms': 0}
-        mock_nvclient.hypervisors.get.return_value = mock_get
         self.action.input_parameters["state"] = (
-            change_node_power_state.NodeState.POWERON.value)
+            m_constants.PowerState.ON.value)
-        mock_irclient.node.get.side_effect = [
+
-            mock.MagicMock(power_state='power on'),
+        mock_nodes = [
-            mock.MagicMock(power_state='power on'),
+            fake_metal_helper.get_mock_metal_node(
-            mock.MagicMock(power_state='power off')]
+                power_state=m_constants.PowerState.ON),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.ON),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.OFF)
+        ]
+        self._metal_helper.get_node.side_effect = mock_nodes
+
         self.action.revert()
 
-        mock_irclient.node.set_power_state.assert_called_once_with(
+        mock_nodes[0].set_power_state.assert_called_once_with(
-            COMPUTE_NODE, change_node_power_state.NodeState.POWEROFF.value)
+            m_constants.PowerState.OFF.value)
 
-    def test_revert_change_node_state_with_poweroff_target(
+    def test_revert_change_node_state_with_poweroff_target(self):
-            self, mock_ironic, mock_nova):
-        mock_irclient = mock_ironic.return_value
         self.action.input_parameters["state"] = (
-            change_node_power_state.NodeState.POWEROFF.value)
+            m_constants.PowerState.OFF.value)
-        mock_irclient.node.get.side_effect = [
+        mock_nodes = [
-            mock.MagicMock(power_state='power off'),
+            fake_metal_helper.get_mock_metal_node(
-            mock.MagicMock(power_state='power on')]
+                power_state=m_constants.PowerState.OFF),
+            fake_metal_helper.get_mock_metal_node(
+                power_state=m_constants.PowerState.ON)
+        ]
+        self._metal_helper.get_node.side_effect = mock_nodes
+
         self.action.revert()
 
-        mock_irclient.node.set_power_state.assert_called_once_with(
+        mock_nodes[0].set_power_state.assert_called_once_with(
-            COMPUTE_NODE, change_node_power_state.NodeState.POWERON.value)
+            m_constants.PowerState.ON.value)

watcher/tests/common/metal_helper/test_base.py

@@ -0,0 +1,96 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+from watcher.common import exception
+from watcher.common.metal_helper import base as m_helper_base
+from watcher.common.metal_helper import constants as m_constants
+from watcher.tests import base
+
+
+# The base classes have abstract methods, we'll need to
+# stub them.
+class MockMetalNode(m_helper_base.BaseMetalNode):
+    def get_power_state(self):
+        raise NotImplementedError()
+
+    def get_id(self):
+        raise NotImplementedError()
+
+    def power_on(self):
+        raise NotImplementedError()
+
+    def power_off(self):
+        raise NotImplementedError()
+
+
+class MockMetalHelper(m_helper_base.BaseMetalHelper):
+    def list_compute_nodes(self):
+        pass
+
+    def get_node(self, node_id):
+        pass
+
+
+class TestBaseMetalNode(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._nova_node = mock.Mock()
+        self._node = MockMetalNode(self._nova_node)
+
+    def test_get_hypervisor_node(self):
+        self.assertEqual(
+            self._nova_node,
+            self._node.get_hypervisor_node())
+
+    def test_get_hypervisor_node_missing(self):
+        node = MockMetalNode()
+        self.assertRaises(
+            exception.Invalid,
+            node.get_hypervisor_node)
+
+    def test_get_hypervisor_hostname(self):
+        self.assertEqual(
+            self._nova_node.hypervisor_hostname,
+            self._node.get_hypervisor_hostname())
+
+    @mock.patch.object(MockMetalNode, 'power_on')
+    @mock.patch.object(MockMetalNode, 'power_off')
+    def test_set_power_state(self,
+                             mock_power_off, mock_power_on):
+        self._node.set_power_state(m_constants.PowerState.ON)
+        mock_power_on.assert_called_once_with()
+
+        self._node.set_power_state(m_constants.PowerState.OFF)
+        mock_power_off.assert_called_once_with()
+
+        self.assertRaises(
+            exception.UnsupportedActionType,
+            self._node.set_power_state,
+            m_constants.PowerState.UNKNOWN)
+
+
+class TestBaseMetalHelper(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._osc = mock.Mock()
+        self._helper = MockMetalHelper(self._osc)
+
+    def test_nova_client_attr(self):
+        self.assertEqual(self._osc.nova.return_value,
+                         self._helper.nova_client)

watcher/tests/common/metal_helper/test_factory.py

@@ -0,0 +1,38 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+from watcher.common import clients
+from watcher.common.metal_helper import factory
+from watcher.common.metal_helper import ironic
+from watcher.common.metal_helper import maas
+from watcher.tests import base
+
+
+class TestMetalHelperFactory(base.TestCase):
+
+    @mock.patch.object(clients, 'OpenStackClients')
+    @mock.patch.object(maas, 'MaasHelper')
+    @mock.patch.object(ironic, 'IronicHelper')
+    def test_factory(self, mock_ironic, mock_maas, mock_osc):
+        self.assertEqual(
+            mock_ironic.return_value,
+            factory.get_helper())
+
+        self.config(url="fake_maas_url", group="maas_client")
+        self.assertEqual(
+            mock_maas.return_value,
+            factory.get_helper())

watcher/tests/common/metal_helper/test_ironic.py

@@ -0,0 +1,128 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+from watcher.common.metal_helper import constants as m_constants
+from watcher.common.metal_helper import ironic
+from watcher.tests import base
+
+
+class TestIronicNode(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._wrapped_node = mock.Mock()
+        self._nova_node = mock.Mock()
+        self._ironic_client = mock.Mock()
+
+        self._node = ironic.IronicNode(
+            self._wrapped_node, self._nova_node, self._ironic_client)
+
+    def test_get_power_state(self):
+        states = (
+            "power on",
+            "power off",
+            "rebooting",
+            "soft power off",
+            "soft reboot",
+            'SomeOtherState')
+        type(self._wrapped_node).power_state = mock.PropertyMock(
+            side_effect=states)
+
+        expected_states = (
+            m_constants.PowerState.ON,
+            m_constants.PowerState.OFF,
+            m_constants.PowerState.ON,
+            m_constants.PowerState.OFF,
+            m_constants.PowerState.ON,
+            m_constants.PowerState.UNKNOWN)
+
+        for expected_state in expected_states:
+            actual_state = self._node.get_power_state()
+            self.assertEqual(expected_state, actual_state)
+
+    def test_get_id(self):
+        self.assertEqual(
+            self._wrapped_node.uuid,
+            self._node.get_id())
+
+    def test_power_on(self):
+        self._node.power_on()
+        self._ironic_client.node.set_power_state.assert_called_once_with(
+            self._wrapped_node.uuid, "on")
+
+    def test_power_off(self):
+        self._node.power_off()
+        self._ironic_client.node.set_power_state.assert_called_once_with(
+            self._wrapped_node.uuid, "off")
+
+
+class TestIronicHelper(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._mock_osc = mock.Mock()
+        self._mock_nova_client = self._mock_osc.nova.return_value
+        self._mock_ironic_client = self._mock_osc.ironic.return_value
+        self._helper = ironic.IronicHelper(osc=self._mock_osc)
+
+    def test_list_compute_nodes(self):
+        mock_machines = [
+            mock.Mock(
+                extra=dict(compute_node_id=mock.sentinel.compute_node_id)),
+            mock.Mock(
+                extra=dict(compute_node_id=mock.sentinel.compute_node_id2)),
+            mock.Mock(
+                extra=dict())
+        ]
+        mock_hypervisor = mock.Mock()
+
+        self._mock_ironic_client.node.list.return_value = mock_machines
+        self._mock_ironic_client.node.get.side_effect = mock_machines
+        self._mock_nova_client.hypervisors.get.side_effect = (
+            mock_hypervisor, None)
+
+        out_nodes = self._helper.list_compute_nodes()
+        self.assertEqual(1, len(out_nodes))
+
+        out_node = out_nodes[0]
+        self.assertIsInstance(out_node, ironic.IronicNode)
+        self.assertEqual(mock_hypervisor, out_node._nova_node)
+        self.assertEqual(mock_machines[0], out_node._ironic_node)
+        self.assertEqual(self._mock_ironic_client, out_node._ironic_client)
+
+    def test_get_node(self):
+        mock_machine = mock.Mock(
+            extra=dict(compute_node_id=mock.sentinel.compute_node_id))
+        self._mock_ironic_client.node.get.return_value = mock_machine
+
+        out_node = self._helper.get_node(mock.sentinel.id)
+
+        self.assertEqual(self._mock_nova_client.hypervisors.get.return_value,
+                         out_node._nova_node)
+        self.assertEqual(self._mock_ironic_client, out_node._ironic_client)
+        self.assertEqual(mock_machine, out_node._ironic_node)
+
+    def test_get_node_not_a_hypervisor(self):
+        mock_machine = mock.Mock(extra=dict(compute_node_id=None))
+        self._mock_ironic_client.node.get.return_value = mock_machine
+
+        out_node = self._helper.get_node(mock.sentinel.id)
+
+        self._mock_nova_client.hypervisors.get.assert_not_called()
+        self.assertIsNone(out_node._nova_node)
+        self.assertEqual(self._mock_ironic_client, out_node._ironic_client)
+        self.assertEqual(mock_machine, out_node._ironic_node)

watcher/tests/common/metal_helper/test_maas.py

@@ -0,0 +1,126 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+try:
+    from maas.client import enum as maas_enum
+except ImportError:
+    maas_enum = None
+
+from watcher.common.metal_helper import constants as m_constants
+from watcher.common.metal_helper import maas
+from watcher.tests import base
+
+
+class TestMaasNode(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._wrapped_node = mock.Mock()
+        self._nova_node = mock.Mock()
+        self._maas_client = mock.Mock()
+
+        self._node = maas.MaasNode(
+            self._wrapped_node, self._nova_node, self._maas_client)
+
+    def test_get_power_state(self):
+        if not maas_enum:
+            self.skipTest("python-libmaas not intalled.")
+
+        self._wrapped_node.query_power_state.side_effect = (
+            maas_enum.PowerState.ON,
+            maas_enum.PowerState.OFF,
+            maas_enum.PowerState.ERROR,
+            maas_enum.PowerState.UNKNOWN,
+            'SomeOtherState')
+
+        expected_states = (
+            m_constants.PowerState.ON,
+            m_constants.PowerState.OFF,
+            m_constants.PowerState.ERROR,
+            m_constants.PowerState.UNKNOWN,
+            m_constants.PowerState.UNKNOWN)
+
+        for expected_state in expected_states:
+            actual_state = self._node.get_power_state()
+            self.assertEqual(expected_state, actual_state)
+
+    def test_get_id(self):
+        self.assertEqual(
+            self._wrapped_node.system_id,
+            self._node.get_id())
+
+    def test_power_on(self):
+        self._node.power_on()
+        self._wrapped_node.power_on.assert_called_once_with()
+
+    def test_power_off(self):
+        self._node.power_off()
+        self._wrapped_node.power_off.assert_called_once_with()
+
+
+class TestMaasHelper(base.TestCase):
+    def setUp(self):
+        super().setUp()
+
+        self._mock_osc = mock.Mock()
+        self._mock_nova_client = self._mock_osc.nova.return_value
+        self._mock_maas_client = self._mock_osc.maas.return_value
+        self._helper = maas.MaasHelper(osc=self._mock_osc)
+
+    def test_list_compute_nodes(self):
+        compute_fqdn = "compute-0"
+        # some other MAAS node, not a Nova node
+        ctrl_fqdn = "ctrl-1"
+
+        mock_machines = [
+            mock.Mock(fqdn=compute_fqdn,
+                      system_id=mock.sentinel.compute_node_id),
+            mock.Mock(fqdn=ctrl_fqdn,
+                      system_id=mock.sentinel.ctrl_node_id),
+        ]
+        mock_hypervisors = [
+            mock.Mock(hypervisor_hostname=compute_fqdn),
+        ]
+
+        self._mock_maas_client.machines.list.return_value = mock_machines
+        self._mock_nova_client.hypervisors.list.return_value = mock_hypervisors
+
+        out_nodes = self._helper.list_compute_nodes()
+        self.assertEqual(1, len(out_nodes))
+
+        out_node = out_nodes[0]
+        self.assertIsInstance(out_node, maas.MaasNode)
+        self.assertEqual(mock.sentinel.compute_node_id, out_node.get_id())
+        self.assertEqual(compute_fqdn, out_node.get_hypervisor_hostname())
+
+    def test_get_node(self):
+        mock_machine = mock.Mock(fqdn='compute-0')
+        self._mock_maas_client.machines.get.return_value = mock_machine
+
+        mock_compute_nodes = [
+            mock.Mock(hypervisor_hostname="compute-011"),
+            mock.Mock(hypervisor_hostname="compute-0"),
+            mock.Mock(hypervisor_hostname="compute-01"),
+        ]
+        self._mock_nova_client.hypervisors.search.return_value = (
+            mock_compute_nodes)
+
+        out_node = self._helper.get_node(mock.sentinel.id)
+
+        self.assertEqual(mock_compute_nodes[1], out_node._nova_node)
+        self.assertEqual(self._mock_maas_client, out_node._maas_client)
+        self.assertEqual(mock_machine, out_node._maas_node)

watcher/tests/common/test_utils.py

@@ -0,0 +1,52 @@
+# Copyright 2023 Cloudbase Solutions
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import asyncio
+import time
+from unittest import mock
+
+from watcher.common import utils
+from watcher.tests import base
+
+
+class TestCommonUtils(base.TestCase):
+    async def test_coro(self, sleep=0, raise_exc=None):
+        time.sleep(sleep)
+        if raise_exc:
+            raise raise_exc
+        return mock.sentinel.ret_val
+
+    def test_async_compat(self):
+        ret_val = utils.async_compat_call(self.test_coro)
+        self.assertEqual(mock.sentinel.ret_val, ret_val)
+
+    def test_async_compat_exc(self):
+        self.assertRaises(
+            IOError,
+            utils.async_compat_call,
+            self.test_coro,
+            raise_exc=IOError('fake error'))
+
+    def test_async_compat_timeout(self):
+        # Timeout not reached.
+        ret_val = utils.async_compat_call(self.test_coro, timeout=10)
+        self.assertEqual(mock.sentinel.ret_val, ret_val)
+
+        # Timeout reached.
+        self.assertRaises(
+            asyncio.TimeoutError,
+            utils.async_compat_call,
+            self.test_coro,
+            sleep=0.5, timeout=0.1)

watcher/tests/decision_engine/datasources/test_gnocchi_helper.py

@@ -40,17 +40,25 @@ class TestGnocchiHelper(base.BaseTestCase):
         self.addCleanup(stat_agg_patcher.stop)
 
     def test_gnocchi_statistic_aggregation(self, mock_gnocchi):
+        vcpus = 2
+        mock_instance = mock.Mock(
+            id='16a86790-327a-45f9-bc82-45839f062fdc',
+            vcpus=vcpus)
+
         gnocchi = mock.MagicMock()
+        # cpu time rate of change (ns)
+        mock_rate_measure = 360 * 10e+8 * vcpus * 5.5 / 100
         expected_result = 5.5
 
-        expected_measures = [["2017-02-02T09:00:00.000000", 360, 5.5]]
+        expected_measures = [
+            ["2017-02-02T09:00:00.000000", 360, mock_rate_measure]]
 
         gnocchi.metric.get_measures.return_value = expected_measures
         mock_gnocchi.return_value = gnocchi
 
         helper = gnocchi_helper.GnocchiHelper()
         result = helper.statistic_aggregation(
-            resource=mock.Mock(id='16a86790-327a-45f9-bc82-45839f062fdc'),
+            resource=mock_instance,
             resource_type='instance',
             meter_name='instance_cpu_usage',
             period=300,
@@ -59,6 +67,14 @@ class TestGnocchiHelper(base.BaseTestCase):
         )
         self.assertEqual(expected_result, result)
 
+        gnocchi.metric.get_measures.assert_called_once_with(
+            metric="cpu",
+            start=mock.ANY,
+            stop=mock.ANY,
+            resource_id=mock_instance.uuid,
+            granularity=360,
+            aggregation="rate:mean")
+
     def test_gnocchi_statistic_series(self, mock_gnocchi):
         gnocchi = mock.MagicMock()
         expected_result = {

watcher/tests/decision_engine/fake_metal_helper.py

@@ -0,0 +1,47 @@
+# Copyright (c) 2023 Cloudbase Solutions
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest import mock
+import uuid
+
+from watcher.common.metal_helper import constants as m_constants
+
+
+def get_mock_metal_node(node_id=None,
+                        power_state=m_constants.PowerState.ON,
+                        running_vms=0,
+                        hostname=None,
+                        compute_state='up'):
+    node_id = node_id or str(uuid.uuid4())
+    # NOTE(lpetrut): the hostname is important for some of the tests,
+    # which expect it to match the fake cluster model.
+    hostname = hostname or "compute-" + str(uuid.uuid4()).split('-')[0]
+
+    hypervisor_node_dict = {
+        'hypervisor_hostname': hostname,
+        'running_vms': running_vms,
+        'service': {
+            'host': hostname,
+        },
+        'state': compute_state,
+    }
+    hypervisor_node = mock.Mock(**hypervisor_node_dict)
+    hypervisor_node.to_dict.return_value = hypervisor_node_dict
+
+    node = mock.Mock()
+    node.get_power_state.return_value = power_state
+    node.get_id.return_value = uuid
+    node.get_hypervisor_node.return_value = hypervisor_node
+    return node

watcher/tests/decision_engine/model/faker_cluster_and_metrics.py

@@ -80,7 +80,7 @@ class FakerModelCollector(base.BaseClusterDataModelCollector):
         return self.load_model('scenario_4_with_metrics.xml')
 
 
-class FakeCeilometerMetrics(object):
+class FakeGnocchiMetrics(object):
     def __init__(self, model):
         self.model = model
 
@@ -90,6 +90,9 @@ class FakeCeilometerMetrics(object):
         if meter_name == 'host_cpu_usage':
             return self.get_compute_node_cpu_util(
                 resource, period, aggregate, granularity)
+        elif meter_name == 'host_ram_usage':
+            return self.get_compute_node_ram_util(
+                resource, period, aggregate, granularity)
         elif meter_name == 'instance_cpu_usage':
             return self.get_instance_cpu_util(
                 resource, period, aggregate, granularity)
@@ -110,109 +113,27 @@ class FakeCeilometerMetrics(object):
         Returns relative node CPU utilization <0, 100>.
         :param r_id: resource id
         """
-        node_uuid = '%s_%s' % (resource.uuid, resource.hostname)
+        node = self.model.get_node_by_uuid(resource.uuid)
-        node = self.model.get_node_by_uuid(node_uuid)
         instances = self.model.get_node_instances(node)
         util_sum = 0.0
-        for instance_uuid in instances:
+        for instance in instances:
-            instance = self.model.get_instance_by_uuid(instance_uuid)
             total_cpu_util = instance.vcpus * self.get_instance_cpu_util(
-                instance.uuid)
+                instance, period, aggregate, granularity)
             util_sum += total_cpu_util / 100.0
         util_sum /= node.vcpus
         return util_sum * 100.0
 
-    @staticmethod
+    def get_compute_node_ram_util(self, resource, period, aggregate,
-    def get_instance_cpu_util(resource, period, aggregate,
-                              granularity):
-        instance_cpu_util = dict()
-        instance_cpu_util['INSTANCE_0'] = 10
-        instance_cpu_util['INSTANCE_1'] = 30
-        instance_cpu_util['INSTANCE_2'] = 60
-        instance_cpu_util['INSTANCE_3'] = 20
-        instance_cpu_util['INSTANCE_4'] = 40
-        instance_cpu_util['INSTANCE_5'] = 50
-        instance_cpu_util['INSTANCE_6'] = 100
-        instance_cpu_util['INSTANCE_7'] = 100
-        instance_cpu_util['INSTANCE_8'] = 100
-        instance_cpu_util['INSTANCE_9'] = 100
-        return instance_cpu_util[str(resource.uuid)]
-
-    @staticmethod
-    def get_instance_ram_util(resource, period, aggregate,
-                              granularity):
-        instance_ram_util = dict()
-        instance_ram_util['INSTANCE_0'] = 1
-        instance_ram_util['INSTANCE_1'] = 2
-        instance_ram_util['INSTANCE_2'] = 4
-        instance_ram_util['INSTANCE_3'] = 8
-        instance_ram_util['INSTANCE_4'] = 3
-        instance_ram_util['INSTANCE_5'] = 2
-        instance_ram_util['INSTANCE_6'] = 1
-        instance_ram_util['INSTANCE_7'] = 2
-        instance_ram_util['INSTANCE_8'] = 4
-        instance_ram_util['INSTANCE_9'] = 8
-        return instance_ram_util[str(resource.uuid)]
-
-    @staticmethod
-    def get_instance_disk_root_size(resource, period, aggregate,
-                                    granularity):
-        instance_disk_util = dict()
-        instance_disk_util['INSTANCE_0'] = 10
-        instance_disk_util['INSTANCE_1'] = 15
-        instance_disk_util['INSTANCE_2'] = 30
-        instance_disk_util['INSTANCE_3'] = 35
-        instance_disk_util['INSTANCE_4'] = 20
-        instance_disk_util['INSTANCE_5'] = 25
-        instance_disk_util['INSTANCE_6'] = 25
-        instance_disk_util['INSTANCE_7'] = 25
-        instance_disk_util['INSTANCE_8'] = 25
-        instance_disk_util['INSTANCE_9'] = 25
-        return instance_disk_util[str(resource.uuid)]
-
-
-class FakeGnocchiMetrics(object):
-    def __init__(self, model):
-        self.model = model
-
-    def mock_get_statistics(self, resource=None, resource_type=None,
-                            meter_name=None, period=300, aggregate='mean',
-                            granularity=300):
-        if meter_name == 'host_cpu_usage':
-            return self.get_compute_node_cpu_util(
-                resource, period, aggregate, granularity)
-        elif meter_name == 'instance_cpu_usage':
-            return self.get_instance_cpu_util(
-                resource, period, aggregate, granularity)
-        elif meter_name == 'instance_ram_usage':
-            return self.get_instance_ram_util(
-                resource, period, aggregate, granularity)
-        elif meter_name == 'instance_root_disk_size':
-            return self.get_instance_disk_root_size(
-                resource, period, aggregate, granularity)
-
-    def get_compute_node_cpu_util(self, resource, period, aggregate,
                                   granularity):
-        """Calculates node utilization dynamicaly.
+        # Returns mock host ram usage in KB based on the allocated
-
+        # instances.
-        node CPU utilization should consider
+        node = self.model.get_node_by_uuid(resource.uuid)
-        and corelate with actual instance-node mappings
-        provided within a cluster model.
-        Returns relative node CPU utilization <0, 100>.
-
-        :param r_id: resource id
-        """
-        node_uuid = "%s_%s" % (resource.uuid, resource.hostname)
-        node = self.model.get_node_by_uuid(node_uuid)
         instances = self.model.get_node_instances(node)
         util_sum = 0.0
-        for instance_uuid in instances:
+        for instance in instances:
-            instance = self.model.get_instance_by_uuid(instance_uuid)
+            util_sum += self.get_instance_ram_util(
-            total_cpu_util = instance.vcpus * self.get_instance_cpu_util(
+                instance, period, aggregate, granularity)
-                instance.uuid)
+        return util_sum / 1024
-            util_sum += total_cpu_util / 100.0
-        util_sum /= node.vcpus
-        return util_sum * 100.0
 
     @staticmethod
     def get_instance_cpu_util(resource, period, aggregate,
@@ -261,3 +182,9 @@ class FakeGnocchiMetrics(object):
         instance_disk_util['INSTANCE_8'] = 25
         instance_disk_util['INSTANCE_9'] = 25
         return instance_disk_util[str(resource.uuid)]
+
+
+# TODO(lpetrut): consider dropping Ceilometer support, it was deprecated
+# in Ocata.
+class FakeCeilometerMetrics(FakeGnocchiMetrics):
+    pass

watcher/tests/decision_engine/strategy/strategies/test_saving_energy.py

@@ -18,8 +18,10 @@
 from unittest import mock
 
 from watcher.common import clients
+from watcher.common.metal_helper import constants as m_constants
 from watcher.common import utils
 from watcher.decision_engine.strategy import strategies
+from watcher.tests.decision_engine import fake_metal_helper
 from watcher.tests.decision_engine.strategy.strategies.test_base \
     import TestBaseStrategy
 
@@ -29,26 +31,15 @@ class TestSavingEnergy(TestBaseStrategy):
     def setUp(self):
         super(TestSavingEnergy, self).setUp()
 
-        mock_node1_dict = {
+        self.fake_nodes = [fake_metal_helper.get_mock_metal_node(),
-            'uuid': '922d4762-0bc5-4b30-9cb9-48ab644dd861'}
+                           fake_metal_helper.get_mock_metal_node()]
-        mock_node2_dict = {
+        self._metal_helper = mock.Mock()
-            'uuid': '922d4762-0bc5-4b30-9cb9-48ab644dd862'}
+        self._metal_helper.list_compute_nodes.return_value = self.fake_nodes
-        mock_node1 = mock.Mock(**mock_node1_dict)
-        mock_node2 = mock.Mock(**mock_node2_dict)
-        self.fake_nodes = [mock_node1, mock_node2]
 
-        p_ironic = mock.patch.object(
+        p_nova = mock.patch.object(clients.OpenStackClients, 'nova')
-            clients.OpenStackClients, 'ironic')
-        self.m_ironic = p_ironic.start()
-        self.addCleanup(p_ironic.stop)
-
-        p_nova = mock.patch.object(
-            clients.OpenStackClients, 'nova')
         self.m_nova = p_nova.start()
         self.addCleanup(p_nova.stop)
 
-        self.m_ironic.node.list.return_value = self.fake_nodes
-
         self.m_c_model.return_value = self.fake_c_cluster.generate_scenario_1()
 
         self.strategy = strategies.SavingEnergy(
@@ -59,27 +50,20 @@ class TestSavingEnergy(TestBaseStrategy):
              'min_free_hosts_num': 1})
         self.strategy.free_used_percent = 10.0
         self.strategy.min_free_hosts_num = 1
-        self.strategy._ironic_client = self.m_ironic
+        self.strategy._metal_helper = self._metal_helper
         self.strategy._nova_client = self.m_nova
 
     def test_get_hosts_pool_with_vms_node_pool(self):
-        mock_node1_dict = {
+        self._metal_helper.list_compute_nodes.return_value = [
-            'extra': {'compute_node_id': 1},
+            fake_metal_helper.get_mock_metal_node(
-            'power_state': 'power on'}
+                power_state=m_constants.PowerState.ON,
-        mock_node2_dict = {
+                hostname='hostname_0',
-            'extra': {'compute_node_id': 2},
+                running_vms=2),
-            'power_state': 'power off'}
+            fake_metal_helper.get_mock_metal_node(
-        mock_node1 = mock.Mock(**mock_node1_dict)
+                power_state=m_constants.PowerState.OFF,
-        mock_node2 = mock.Mock(**mock_node2_dict)
+                hostname='hostname_1',
-        self.m_ironic.node.get.side_effect = [mock_node1, mock_node2]
+                running_vms=2),
-
+        ]
-        mock_hyper1 = mock.Mock()
-        mock_hyper2 = mock.Mock()
-        mock_hyper1.to_dict.return_value = {
-            'running_vms': 2, 'service': {'host': 'hostname_0'}, 'state': 'up'}
-        mock_hyper2.to_dict.return_value = {
-            'running_vms': 2, 'service': {'host': 'hostname_1'}, 'state': 'up'}
-        self.m_nova.hypervisors.get.side_effect = [mock_hyper1, mock_hyper2]
 
         self.strategy.get_hosts_pool()
 
@@ -88,23 +72,16 @@ class TestSavingEnergy(TestBaseStrategy):
         self.assertEqual(len(self.strategy.free_poweroff_node_pool), 0)
 
     def test_get_hosts_pool_free_poweron_node_pool(self):
-        mock_node1_dict = {
+        self._metal_helper.list_compute_nodes.return_value = [
-            'extra': {'compute_node_id': 1},
+            fake_metal_helper.get_mock_metal_node(
-            'power_state': 'power on'}
+                power_state=m_constants.PowerState.ON,
-        mock_node2_dict = {
+                hostname='hostname_0',
-            'extra': {'compute_node_id': 2},
+                running_vms=0),
-            'power_state': 'power on'}
+            fake_metal_helper.get_mock_metal_node(
-        mock_node1 = mock.Mock(**mock_node1_dict)
+                power_state=m_constants.PowerState.ON,
-        mock_node2 = mock.Mock(**mock_node2_dict)
+                hostname='hostname_1',
-        self.m_ironic.node.get.side_effect = [mock_node1, mock_node2]
+                running_vms=0),
-
+        ]
-        mock_hyper1 = mock.Mock()
-        mock_hyper2 = mock.Mock()
-        mock_hyper1.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_0'}, 'state': 'up'}
-        mock_hyper2.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_1'}, 'state': 'up'}
-        self.m_nova.hypervisors.get.side_effect = [mock_hyper1, mock_hyper2]
 
         self.strategy.get_hosts_pool()
 
@@ -113,23 +90,16 @@ class TestSavingEnergy(TestBaseStrategy):
         self.assertEqual(len(self.strategy.free_poweroff_node_pool), 0)
 
     def test_get_hosts_pool_free_poweroff_node_pool(self):
-        mock_node1_dict = {
+        self._metal_helper.list_compute_nodes.return_value = [
-            'extra': {'compute_node_id': 1},
+            fake_metal_helper.get_mock_metal_node(
-            'power_state': 'power off'}
+                power_state=m_constants.PowerState.OFF,
-        mock_node2_dict = {
+                hostname='hostname_0',
-            'extra': {'compute_node_id': 2},
+                running_vms=0),
-            'power_state': 'power off'}
+            fake_metal_helper.get_mock_metal_node(
-        mock_node1 = mock.Mock(**mock_node1_dict)
+                power_state=m_constants.PowerState.OFF,
-        mock_node2 = mock.Mock(**mock_node2_dict)
+                hostname='hostname_1',
-        self.m_ironic.node.get.side_effect = [mock_node1, mock_node2]
+                running_vms=0),
-
+        ]
-        mock_hyper1 = mock.Mock()
-        mock_hyper2 = mock.Mock()
-        mock_hyper1.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_0'}, 'state': 'up'}
-        mock_hyper2.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_1'}, 'state': 'up'}
-        self.m_nova.hypervisors.get.side_effect = [mock_hyper1, mock_hyper2]
 
         self.strategy.get_hosts_pool()
 
@@ -138,26 +108,16 @@ class TestSavingEnergy(TestBaseStrategy):
         self.assertEqual(len(self.strategy.free_poweroff_node_pool), 2)
 
     def test_get_hosts_pool_with_node_out_model(self):
-        mock_node1_dict = {
+        self._metal_helper.list_compute_nodes.return_value = [
-            'extra': {'compute_node_id': 1},
+            fake_metal_helper.get_mock_metal_node(
-            'power_state': 'power off'}
+                power_state=m_constants.PowerState.OFF,
-        mock_node2_dict = {
+                hostname='hostname_0',
-            'extra': {'compute_node_id': 2},
+                running_vms=0),
-            'power_state': 'power off'}
+            fake_metal_helper.get_mock_metal_node(
-        mock_node1 = mock.Mock(**mock_node1_dict)
+                power_state=m_constants.PowerState.OFF,
-        mock_node2 = mock.Mock(**mock_node2_dict)
+                hostname='hostname_10',
-        self.m_ironic.node.get.side_effect = [mock_node1, mock_node2]
+                running_vms=0),
-
+        ]
-        mock_hyper1 = mock.Mock()
-        mock_hyper2 = mock.Mock()
-        mock_hyper1.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_0'},
-            'state': 'up'}
-        mock_hyper2.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_10'},
-            'state': 'up'}
-        self.m_nova.hypervisors.get.side_effect = [mock_hyper1, mock_hyper2]
-
         self.strategy.get_hosts_pool()
 
         self.assertEqual(len(self.strategy.with_vms_node_pool), 0)
@@ -166,9 +126,9 @@ class TestSavingEnergy(TestBaseStrategy):
 
     def test_save_energy_poweron(self):
         self.strategy.free_poweroff_node_pool = [
-            mock.Mock(uuid='922d4762-0bc5-4b30-9cb9-48ab644dd861'),
+            fake_metal_helper.get_mock_metal_node(),
-            mock.Mock(uuid='922d4762-0bc5-4b30-9cb9-48ab644dd862')
+            fake_metal_helper.get_mock_metal_node(),
-            ]
+        ]
         self.strategy.save_energy()
         self.assertEqual(len(self.strategy.solution.actions), 1)
         action = self.strategy.solution.actions[0]
@@ -185,23 +145,16 @@ class TestSavingEnergy(TestBaseStrategy):
         self.assertEqual(action.get('input_parameters').get('state'), 'off')
 
     def test_execute(self):
-        mock_node1_dict = {
+        self._metal_helper.list_compute_nodes.return_value = [
-            'extra': {'compute_node_id': 1},
+            fake_metal_helper.get_mock_metal_node(
-            'power_state': 'power on'}
+                power_state=m_constants.PowerState.ON,
-        mock_node2_dict = {
+                hostname='hostname_0',
-            'extra': {'compute_node_id': 2},
+                running_vms=0),
-            'power_state': 'power on'}
+            fake_metal_helper.get_mock_metal_node(
-        mock_node1 = mock.Mock(**mock_node1_dict)
+                power_state=m_constants.PowerState.ON,
-        mock_node2 = mock.Mock(**mock_node2_dict)
+                hostname='hostname_1',
-        self.m_ironic.node.get.side_effect = [mock_node1, mock_node2]
+                running_vms=0),
-
+        ]
-        mock_hyper1 = mock.Mock()
-        mock_hyper2 = mock.Mock()
-        mock_hyper1.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_0'}, 'state': 'up'}
-        mock_hyper2.to_dict.return_value = {
-            'running_vms': 0, 'service': {'host': 'hostname_1'}, 'state': 'up'}
-        self.m_nova.hypervisors.get.side_effect = [mock_hyper1, mock_hyper2]
 
         model = self.fake_c_cluster.generate_scenario_1()
         self.m_c_model.return_value = model

watcher/tests/decision_engine/strategy/strategies/test_vm_workload_consolidation.py

@@ -64,6 +64,10 @@ class TestVMWorkloadConsolidation(TestBaseStrategy):
                 self.fake_metrics.get_instance_ram_util),
             get_instance_root_disk_size=(
                 self.fake_metrics.get_instance_disk_root_size),
+            get_host_cpu_usage=(
+                self.fake_metrics.get_compute_node_cpu_util),
+            get_host_ram_usage=(
+                self.fake_metrics.get_compute_node_ram_util)
             )
         self.strategy = strategies.VMWorkloadConsolidation(
             config=mock.Mock(datasources=self.datasource))
@@ -88,6 +92,71 @@ class TestVMWorkloadConsolidation(TestBaseStrategy):
             node_util,
             self.strategy.get_node_utilization(node_0))
 
+    def test_get_node_utilization_using_host_metrics(self):
+        model = self.fake_c_cluster.generate_scenario_1()
+        self.m_c_model.return_value = model
+        self.fake_metrics.model = model
+        node_0 = model.get_node_by_uuid("Node_0")
+
+        # "get_node_utilization" is expected to return the maximum
+        # between the host metrics and the sum of the instance metrics.
+        data_src = self.m_datasource.return_value
+        cpu_usage = 30
+        data_src.get_host_cpu_usage = mock.Mock(return_value=cpu_usage)
+        data_src.get_host_ram_usage = mock.Mock(return_value=512 * 1024)
+
+        exp_cpu_usage = cpu_usage * node_0.vcpus / 100
+        exp_node_util = dict(cpu=exp_cpu_usage, ram=512, disk=10)
+        self.assertEqual(
+            exp_node_util,
+            self.strategy.get_node_utilization(node_0))
+
+    def test_get_node_utilization_after_migrations(self):
+        model = self.fake_c_cluster.generate_scenario_1()
+        self.m_c_model.return_value = model
+        self.fake_metrics.model = model
+        node_0 = model.get_node_by_uuid("Node_0")
+        node_1 = model.get_node_by_uuid("Node_1")
+
+        data_src = self.m_datasource.return_value
+        cpu_usage = 30
+        host_ram_usage_mb = 512
+        data_src.get_host_cpu_usage = mock.Mock(return_value=cpu_usage)
+        data_src.get_host_ram_usage = mock.Mock(
+            return_value=host_ram_usage_mb * 1024)
+
+        instance_uuid = 'INSTANCE_0'
+        instance = model.get_instance_by_uuid(instance_uuid)
+        self.strategy.add_migration(instance, node_0, node_1)
+
+        instance_util = self.strategy.get_instance_utilization(instance)
+
+        # Ensure that we take into account planned migrations when
+        # determining node utilization
+        exp_node_0_cpu_usage = (
+            cpu_usage * node_0.vcpus) / 100 - instance_util['cpu']
+        exp_node_1_cpu_usage = (
+            cpu_usage * node_1.vcpus) / 100 + instance_util['cpu']
+
+        exp_node_0_ram_usage = host_ram_usage_mb - instance.memory
+        exp_node_1_ram_usage = host_ram_usage_mb + instance.memory
+
+        exp_node_0_util = dict(
+            cpu=exp_node_0_cpu_usage,
+            ram=exp_node_0_ram_usage,
+            disk=0)
+        exp_node_1_util = dict(
+            cpu=exp_node_1_cpu_usage,
+            ram=exp_node_1_ram_usage,
+            disk=25)
+
+        self.assertEqual(
+            exp_node_0_util,
+            self.strategy.get_node_utilization(node_0))
+        self.assertEqual(
+            exp_node_1_util,
+            self.strategy.get_node_utilization(node_1))
+
     def test_get_node_capacity(self):
         model = self.fake_c_cluster.generate_scenario_1()
         self.m_c_model.return_value = model
@@ -113,7 +182,8 @@ class TestVMWorkloadConsolidation(TestBaseStrategy):
         expected_cru = {'cpu': 0.05, 'disk': 0.05, 'ram': 0.0234375}
         self.assertEqual(expected_cru, cru)
 
-    def test_add_migration_with_active_state(self):
+    def _test_add_migration(self, instance_state, expect_migration=True,
+                            expected_migration_type="live"):
         model = self.fake_c_cluster.generate_scenario_1()
         self.m_c_model.return_value = model
         self.fake_metrics.model = model
@@ -121,38 +191,36 @@ class TestVMWorkloadConsolidation(TestBaseStrategy):
         n2 = model.get_node_by_uuid('Node_1')
         instance_uuid = 'INSTANCE_0'
         instance = model.get_instance_by_uuid(instance_uuid)
+        instance.state = instance_state
         self.strategy.add_migration(instance, n1, n2)
-        self.assertEqual(1, len(self.strategy.solution.actions))
+
-        expected = {'action_type': 'migrate',
+        if expect_migration:
-                    'input_parameters': {'destination_node': n2.hostname,
+            self.assertEqual(1, len(self.strategy.solution.actions))
-                                         'source_node': n1.hostname,
+
-                                         'migration_type': 'live',
+            expected = {'action_type': 'migrate',
-                                         'resource_id': instance.uuid,
+                        'input_parameters': {
-                                         'resource_name': instance.name}}
+                            'destination_node': n2.hostname,
-        self.assertEqual(expected, self.strategy.solution.actions[0])
+                            'source_node': n1.hostname,
+                            'migration_type': expected_migration_type,
+                            'resource_id': instance.uuid,
+                            'resource_name': instance.name}}
+            self.assertEqual(expected, self.strategy.solution.actions[0])
+        else:
+            self.assertEqual(0, len(self.strategy.solution.actions))
+
+    def test_add_migration_with_active_state(self):
+        self._test_add_migration(element.InstanceState.ACTIVE.value)
 
     def test_add_migration_with_paused_state(self):
-        model = self.fake_c_cluster.generate_scenario_1()
+        self._test_add_migration(element.InstanceState.PAUSED.value)
-        self.m_c_model.return_value = model
-        self.fake_metrics.model = model
-        n1 = model.get_node_by_uuid('Node_0')
-        n2 = model.get_node_by_uuid('Node_1')
-        instance_uuid = 'INSTANCE_0'
-        instance = model.get_instance_by_uuid(instance_uuid)
-        setattr(instance, 'state', element.InstanceState.ERROR.value)
-        self.strategy.add_migration(instance, n1, n2)
-        self.assertEqual(0, len(self.strategy.solution.actions))
 
-        setattr(instance, 'state', element.InstanceState.PAUSED.value)
+    def test_add_migration_with_error_state(self):
-        self.strategy.add_migration(instance, n1, n2)
+        self._test_add_migration(element.InstanceState.ERROR.value,
-        self.assertEqual(1, len(self.strategy.solution.actions))
+                                 expect_migration=False)
-        expected = {'action_type': 'migrate',
+
-                    'input_parameters': {'destination_node': n2.hostname,
+    def test_add_migration_with_stopped_state(self):
-                                         'source_node': n1.hostname,
+        self._test_add_migration(element.InstanceState.STOPPED.value,
-                                         'migration_type': 'live',
+                                 expected_migration_type="cold")
-                                         'resource_id': instance.uuid,
-                                         'resource_name': instance.name}}
-        self.assertEqual(expected, self.strategy.solution.actions[0])
 
     def test_is_overloaded(self):
         model = self.fake_c_cluster.generate_scenario_1()

watcher/tests/objects/test_objects.py

@@ -531,6 +531,7 @@ class TestRegistry(test_base.TestCase):
 
     @mock.patch('watcher.objects.base.objects')
     def test_hook_chooses_newer_properly(self, mock_objects):
+        mock_objects.MyObj.VERSION = MyObj.VERSION
         reg = base.WatcherObjectRegistry()
         reg.registration_hook(MyObj, 0)
 
@@ -547,6 +548,7 @@ class TestRegistry(test_base.TestCase):
 
     @mock.patch('watcher.objects.base.objects')
     def test_hook_keeps_newer_properly(self, mock_objects):
+        mock_objects.MyObj.VERSION = MyObj.VERSION
         reg = base.WatcherObjectRegistry()
         reg.registration_hook(MyObj, 0)