Move queue declaration to project level

This moves the watcher queue declaration from the pipeline level
(where it is no longer valid) to the project level.

https: //lists.openstack.org/pipermail/openstack-discuss/2022-May/028603.html
Change-Id: I06923abb00f7eecd59587f44cd1f6a069e88a9fc
(cherry picked from commit 6003322711)

.gitreview

@@ -1,4 +1,5 @@
 [gerrit]
-host=review.openstack.org
+host=review.opendev.org
 port=29418
 project=openstack/watcher.git
+defaultbranch=stable/stein

.zuul.yaml

@@ -1,4 +1,5 @@
 - project:
+    queue: watcher
     templates:
       - check-requirements
       - openstack-cover-jobs
@@ -11,19 +12,18 @@
     check:
       jobs:
         - watcher-tempest-functional
-        - watcher-grenade
+        - watcher-grenade:
+            voting: false
         - watcher-tempest-dummy_optim
         - watcher-tempest-actuator
         - watcher-tempest-basic_optim
         - watcher-tempest-vm_workload_consolidation
         - watcher-tempest-workload_balancing
-        - watcherclient-tempest-functional
         - watcher-tempest-zone_migration
         - watcher-tempest-host_maintenance
         - watcher-tempest-storage_balance
         - watcher-tls-test
     gate:
-      queue: watcher
       jobs:
         - watcher-tempest-functional
 
@@ -109,7 +109,7 @@
 - job:
     name: watcher-tempest-multinode
     parent: watcher-tempest-functional
-    nodeset: openstack-two-node
+    nodeset: openstack-two-node-bionic
     roles:
       - zuul: openstack/tempest
     group-vars:
@@ -156,7 +156,7 @@
               live_migration: true
               block_migration_for_live_migration: true
       devstack_plugins:
-        ceilometer: https://git.openstack.org/openstack/ceilometer
+        ceilometer: https://opendev.org/openstack/ceilometer
 
 - job:
     name: watcher-tempest-functional
@@ -164,7 +164,7 @@
     timeout: 7200
     required-projects:
       - openstack/ceilometer
-      - openstack-infra/devstack-gate
+      - openstack/devstack-gate
       - openstack/python-openstackclient
       - openstack/python-watcherclient
       - openstack/watcher
@@ -172,7 +172,7 @@
       - openstack/tempest
     vars:
       devstack_plugins:
-        watcher: https://git.openstack.org/openstack/watcher
+        watcher: https://opendev.org/openstack/watcher
       devstack_services:
         tls-proxy: false
         watcher-api: true
@@ -211,8 +211,8 @@
       - ^tools/.*$
       - ^tox.ini$
     required-projects:
-      - openstack-dev/grenade
+      - openstack/grenade
-      - openstack-infra/devstack-gate
+      - openstack/devstack-gate
       - openstack/watcher
       - openstack/python-watcherclient
       - openstack/watcher-tempest-plugin

devstack/local.conf.compute

@@ -35,7 +35,7 @@ VNCSERVER_PROXYCLIENT_ADDRESS=$HOST_IP
 NOVA_INSTANCES_PATH=/opt/stack/data/instances
 
 # Enable the Ceilometer plugin for the compute agent
-enable_plugin ceilometer git://git.openstack.org/openstack/ceilometer
+enable_plugin ceilometer https://git.openstack.org/openstack/ceilometer
 disable_service ceilometer-acentral,ceilometer-collector,ceilometer-api
 
 LOGFILE=$DEST/logs/stack.sh.log

devstack/local.conf.controller

@@ -25,13 +25,13 @@ MULTI_HOST=1
 disable_service n-cpu
 
 # Enable the Watcher Dashboard plugin
-enable_plugin watcher-dashboard git://git.openstack.org/openstack/watcher-dashboard
+enable_plugin watcher-dashboard https://git.openstack.org/openstack/watcher-dashboard
 
 # Enable the Watcher plugin
-enable_plugin watcher git://git.openstack.org/openstack/watcher
+enable_plugin watcher https://git.openstack.org/openstack/watcher
 
 # Enable the Ceilometer plugin
-enable_plugin ceilometer git://git.openstack.org/openstack/ceilometer
+enable_plugin ceilometer https://git.openstack.org/openstack/ceilometer
 
 # This is the controller node, so disable the ceilometer compute agent
 disable_service ceilometer-acompute

doc/requirements.txt

@@ -2,7 +2,8 @@
 # of appearance. Changing the order has an impact on the overall integration
 # process, which may cause wedges in the gate later.
 openstackdocstheme>=1.20.0 # Apache-2.0
-sphinx>=1.6.5,!=1.6.6,!=1.6.7 # BSD
+sphinx>=1.6.5,!=1.6.6,!=1.6.7,<2.0.0;python_version=='2.7' # BSD
+sphinx>=1.6.5,!=1.6.6,!=1.6.7;python_version>='3.4' # BSD
 sphinxcontrib-pecanwsme>=0.8.0 # Apache-2.0
 reno>=2.7.0 # Apache-2.0
 sphinxcontrib-apidoc>=0.2.0  # BSD

doc/source/configuration/configuring.rst

@@ -178,7 +178,7 @@ You can easily generate and update a sample configuration file
 named :ref:`watcher.conf.sample <watcher_sample_configuration_files>` by using
 these following commands::
 
-    $ git clone git://git.openstack.org/openstack/watcher
+    $ git clone https://git.openstack.org/openstack/watcher
     $ cd watcher/
     $ tox -e genconfig
     $ vi etc/watcher/watcher.conf.sample

doc/source/contributor/devstack.rst

@@ -19,7 +19,7 @@ model. To enable the Watcher plugin with DevStack, add the following to the
 `[[local|localrc]]` section of your controller's `local.conf` to enable the
 Watcher plugin::
 
-    enable_plugin watcher git://git.openstack.org/openstack/watcher
+    enable_plugin watcher https://git.openstack.org/openstack/watcher
 
 For more detailed instructions, see `Detailed DevStack Instructions`_. Check
 out the `DevStack documentation`_ for more information regarding DevStack.

playbooks/legacy/grenade-devstack-watcher/run.yaml

@@ -13,12 +13,12 @@
           set -x
           cat > clonemap.yaml << EOF
           clonemap:
-            - name: openstack-infra/devstack-gate
+            - name: openstack/devstack-gate
               dest: devstack-gate
           EOF
           /usr/zuul-env/bin/zuul-cloner -m clonemap.yaml --cache-dir /opt/git \
-              git://git.openstack.org \
+              https://opendev.org \
-              openstack-infra/devstack-gate
+              openstack/devstack-gate
         executable: /bin/bash
         chdir: '{{ ansible_user_dir }}/workspace'
       environment: '{{ zuul | zuul_legacy_vars }}'
@@ -29,13 +29,13 @@
           set -x
           export PYTHONUNBUFFERED=true
 
-          export PROJECTS="openstack-dev/grenade $PROJECTS"
+          export PROJECTS="openstack/grenade $PROJECTS"
           export PROJECTS="openstack/watcher $PROJECTS"
           export PROJECTS="openstack/watcher-tempest-plugin $PROJECTS"
           export PROJECTS="openstack/python-watcherclient $PROJECTS"
           export DEVSTACK_PROJECT_FROM_GIT="python-watcherclient $DEVSTACK_PROJECT_FROM_GIT"
 
-          export GRENADE_PLUGINRC="enable_grenade_plugin watcher https://git.openstack.org/openstack/watcher"
+          export GRENADE_PLUGINRC="enable_grenade_plugin watcher https://opendev.org/openstack/watcher"
           export DEVSTACK_LOCAL_CONFIG+=$'\n'"export TEMPEST_PLUGINS='/opt/stack/new/watcher-tempest-plugin'"
 
           export DEVSTACK_GATE_TEMPEST_NOTESTS=1

requirements.txt

@@ -45,5 +45,7 @@ stevedore>=1.28.0 # Apache-2.0
 taskflow>=3.1.0 # Apache-2.0
 WebOb>=1.7.4 # MIT
 WSME>=0.9.2 # MIT
-networkx>=1.11 # BSD
+# NOTE(fdegir): NetworkX 2.3 dropped support for Python 2
+networkx>=1.11,<2.3;python_version<'3.0'  # BSD
+networkx>=1.11;python_version>='3.4' # BSD
 microversion_parse>=0.2.1 # Apache-2.0

test-requirements.txt

@@ -13,4 +13,4 @@ testscenarios>=0.5.0 # Apache-2.0/BSD
 testtools>=2.3.0 # MIT
 stestr>=2.0.0 # Apache-2.0
 os-api-ref>=1.4.0 # Apache-2.0
-bandit>=1.1.0 # Apache-2.0
+bandit>=1.6.0 # Apache-2.0

tox.ini

@@ -7,7 +7,7 @@ skipsdist = True
 usedevelop = True
 whitelist_externals = find
                       rm
-install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
+install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/stein} {opts} {packages}
 setenv =
    VIRTUAL_ENV={envdir}
 deps = -r{toxinidir}/test-requirements.txt
@@ -22,7 +22,7 @@ basepython = python3
 commands =
     doc8 doc/source/ CONTRIBUTING.rst HACKING.rst README.rst
     flake8
-    bandit -r watcher -x tests -n5 -ll -s B320
+    bandit -r watcher -x watcher/tests/* -n5 -ll -s B320
 
 [testenv:venv]
 basepython = python3
@@ -98,7 +98,7 @@ commands = sphinx-build -a -W -E -d releasenotes/build/doctrees -b html releasen
 [testenv:bandit]
 basepython = python3
 deps = -r{toxinidir}/test-requirements.txt
-commands = bandit -r watcher -x tests -n5 -ll -s B320
+commands = bandit -r watcher -x watcher/tests/* -n5 -ll -s B320
 
 [testenv:lower-constraints]
 basepython = python3

watcher/cmd/__init__.py

@@ -0,0 +1,20 @@
+#
+# 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.
+
+# NOTE(licanwei): Do eventlet monkey patching here, instead of in
+# common/service.py.  This allows the API service to run without monkey
+# patching under Apache (which uses its own concurrency model). Mixing
+# concurrency models can cause undefined behavior and potentially API timeouts.
+import eventlet
+
+eventlet.monkey_patch()

watcher/common/config.py

@@ -21,12 +21,15 @@ from watcher.common import rpc
 from watcher import version
 
 
-def parse_args(argv, default_config_files=None):
+def parse_args(argv, default_config_files=None, default_config_dirs=None):
     default_config_files = (default_config_files or
                             cfg.find_config_files(project='watcher'))
+    default_config_dirs = (default_config_dirs or
+                           cfg.find_config_dirs(project='watcher'))
     rpc.set_defaults(control_exchange='watcher')
     cfg.CONF(argv[1:],
              project='python-watcher',
              version=version.version_info.release_string(),
+             default_config_dirs=default_config_dirs,
              default_config_files=default_config_files)
     rpc.init(cfg.CONF)

watcher/common/nova_helper.py

@@ -76,9 +76,25 @@ class NovaHelper(object):
             LOG.exception(exc)
             raise exception.ComputeNodeNotFound(name=node_hostname)
 
-    def get_instance_list(self):
+    def get_instance_list(self, filters=None, limit=-1):
-        return self.nova.servers.list(search_opts={'all_tenants': True},
+        """List servers for all tenants with details.
-                                      limit=-1)
+
+        This always gets servers with the all_tenants=True filter.
+
+        :param filters: dict of additional filters (optional).
+        :param limit: Maximum number of servers to return (optional).
+                      If limit == -1, all servers will be returned,
+                      note that limit == -1 will have a performance
+                      penalty. For details, please see:
+                      https://bugs.launchpad.net/watcher/+bug/1834679
+        :returns: list of novaclient Server objects
+        """
+        search_opts = {'all_tenants': True}
+        if filters:
+            search_opts.update(filters)
+        # TODO(chenker) Add marker param to list Server objects.
+        return self.nova.servers.list(search_opts=search_opts,
+                                      limit=limit)
 
     def get_flavor_list(self):
         return self.nova.flavors.list(**{'is_public': None})

watcher/common/service.py

@@ -17,7 +17,6 @@
 import datetime
 import socket
 
-import eventlet
 from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_log import _options
@@ -42,12 +41,6 @@ from watcher.objects import base
 from watcher.objects import fields as wfields
 from watcher import version
 
-# NOTE:
-# Ubuntu 14.04 forces librabbitmq when kombu is used
-# Unfortunately it forces a version that has a crash
-# bug.  Calling eventlet.monkey_patch() tells kombu
-# to use libamqp instead.
-eventlet.monkey_patch()
 
 NOTIFICATION_OPTS = [
     cfg.StrOpt('notification_level',

watcher/datasource/ceilometer.py

@@ -18,7 +18,6 @@
 
 import datetime
 
-from ceilometerclient import exc
 from oslo_log import log
 from oslo_utils import timeutils
 
@@ -31,6 +30,13 @@ from watcher.datasource import base
 LOG = log.getLogger(__name__)
 
 
+try:
+    from ceilometerclient import exc
+    HAS_CEILCLIENT = True
+except ImportError:
+    HAS_CEILCLIENT = False
+
+
 class CeilometerHelper(base.DataSourceBase):
 
     NAME = 'ceilometer'
@@ -130,15 +136,15 @@ class CeilometerHelper(base.DataSourceBase):
             self.osc.reset_clients()
             self.ceilometer = self.osc.ceilometer()
             return f(*args, **kargs)
-        except Exception:
+        except Exception as e:
-            raise
+            LOG.exception(e)
 
     def check_availability(self):
-        try:
+        status = self.query_retry(self.ceilometer.resources.list)
-            self.query_retry(self.ceilometer.resources.list)
+        if status:
-        except Exception:
+            return 'available'
+        else:
             return 'not available'
-        return 'available'
 
     def query_sample(self, meter_name, query, limit=1):
         return self.query_retry(f=self.ceilometer.samples.list,
@@ -156,9 +162,8 @@ class CeilometerHelper(base.DataSourceBase):
 
     def list_metrics(self):
         """List the user's meters."""
-        try:
+        meters = self.query_retry(f=self.ceilometer.meters.list)
-            meters = self.query_retry(f=self.ceilometer.meters.list)
+        if not meters:
-        except Exception:
             return set()
         else:
             return meters

watcher/datasource/gnocchi.py

@@ -24,7 +24,6 @@ from oslo_config import cfg
 from oslo_log import log
 
 from watcher.common import clients
-from watcher.common import exception
 from watcher.common import utils as common_utils
 from watcher.datasource import base
 
@@ -60,20 +59,18 @@ class GnocchiHelper(base.DataSourceBase):
             except Exception as e:
                 LOG.exception(e)
                 time.sleep(CONF.gnocchi_client.query_timeout)
-        raise exception.DataSourceNotAvailable(datasource='gnocchi')
 
     def check_availability(self):
-        try:
+        status = self.query_retry(self.gnocchi.status.get)
-            self.query_retry(self.gnocchi.status.get)
+        if status:
-        except Exception:
+            return 'available'
+        else:
             return 'not available'
-        return 'available'
 
     def list_metrics(self):
         """List the user's meters."""
-        try:
+        response = self.query_retry(f=self.gnocchi.metric.list)
-            response = self.query_retry(f=self.gnocchi.metric.list)
+        if not response:
-        except Exception:
             return set()
         else:
             return set([metric['name'] for metric in response])
@@ -106,8 +103,9 @@ class GnocchiHelper(base.DataSourceBase):
                 f=self.gnocchi.resource.search, **kwargs)
 
             if not resources:
-                raise exception.ResourceNotFound(name='gnocchi',
+                LOG.warning("The {0} resource {1} could not be "
-                                                 id=resource_id)
+                            "found".format(self.NAME, resource_id))
+                return
 
             resource_id = resources[0]['id']
 

watcher/datasource/monasca.py

@@ -19,11 +19,14 @@
 import datetime
 
 from monascaclient import exc
+from oslo_log import log
 
 from watcher.common import clients
 from watcher.common import exception
 from watcher.datasource import base
 
+LOG = log.getLogger(__name__)
+
 
 class MonascaHelper(base.DataSourceBase):
 
@@ -53,8 +56,8 @@ class MonascaHelper(base.DataSourceBase):
             self.osc.reset_clients()
             self.monasca = self.osc.monasca()
             return f(*args, **kwargs)
-        except Exception:
+        except Exception as e:
-            raise
+            LOG.exception(e)
 
     def _format_time_params(self, start_time, end_time, period):
         """Format time-related params to the correct Monasca format
@@ -78,11 +81,11 @@ class MonascaHelper(base.DataSourceBase):
         return start_timestamp, end_timestamp, period
 
     def check_availability(self):
-        try:
+        status = self.query_retry(self.monasca.metrics.list)
-            self.query_retry(self.monasca.metrics.list)
+        if status:
-        except Exception:
+            return 'available'
+        else:
             return 'not available'
-        return 'available'
 
     def list_metrics(self):
         # TODO(alexchadin): this method should be implemented in accordance to

watcher/decision_engine/model/collector/nova.py

@@ -15,7 +15,6 @@
 
 from oslo_log import log
 
-from watcher.common import exception
 from watcher.common import nova_helper
 from watcher.decision_engine.model.collector import base
 from watcher.decision_engine.model import element
@@ -208,10 +207,9 @@ class ModelBuilder(object):
         self.osc = osc
         self.model = None
         self.model_scope = dict()
+        self.no_model_scope_flag = False
         self.nova = osc.nova()
         self.nova_helper = nova_helper.NovaHelper(osc=self.osc)
-        # self.neutron = osc.neutron()
-        # self.cinder = osc.cinder()
 
     def _collect_aggregates(self, host_aggregates, _nodes):
         aggregate_list = self.nova_helper.get_aggregate_list()
@@ -237,7 +235,7 @@ class ModelBuilder(object):
             include_all_nodes = True
         for service in service_list:
             if service.zone in zone_names or include_all_nodes:
-                _nodes.update(service.host)
+                _nodes.add(service.host)
 
     def _add_physical_layer(self):
         """Add the physical layer of the graph.
@@ -254,22 +252,31 @@ class ModelBuilder(object):
             self._collect_zones(availability_zones, compute_nodes)
 
         if not compute_nodes:
+            self.no_model_scope_flag = True
             all_nodes = self.nova_helper.get_compute_node_list()
             compute_nodes = set(
                 [node.hypervisor_hostname for node in all_nodes])
         LOG.debug("compute nodes: %s", compute_nodes)
         for node_name in compute_nodes:
+            # TODO(mriedem): Change this to list hypervisors with details
+            # so we don't have to call get_compute_node_by_id. It requires
+            # changes to python-novaclient.
             cnode = self.nova_helper.get_compute_node_by_name(node_name,
                                                               servers=True)
             if cnode:
-                self.add_compute_node(cnode[0])
+                # Get the node details (like the service.host).
-                self.add_instance_node(cnode[0])
+                node_info = self.nova_helper.get_compute_node_by_id(
+                    cnode[0].id)
+                self.add_compute_node(node_info)
+                # node.servers is a list of server objects
+                # New in nova version 2.53
+                instances = getattr(cnode[0], "servers", None)
+                self.add_instance_node(node_info, instances)
 
     def add_compute_node(self, node):
         # Build and add base node.
-        node_info = self.nova_helper.get_compute_node_by_id(node.id)
+        LOG.debug("node info: %s", node)
-        LOG.debug("node info: %s", node_info)
+        compute_node = self.build_compute_node(node)
-        compute_node = self.build_compute_node(node_info)
         self.model.add_node(compute_node)
 
         # NOTE(v-francoise): we can encapsulate capabilities of the node
@@ -315,99 +322,28 @@ class ModelBuilder(object):
         #                                 node_attributes)
         return compute_node
 
-    # def _build_network_compute_node(self, base_node):
+    def add_instance_node(self, node, instances):
-    #     attributes = {}
-    #     net_node = self._build_node("physical", "network", "NIC", attributes)
-    #     net_id = "{}_network".format(base_node)
-    #     return net_id, net_node
-
-    # def build_disk_compute_node(self, base_node, compute):
-    #     # Build disk node attributes.
-    #     disk_attributes = {
-    #         "size_gb": compute.local_gb,
-    #         "used_gb": compute.local_gb_used,
-    #         "available_gb": compute.free_disk_gb}
-    #     disk_node = self._build_node("physical", "storage", "disk",
-    #                                  disk_attributes)
-    #     disk_id = "{}_disk".format(base_node)
-    #     return disk_id, disk_node
-
-    # def build_memory_compute_node(self, base_node, compute):
-    #     # Build memory node attributes.
-    #     memory_attrs = {"size_mb": compute.memory_mb,
-    #                     "used_mb": compute.memory_mb_used,
-    #                     "available_mb": compute.free_ram_mb}
-    #     memory_node = self._build_node("physical", "memory", "memory",
-    #                                    memory_attrs)
-    #     memory_id = "{}_memory".format(base_node)
-    #     return memory_id, memory_node
-
-    # def build_cpu_compute_node(self, base_node, compute):
-    #     # Build memory node attributes.
-    #     cpu_attributes = {"vcpus": compute.vcpus,
-    #                       "vcpus_used": compute.vcpus_used,
-    #                       "info": jsonutils.loads(compute.cpu_info)}
-    #     cpu_node = self._build_node("physical", "cpu", "cpu", cpu_attributes)
-    #     cpu_id = "{}_cpu".format(base_node)
-    #     return cpu_id, cpu_node
-
-    # @staticmethod
-    # def _build_node(layer, category, node_type, attributes):
-    #     return {"layer": layer, "category": category, "type": node_type,
-    #             "attributes": attributes}
-
-    def _add_virtual_layer(self):
-        """Add the virtual layer to the graph.
-
-        This layer is the virtual components of the infrastructure,
-        such as vms.
-        """
-        self._add_virtual_servers()
-        # self._add_virtual_network()
-        # self._add_virtual_storage()
-
-    def _add_virtual_servers(self):
-        all_instances = self.nova_helper.get_instance_list()
-        for inst in all_instances:
-            # Add Node
-            instance = self._build_instance_node(inst)
-            self.model.add_instance(instance)
-            # Get the cnode_name uuid.
-            cnode_uuid = getattr(inst, "OS-EXT-SRV-ATTR:host")
-            if cnode_uuid is None:
-                # The instance is not attached to any Compute node
-                continue
-            try:
-                # Nova compute node
-                # cnode = self.nova_helper.get_compute_node_by_hostname(
-                #     cnode_uuid)
-                compute_node = self.model.get_node_by_uuid(
-                    cnode_uuid)
-                # Connect the instance to its compute node
-                self.model.map_instance(instance, compute_node)
-            except exception.ComputeNodeNotFound:
-                continue
-
-    def add_instance_node(self, node):
-        # node.servers is a list of server objects
-        # New in nova version 2.53
-        instances = getattr(node, "servers", None)
         if instances is None:
             # no instances on this node
             return
-        instance_uuids = [s['uuid'] for s in instances]
+        host = node.service["host"]
-        for uuid in instance_uuids:
+        compute_node = self.model.get_node_by_uuid(host)
-            try:
+        filters = {'host': host}
-                inst = self.nova_helper.find_instance(uuid)
+        limit = len(instances) if len(instances) <= 1000 else -1
-            except Exception as exc:
+        # Get all servers on this compute host.
-                LOG.exception(exc)
+        # Note that the advantage of passing the limit parameter is
-                continue
+        # that it can speed up the call time of novaclient. 1000 is
+        # the default maximum number of return servers provided by
+        # compute API. If we need to request more than 1000 servers,
+        # we can set limit=-1. For details, please see:
+        # https://bugs.launchpad.net/watcher/+bug/1834679
+        instances = self.nova_helper.get_instance_list(
+            filters=filters,
+            limit=limit)
+        for inst in instances:
             # Add Node
             instance = self._build_instance_node(inst)
             self.model.add_instance(instance)
-            cnode_uuid = getattr(inst, "OS-EXT-SRV-ATTR:host")
-            compute_node = self.model.get_node_by_uuid(
-                cnode_uuid)
             # Connect the instance to its compute node
             self.model.map_instance(instance, compute_node)
 
@@ -439,137 +375,23 @@ class ModelBuilder(object):
         # node_attributes["attributes"] = instance_attributes
         return element.Instance(**instance_attributes)
 
-    # def _add_virtual_storage(self):
-    #     try:
-    #         volumes = self.cinder.volumes.list()
-    #     except Exception:
-    #         return
-    #     for volume in volumes:
-    #         volume_id, volume_node = self._build_storage_node(volume)
-    #         self.add_node(volume_id, volume_node)
-    #         host = self._get_volume_host_id(volume_node)
-    #         self.add_edge(volume_id, host)
-    #         # Add connections to an instance.
-    #         if volume_node['attributes']['attachments']:
-    #             for attachment in volume_node['attributes']['attachments']:
-    #                 self.add_edge(volume_id, attachment['server_id'],
-    #                               label='ATTACHED_TO')
-    #             volume_node['attributes'].pop('attachments')
-
-    # def _add_virtual_network(self):
-    #     try:
-    #         routers = self.neutron.list_routers()
-    #     except Exception:
-    #         return
-
-    #     for network in self.neutron.list_networks()['networks']:
-    #         self.add_node(*self._build_network(network))
-
-    #     for router in routers['routers']:
-    #         self.add_node(*self._build_router(router))
-
-    #     router_interfaces, _, compute_ports = self._group_ports()
-    #     for router_interface in router_interfaces:
-    #         interface = self._build_router_interface(router_interface)
-    #         router_interface_id = interface[0]
-    #         router_interface_node = interface[1]
-    #         router_id = interface[2]
-    #         self.add_node(router_interface_id, router_interface_node)
-    #         self.add_edge(router_id, router_interface_id)
-    #         network_id = router_interface_node['attributes']['network_id']
-    #         self.add_edge(router_interface_id, network_id)
-
-    #     for compute_port in compute_ports:
-    #         cp_id, cp_node, instance_id = self._build_compute_port_node(
-    #             compute_port)
-    #         self.add_node(cp_id, cp_node)
-    #         self.add_edge(cp_id, vm_id)
-    #         net_id = cp_node['attributes']['network_id']
-    #         self.add_edge(net_id, cp_id)
-    #         # Connect port to physical node
-    #         phys_net_node = "{}_network".format(cp_node['attributes']
-    #                                             ['binding:host_id'])
-    #         self.add_edge(cp_id, phys_net_node)
-
-    # def _get_volume_host_id(self, volume_node):
-    #     host = volume_node['attributes']['os-vol-host-attr:host']
-    #     if host.find('@') != -1:
-    #         host = host.split('@')[0]
-    #     elif host.find('#') != -1:
-    #         host = host.split('#')[0]
-    #     return "{}_disk".format(host)
-
-    # def _build_storage_node(self, volume_obj):
-    #     volume = volume_obj.__dict__
-    #     volume["name"] = volume["id"]
-    #     volume.pop("id")
-    #     volume.pop("manager")
-    #     node = self._build_node("virtual", "storage", 'volume', volume)
-    #     return volume["name"], node
-
-    # def _build_compute_port_node(self, compute_port):
-    #     compute_port["name"] = compute_port["id"]
-    #     compute_port.pop("id")
-    #     nde_type = "{}_port".format(
-    #         compute_port["device_owner"].split(":")[0])
-    #     compute_port.pop("device_owner")
-    #     device_id = compute_port["device_id"]
-    #     compute_port.pop("device_id")
-    #     node = self._build_node("virtual", "network", nde_type, compute_port)
-    #     return compute_port["name"], node, device_id
-
-    # def _group_ports(self):
-    #     router_interfaces = []
-    #     floating_ips = []
-    #     compute_ports = []
-    #     interface_types = ["network:router_interface",
-    #                        'network:router_gateway']
-
-    #     for port in self.neutron.list_ports()['ports']:
-    #         if port['device_owner'] in interface_types:
-    #             router_interfaces.append(port)
-    #         elif port['device_owner'].startswith('compute:'):
-    #             compute_ports.append(port)
-    #         elif port['device_owner'] == 'network:floatingip':
-    #             floating_ips.append(port)
-
-    #     return router_interfaces, floating_ips, compute_ports
-
-    # def _build_router_interface(self, interface):
-    #     interface["name"] = interface["id"]
-    #     interface.pop("id")
-    #     node_type = interface["device_owner"].split(":")[1]
-    #     node = self._build_node("virtual", "network", node_type, interface)
-    #     return interface["name"], node, interface["device_id"]
-
-    # def _build_router(self, router):
-    #     router_attrs = {"uuid": router['id'],
-    #                     "name": router['name'],
-    #                     "state": router['status']}
-    #     node = self._build_node('virtual', 'network', 'router', router_attrs)
-    #     return str(router['id']), node
-
-    # def _build_network(self, network):
-    #     node = self._build_node('virtual', 'network', 'network', network)
-    #     return network['id'], node
-
     def _merge_compute_scope(self, compute_scope):
         model_keys = self.model_scope.keys()
         update_flag = False
 
         role_keys = ("host_aggregates", "availability_zones")
         for role in compute_scope:
-            role_key = role.keys()[0]
+            role_key = list(role.keys())[0]
             if role_key not in role_keys:
                 continue
-            role_values = role.values()[0]
+            role_values = list(role.values())[0]
             if role_key in model_keys:
                 for value in role_values:
                     if value not in self.model_scope[role_key]:
-                        self.model_scope[role_key].sppend(value)
+                        self.model_scope[role_key].append(value)
                         update_flag = True
             else:
-                self.self.model_scope[role_key] = role_values
+                self.model_scope[role_key] = role_values
                 update_flag = True
         return update_flag
 
@@ -581,10 +403,9 @@ class ModelBuilder(object):
                 compute_scope = _scope['compute']
                 break
 
-        if self.model_scope:
+        if self.no_model_scope_flag is False:
-            if model_scope:
+            if compute_scope:
-                if compute_scope:
+                update_flag = self._merge_compute_scope(compute_scope)
-                    update_flag = self._merge_compute_scope(compute_scope)
             else:
                 self.model_scope = dict()
                 update_flag = True

watcher/decision_engine/model/notification/nova.py

@@ -256,7 +256,7 @@ class VersionedNotification(NovaNotification):
         'instance.rescue.end': instance_updated,
         'instance.update': instance_updated,
         'instance.live_migration_force_complete.end': instance_updated,
-        'instance.live_migration_post_dest.end': instance_updated,
+        'instance.live_migration_post.end': instance_updated,
         'instance.delete.end': instance_deleted,
         'instance.soft_delete.end': instance_deleted,
         'service.create': service_updated,
@@ -279,5 +279,10 @@ class VersionedNotification(NovaNotification):
                       metadata=metadata))
         func = self.notification_mapping.get(event_type)
         if func:
-            LOG.debug(payload)
+            # The nova CDM is not built until an audit is performed.
-            func(self, payload)
+            if self.cluster_data_model:
+                LOG.debug(payload)
+                func(self, payload)
+            else:
+                LOG.debug('Nova CDM has not yet been built; ignoring '
+                          'notifications until an audit is performed.')

watcher/decision_engine/strategy/strategies/workload_stabilization.py

@@ -271,6 +271,7 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
                     "for %(metric_name)s", dict(
                         resource_id=instance.uuid, metric_name=meter))
                 return
+            # cpu_util has been deprecated since Stein.
             if meter == 'cpu_util':
                 avg_meter /= float(100)
             LOG.debug('Load of %(metric)s for %(instance)s is %(value)s',
@@ -324,6 +325,9 @@ class WorkloadStabilization(base.WorkloadStabilizationBaseStrategy):
                         avg_meter /= oslo_utils.units.Ki
                     if meter_name == 'compute.node.cpu.percent':
                         avg_meter /= 100
+                    # hardware.cpu.util has been deprecated since Stein.
+                    if meter_name == 'hardware.cpu.util':
+                        avg_meter /= 100
                 LOG.debug('Load of %(metric)s for %(node)s is %(value)s',
                           {'metric': metric,
                            'node': node_id,

watcher/tests/common/test_nova_helper.py

@@ -146,6 +146,22 @@ class TestNovaHelper(base.TestCase):
             nova_util.get_compute_node_by_hostname,
             hypervisor_name)
 
+    def test_get_instance_list(self, *args):
+        nova_util = nova_helper.NovaHelper()
+        # Call it once with no filters.
+        with mock.patch.object(nova_util, 'nova') as nova_mock:
+            result = nova_util.get_instance_list()
+            nova_mock.servers.list.assert_called_once_with(
+                search_opts={'all_tenants': True}, limit=-1)
+            self.assertIs(result, nova_mock.servers.list.return_value)
+        # Call it again with filters.
+        with mock.patch.object(nova_util, 'nova') as nova_mock:
+            result = nova_util.get_instance_list(filters={'host': 'fake-host'})
+            nova_mock.servers.list.assert_called_once_with(
+                search_opts={'all_tenants': True, 'host': 'fake-host'},
+                limit=-1)
+            self.assertIs(result, nova_mock.servers.list.return_value)
+
     @mock.patch.object(time, 'sleep', mock.Mock())
     def test_stop_instance(self, mock_glance, mock_cinder, mock_neutron,
                            mock_nova):

watcher/tests/decision_engine/cluster/test_model_builder.py

@@ -0,0 +1,175 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2019 European Organization for Nuclear Research (CERN)
+#
+# Authors: Corne Lukken <info@dantalion.nl>
+#
+# 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 mock
+
+from oslo_config import cfg
+from oslo_log import log
+
+from watcher.common import nova_helper
+from watcher.decision_engine.model.collector import nova
+from watcher.tests import base
+
+CONF = cfg.CONF
+LOG = log.getLogger(__name__)
+
+
+class TestModelBuilder(base.BaseTestCase):
+    """Test the collector ModelBuilder
+
+        Objects under test are preceded with t_ and mocked objects are preceded
+        with m_ , additionally, patched objects are preceded with p_ no object
+        under test should be created in setUp this can influence the results.
+        """
+
+    def setUp(self):
+        super(TestModelBuilder, self).setUp()
+
+    def test_check_model(self):
+        """Initialize collector ModelBuilder and test check model"""
+
+        m_scope = [{"compute": [
+            {"host_aggregates": [{"id": 5}]},
+            {"availability_zones": [{"name": "av_a"}]}
+        ]}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        self.assertTrue(t_nova_cluster._check_model_scope(m_scope))
+
+    def test_check_model_update_false(self):
+        """Initialize check model with multiple identical scopes
+
+        The seconds check_model should return false as the models are the same
+        """
+
+        m_scope = [{"compute": [
+            {"host_aggregates": [{"id": 5}]},
+            {"availability_zones": [{"name": "av_a"}]}
+        ]}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        self.assertTrue(t_nova_cluster._check_model_scope(m_scope))
+        self.assertFalse(t_nova_cluster._check_model_scope(m_scope))
+
+    def test_check_model_update_true(self):
+        """Initialize check model with multiple different scopes
+
+        Since the models differ both should return True for the update flag
+        """
+
+        m_scope_one = [{"compute": [
+            {"host_aggregates": [{"id": 5}]},
+            {"availability_zones": [{"name": "av_a"}]}
+        ]}]
+
+        m_scope_two = [{"compute": [
+            {"host_aggregates": [{"id": 2}]},
+            {"availability_zones": [{"name": "av_b"}]}
+        ]}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        self.assertTrue(t_nova_cluster._check_model_scope(m_scope_one))
+        self.assertTrue(t_nova_cluster._check_model_scope(m_scope_two))
+
+    def test_merge_compute_scope(self):
+        """"""
+
+        m_scope_one = [
+            {"host_aggregates": [{"id": 5}]},
+            {"availability_zones": [{"name": "av_a"}]}
+        ]
+
+        m_scope_two = [
+            {"host_aggregates": [{"id": 4}]},
+            {"availability_zones": [{"name": "av_b"}]}
+        ]
+
+        reference = {'availability_zones':
+                     [{'name': 'av_a'}, {'name': 'av_b'}],
+                     'host_aggregates':
+                     [{'id': 5}, {'id': 4}]}
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        t_nova_cluster._merge_compute_scope(m_scope_one)
+        t_nova_cluster._merge_compute_scope(m_scope_two)
+
+        self.assertEqual(reference, t_nova_cluster.model_scope)
+
+    @mock.patch.object(nova_helper, 'NovaHelper')
+    def test_collect_aggregates(self, m_nova):
+        """"""
+
+        m_nova.return_value.get_aggregate_list.return_value = \
+            [mock.Mock(id=1, name='example'),
+             mock.Mock(id=5, name='example', hosts=['hostone', 'hosttwo'])]
+
+        m_nova.return_value.get_compute_node_by_name.return_value = False
+
+        m_scope = [{'id': 5}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        result = set()
+        t_nova_cluster._collect_aggregates(m_scope, result)
+
+        self.assertEqual(set(['hostone', 'hosttwo']), result)
+
+    @mock.patch.object(nova_helper, 'NovaHelper')
+    def test_collect_zones(self, m_nova):
+        """"""
+
+        m_nova.return_value.get_service_list.return_value = \
+            [mock.Mock(zone='av_b'),
+             mock.Mock(zone='av_a', host='hostone')]
+
+        m_nova.return_value.get_compute_node_by_name.return_value = False
+
+        m_scope = [{'name': 'av_a'}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        result = set()
+        t_nova_cluster._collect_zones(m_scope, result)
+
+        self.assertEqual(set(['hostone']), result)
+
+    @mock.patch.object(nova_helper, 'NovaHelper')
+    def test_add_physical_layer(self, m_nova):
+        """"""
+
+        m_nova.return_value.get_aggregate_list.return_value = \
+            [mock.Mock(id=1, name='example'),
+             mock.Mock(id=5, name='example', hosts=['hostone', 'hosttwo'])]
+
+        m_nova.return_value.get_service_list.return_value = \
+            [mock.Mock(zone='av_b', host='hostthree'),
+             mock.Mock(zone='av_a', host='hostone')]
+
+        m_nova.return_value.get_compute_node_by_name.return_value = False
+
+        m_scope = [{"compute": [
+            {"host_aggregates": [{"id": 5}]},
+            {"availability_zones": [{"name": "av_a"}]}
+        ]}]
+
+        t_nova_cluster = nova.ModelBuilder(mock.Mock())
+        t_nova_cluster.execute(m_scope)
+        m_nova.return_value.get_compute_node_by_name.assert_any_call(
+            'hostone', servers=True)
+        m_nova.return_value.get_compute_node_by_name.assert_any_call(
+            'hosttwo', servers=True)
+        self.assertEqual(
+            m_nova.return_value.get_compute_node_by_name.call_count, 2)

watcher/tests/decision_engine/cluster/test_nova_cdmc.py

@@ -43,23 +43,30 @@ class TestNovaClusterDataModelCollector(base.TestCase):
             state='up',
             disabled_reason='',
         )
-
+        minimal_node = dict(
-        fake_compute_node = mock.Mock(
             id=1337,
+            hypervisor_hostname='test_hostname',
+            state='TEST_STATE',
+            status='TEST_STATUS',
+        )
+        minimal_node_with_servers = dict(
+            servers=[
+                {'name': 'fake_instance',
+                 'uuid': 'ef500f7e-dac8-470f-960c-169486fce71b'}
+            ],
+            **minimal_node
+        )
+        fake_compute_node = mock.Mock(
             service={'id': 123, 'host': 'test_hostname',
                      'disabled_reason': ''},
-            hypervisor_hostname='test_hostname',
             memory_mb=333,
             free_disk_gb=222,
             local_gb=111,
             vcpus=4,
-            state='TEST_STATE',
+            servers=None,  # Don't let the mock return a value for servers.
-            status='TEST_STATUS',
+            **minimal_node
-            servers=[
-                {'name': 'fake_instance',
-                 'uuid': 'ef500f7e-dac8-470f-960c-169486fce71b'}
-                ]
         )
+        fake_compute_node_with_servers = mock.Mock(**minimal_node_with_servers)
         fake_instance = mock.Mock(
             id='ef500f7e-dac8-470f-960c-169486fce71b',
             human_id='fake_instance',
@@ -68,12 +75,14 @@ class TestNovaClusterDataModelCollector(base.TestCase):
             tenant_id='ff560f7e-dbc8-771f-960c-164482fce21b',
         )
         setattr(fake_instance, 'OS-EXT-STS:vm_state', 'VM_STATE')
-        setattr(fake_instance, 'OS-EXT-SRV-ATTR:host', 'test_hostname')
+        # Returns the hypervisors with details (service) but no servers.
         m_nova_helper.get_compute_node_list.return_value = [fake_compute_node]
+        # Returns the hypervisor with servers but no details (service).
         m_nova_helper.get_compute_node_by_name.return_value = [
-            fake_compute_node]
+            fake_compute_node_with_servers]
+        # Returns the hypervisor with details (service) but no servers.
         m_nova_helper.get_compute_node_by_id.return_value = fake_compute_node
-        m_nova_helper.find_instance.return_value = fake_instance
+        m_nova_helper.get_instance_list.return_value = [fake_instance]
 
         m_config = mock.Mock()
         m_osc = mock.Mock()
@@ -95,3 +104,28 @@ class TestNovaClusterDataModelCollector(base.TestCase):
 
         self.assertEqual(node.uuid, 'test_hostname')
         self.assertEqual(instance.uuid, 'ef500f7e-dac8-470f-960c-169486fce71b')
+
+        m_nova_helper.get_instance_list.assert_called_once_with(
+            filters={'host': fake_compute_node.service['host']}, limit=1)
+
+
+class TestModelBuilder(base.TestCase):
+
+    @mock.patch.object(nova_helper, 'NovaHelper', mock.MagicMock())
+    def test_add_instance_node(self):
+        model_builder = nova.ModelBuilder(osc=mock.MagicMock())
+        model_builder.model = mock.MagicMock()
+        mock_node = mock.MagicMock()
+        mock_host = mock_node.service["host"]
+        mock_instances = [mock.MagicMock()]
+        model_builder.add_instance_node(mock_node, mock_instances)
+        # verify that when len(instances) <= 1000, limit == len(instance).
+        model_builder.nova_helper.get_instance_list.assert_called_once_with(
+            filters={'host': mock_host}, limit=1)
+
+        # verify that when len(instances) > 1000, limit == -1.
+        mock_instance = mock.Mock()
+        mock_instances = [mock_instance] * 1001
+        model_builder.add_instance_node(mock_node, mock_instances)
+        model_builder.nova_helper.get_instance_list.assert_called_with(
+            filters={'host': mock_host}, limit=-1)

watcher/tests/decision_engine/model/notification/data/instance-live_migration_post-end.json

@@ -1,5 +1,5 @@
 {
-    "event_type": "instance.live_migration_post_dest.end",
+    "event_type": "instance.live_migration_post.end",
     "payload": {
         "nova_object.data": {
             "action_initiator_project": "6f70656e737461636b20342065766572",

watcher/tests/decision_engine/model/notification/test_nova_notifications.py

@@ -69,8 +69,8 @@ class TestReceiveNovaNotifications(NotificationTestCase):
         'instance.update': 'instance-update.json',
         'instance.live_migration_force_complete.end':
         'instance-live_migration_force_complete-end.json',
-        'instance.live_migration_post_dest.end':
+        'instance.live_migration_post.end':
-        'instance-live_migration_post_dest-end.json',
+        'instance-live_migration_post-end.json',
         'instance.delete.end': 'instance-delete-end.json',
         'instance.soft_delete.end': 'instance-soft_delete-end.json',
         'service.create': 'service-create.json',
@@ -437,7 +437,7 @@ class TestNovaNotifications(NotificationTestCase):
         node = compute_model.get_node_by_instance_uuid(instance0_uuid)
         self.assertEqual('Node_0', node.uuid)
         message = self.load_message(
-            'instance-live_migration_post_dest-end.json')
+            'instance-live_migration_post-end.json')
         handler.info(
             ctxt=self.context,
             publisher_id=message['publisher_id'],
@@ -743,3 +743,21 @@ class TestNovaNotifications(NotificationTestCase):
 
         self.assertEqual(
             element.InstanceState.SUSPENDED.value, instance0.state)
+
+    def test_info_no_cdm(self):
+        # Tests that a notification is received before an audit has been
+        # performed which would create the nova CDM.
+        mock_collector = mock.Mock(cluster_data_model=None)
+        handler = novanotification.VersionedNotification(mock_collector)
+        payload = {
+            'nova_object.data': {
+                'uuid': '9966d6bd-a45c-4e1c-9d57-3054899a3ec7',
+                'host': None
+            }
+        }
+        with mock.patch.object(handler, 'update_instance') as update_instance:
+            handler.info(mock.sentinel.ctxt, 'publisher_id', 'instance.update',
+                         payload, metadata={})
+            # update_instance should not be called since we did not add an
+            # Instance object to the CDM since the CDM does not exist yet.
+            update_instance.assert_not_called()