Adds tooz locking driver

Adds tooz requirements, configuration options and locking support.

Story: 2001842
Task: 30376

Change-Id: I78ffdeeb502cbed0550152b19efbb224f47c65af

ironic_inspector/cmd/all.py

@@ -17,6 +17,7 @@ import sys
 from oslo_config import cfg
 from oslo_service import service
 
+from ironic_inspector.common.i18n import _
 from ironic_inspector.common.rpc_service import RPCService
 from ironic_inspector.common import service_utils
 from ironic_inspector import wsgi_service
@@ -28,6 +29,11 @@ def main(args=sys.argv[1:]):
     # Parse config file and command line options, then start logging
     service_utils.prepare_service(args)
 
+    if not CONF.standalone:
+        msg = _('To run ironic-inspector in standalone mode, '
+                '[DEFAULT]standalone should be set to True.')
+        sys.exit(msg)
+
     launcher = service.ServiceLauncher(CONF, restart_method='mutate')
     launcher.launch_service(wsgi_service.WSGIService())
     launcher.launch_service(RPCService(CONF.host))

ironic_inspector/common/locking.py

@@ -14,8 +14,12 @@
 import abc
 
 from oslo_concurrency import lockutils
+from oslo_config import cfg
 import six
 
+from ironic_inspector import utils
+
+CONF = cfg.CONF
 _LOCK_TEMPLATE = 'node-%s'
 _SEMAPHORES = lockutils.Semaphores()
 
@@ -24,7 +28,7 @@ _SEMAPHORES = lockutils.Semaphores()
 class BaseLock(object):
 
     @abc.abstractmethod
-    def acquire(self, blocking=True, timeout=None):
+    def acquire(self, blocking=True):
         """Acquire lock."""
 
     @abc.abstractmethod
@@ -44,8 +48,7 @@ class InternalLock(BaseLock):
                                              semaphores=_SEMAPHORES)
         self._locked = False
 
-    def acquire(self, blocking=True, timeout=None):
-        # NOTE(kaifeng) timeout is only available on python3
+    def acquire(self, blocking=True):
         if not self._locked:
             self._locked = self._lock.acquire(blocking=blocking)
         return self._locked
@@ -66,5 +69,36 @@ class InternalLock(BaseLock):
         self._lock.release()
 
 
+class ToozLock(BaseLock):
+    """Locking mechanism based on tooz."""
+
+    def __init__(self, coordinator, uuid, prefix='ironic_inspector_'):
+        name = (prefix + uuid).encode()
+        self._lock = coordinator.get_lock(name)
+
+    def acquire(self, blocking=True):
+        if not self._lock.acquired:
+            self._lock.acquire(blocking=blocking)
+        return self._lock.acquired
+
+    def release(self):
+        if self._lock.acquired:
+            self._lock.release()
+
+    def is_locked(self):
+        return self._lock.acquired
+
+    def __enter__(self):
+        self._lock.acquire()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._lock.release()
+
+
 def get_lock(uuid):
-    return InternalLock(uuid)
+    if CONF.standalone:
+        return InternalLock(uuid)
+
+    coordinator = utils.get_coordinator()
+    return ToozLock(coordinator, uuid)

ironic_inspector/conductor/manager.py

@@ -21,6 +21,7 @@ from ironic_lib import mdns
 from oslo_config import cfg
 from oslo_log import log
 import oslo_messaging as messaging
+from oslo_utils import excutils
 from oslo_utils import reflection
 
 from ironic_inspector.common.i18n import _
@@ -95,6 +96,18 @@ class ConductorManager(object):
             self._zeroconf.register_service('baremetal-introspection',
                                             endpoint)
 
+        if not CONF.standalone:
+            try:
+                coordinator = utils.get_coordinator()
+                coordinator.start()
+            except Exception:
+                with excutils.save_and_reraise_exception():
+                    LOG.critical('Failed when connecting to coordination '
+                                 'backend.')
+                    self.del_host()
+            else:
+                LOG.info('Successfully connected to coordination backend.')
+
     def del_host(self):
 
         if not self._shutting_down.acquire(blocking=False):
@@ -119,6 +132,15 @@ class ConductorManager(object):
             self._zeroconf = None
 
         self._shutting_down.release()
+
+        if not CONF.standalone:
+            try:
+                coordinator = utils.get_coordinator()
+                if coordinator and coordinator.is_started:
+                    coordinator.stop()
+            except Exception:
+                LOG.exception('Failed to stop coordinator')
+
         LOG.info('Shut down successfully')
 
     def _periodics_watchdog(self, callable_, activity, spacing, exc_info,

ironic_inspector/conf/__init__.py

@@ -13,6 +13,7 @@
 from oslo_config import cfg
 
 from ironic_inspector.conf import capabilities
+from ironic_inspector.conf import coordination
 from ironic_inspector.conf import default
 from ironic_inspector.conf import discovery
 from ironic_inspector.conf import dnsmasq_pxe_filter
@@ -29,6 +30,7 @@ CONF = cfg.CONF
 
 
 capabilities.register_opts(CONF)
+coordination.register_opts(CONF)
 discovery.register_opts(CONF)
 default.register_opts(CONF)
 dnsmasq_pxe_filter.register_opts(CONF)

ironic_inspector/conf/coordination.py

@@ -0,0 +1,35 @@
+# 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 ironic_inspector.common.i18n import _
+
+
+# NOTE(kaifeng) The capability of various backend varies, please check tooz
+# documentation for driver compatibilities:
+# https://docs.openstack.org/tooz/latest/user/compatibility.html
+_OPTS = [
+    cfg.StrOpt('backend_url',
+               default='memcached://localhost:11211',
+               help=_('The backend URL to use for distributed coordination. '
+                      'EXPERIMENTAL.')),
+]
+
+
+def register_opts(conf):
+    conf.register_opts(_OPTS, 'coordination')
+
+
+def list_opts():
+    return _OPTS

ironic_inspector/conf/default.py

@@ -79,6 +79,9 @@ _OPTS = [
     cfg.BoolOpt('enable_mdns', default=False,
                 help=_('Whether to enable publishing the ironic-inspector API '
                        'endpoint via multicast DNS.')),
+    cfg.BoolOpt('standalone', default=True,
+                help=_('Whether to run ironic-inspector as a standalone '
+                       'service. It\'s EXPERIMENTAL to set to False.'))
 ]
 
 

ironic_inspector/conf/opts.py

@@ -58,6 +58,7 @@ def parse_args(args, default_config_files=None):
 def list_opts():
     return [
         ('capabilities', ironic_inspector.conf.capabilities.list_opts()),
+        ('coordination', ironic_inspector.conf.coordination.list_opts()),
         ('DEFAULT', ironic_inspector.conf.default.list_opts()),
         ('discovery', ironic_inspector.conf.discovery.list_opts()),
         ('dnsmasq_pxe_filter',

ironic_inspector/test/unit/test_locking.py

@@ -0,0 +1,119 @@
+# 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 ironic_inspector.common import locking
+from ironic_inspector.test import base as test_base
+from ironic_inspector import utils
+
+CONF = cfg.CONF
+
+
+@mock.patch.object(locking, 'InternalLock', autospec=True)
+@mock.patch.object(locking, 'ToozLock', autospec=True)
+class TestGetLock(test_base.NodeTest):
+    def test_get_lock_internal(self, mock_tooz, mock_internal):
+        locking.get_lock(self.node.uuid)
+
+        mock_internal.assert_called_once_with(self.node.uuid)
+        mock_tooz.assert_not_called()
+
+    @mock.patch.object(utils, 'get_coordinator', autospec=True)
+    def test_get_lock_tooz(self, mock_get_coord, mock_tooz, mock_internal):
+        CONF.set_override('standalone', False)
+        coordinator = mock.MagicMock()
+        mock_get_coord.return_value = coordinator
+
+        locking.get_lock(self.node.uuid)
+
+        mock_tooz.assert_called_once_with(coordinator, self.node.uuid)
+        mock_internal.assert_not_called()
+
+
+class TestInternalLock(test_base.NodeTest):
+    def setUp(self):
+        super(TestInternalLock, self).setUp()
+        self.mock_lock = mock.MagicMock()
+        self.mock_lock.acquire.return_value = True
+
+    @mock.patch.object(locking, 'lockutils', autospec=True)
+    def test_init_lock(self, mock_lockutils):
+        locking.InternalLock(self.node.uuid)
+        mock_lockutils.internal_lock.assert_called_with(
+            'node-%s' % self.node.uuid, mock.ANY)
+
+    def test_acquire(self):
+        lock = locking.InternalLock(self.node.uuid)
+        lock._lock = self.mock_lock
+        lock.acquire()
+        self.mock_lock.acquire.assert_called_once_with(blocking=True)
+
+    def test_release(self):
+        lock = locking.ToozLock(mock.MagicMock(), self.node.uuid)
+        lock._lock = self.mock_lock
+        self.mock_lock._locked = True
+
+        lock.release()
+
+        self.mock_lock.release.assert_called_once_with()
+
+    def test_context(self):
+        lock = locking.InternalLock(self.node.uuid)
+        lock._lock = self.mock_lock
+
+        with lock:
+            self.mock_lock.acquire.assert_called_once_with()
+
+        self.mock_lock.release.assert_called_once_with()
+
+
+class TestToozLock(test_base.NodeTest):
+    def setUp(self):
+        super(TestToozLock, self).setUp()
+        self.mock_lock = mock.MagicMock()
+        self.mock_lock.acquire.return_value = True
+        self.mock_lock.acquired = False
+
+    def test_lock_default_prefix(self):
+        mock_coordinator = mock.MagicMock()
+        locking.ToozLock(mock_coordinator, self.node.uuid)
+        mock_coordinator.get_lock.assert_called_once_with(
+            str.encode('ironic_inspector_%s' % self.node.uuid))
+
+    def test_acquire(self):
+        lock = locking.ToozLock(mock.MagicMock(), self.node.uuid)
+        lock._lock = self.mock_lock
+
+        lock.acquire()
+
+        self.mock_lock.acquire.assert_called_once_with(blocking=True)
+
+    def test_release(self):
+        self.mock_lock.acquired = True
+        lock = locking.ToozLock(mock.MagicMock(), self.node.uuid)
+        lock._lock = self.mock_lock
+
+        lock.release()
+
+        self.mock_lock.release.assert_called_once_with()
+
+    def test_context(self):
+        lock = locking.ToozLock(mock.MagicMock(), self.node.uuid)
+        lock._lock = self.mock_lock
+
+        with lock:
+            self.mock_lock.acquire.assert_called_once_with()
+
+        self.mock_lock.release.assert_called_once_with()

ironic_inspector/test/unit/test_manager.py

@@ -139,6 +139,37 @@ class TestManagerInitHost(BaseManagerTest):
         mock_zc.return_value.register_service.assert_called_once_with(
             'baremetal-introspection', mock_endpoint.return_value)
 
+    @mock.patch.object(utils, 'get_coordinator', autospec=True)
+    @mock.patch.object(keystone, 'get_endpoint', autospec=True)
+    def test_init_host_with_coordinator(self, mock_endpoint, mock_get_coord):
+        CONF.set_override('standalone', False)
+        mock_coordinator = mock.MagicMock()
+        mock_get_coord.return_value = mock_coordinator
+        self.manager.init_host()
+        self.mock_db_init.assert_called_once_with()
+        self.mock_validate_processing_hooks.assert_called_once_with()
+        self.mock_filter.init_filter.assert_called_once_with()
+        self.assert_periodics()
+        mock_get_coord.assert_called_once_with()
+        mock_coordinator.start.assert_called_once_with()
+
+    @mock.patch.object(manager.ConductorManager, 'del_host')
+    @mock.patch.object(utils, 'get_coordinator', autospec=True)
+    @mock.patch.object(keystone, 'get_endpoint', autospec=True)
+    def test_init_host_with_coordinator_failed(self, mock_endpoint,
+                                               mock_get_coord, mock_del_host):
+        CONF.set_override('standalone', False)
+        mock_get_coord.side_effect = (utils.Error('Reaching coordination '
+                                                  'backend failed.'),
+                                      None)
+        self.assertRaises(utils.Error, self.manager.init_host)
+        self.mock_db_init.assert_called_once_with()
+        self.mock_validate_processing_hooks.assert_called_once_with()
+        self.mock_filter.init_filter.assert_called_once_with()
+        self.assert_periodics()
+        mock_get_coord.assert_called_once_with()
+        mock_del_host.assert_called_once_with()
+
 
 class TestManagerDelHost(BaseManagerTest):
     def setUp(self):
@@ -251,6 +282,26 @@ class TestManagerDelHost(BaseManagerTest):
         self.mock_filter.tear_down_filter.assert_called_once_with()
         self.mock__shutting_down.release.assert_called_once_with()
 
+    @mock.patch.object(utils, 'get_coordinator', autospec=True)
+    def test_del_host_with_coordinator(self, mock_get_coord):
+        CONF.set_override('standalone', False)
+        mock_coordinator = mock.MagicMock()
+        mock_coordinator.is_started = True
+        mock_get_coord.return_value = mock_coordinator
+
+        self.manager.del_host()
+
+        self.assertIsNone(self.manager._zeroconf)
+        self.mock__shutting_down.acquire.assert_called_once_with(
+            blocking=False)
+        self.mock__periodic_worker.stop.assert_called_once_with()
+        self.mock__periodic_worker.wait.assert_called_once_with()
+        self.assertIsNone(self.manager._periodics_worker)
+        self.mock_executor.shutdown.assert_called_once_with(wait=True)
+        self.mock_filter.tear_down_filter.assert_called_once_with()
+        self.mock__shutting_down.release.assert_called_once_with()
+        mock_coordinator.stop.called_once_with()
+
 
 class TestManagerPeriodicWatchDog(BaseManagerTest):
     def setUp(self):

ironic_inspector/test/unit/test_utils.py

@@ -154,3 +154,14 @@ class TestIsoTimestamp(base.BaseTest):
 
     def test_none(self):
         self.assertIsNone(utils.iso_timestamp(None))
+
+
+@mock.patch.object(utils, 'coordination', autospec=True)
+class TestGetCoordinator(base.BaseTest):
+    def test_get(self, mock_coordination):
+        CONF.set_override('backend_url', 'etcd3://1.2.3.4:2379',
+                          'coordination')
+        CONF.set_override('host', '1.2.3.5')
+        utils.get_coordinator()
+        mock_coordination.get_coordinator.assert_called_once_with(
+            'etcd3://1.2.3.4:2379', b'1.2.3.5')

ironic_inspector/utils.py

@@ -21,6 +21,7 @@ from oslo_config import cfg
 from oslo_log import log
 from oslo_middleware import cors as cors_middleware
 import pytz
+from tooz import coordination
 
 from ironic_inspector.common.i18n import _
 from ironic_inspector import policy
@@ -247,3 +248,14 @@ def iso_timestamp(timestamp=None, tz=pytz.timezone('utc')):
         return None
     date = datetime.datetime.fromtimestamp(timestamp, tz=tz)
     return date.isoformat()
+
+
+_COORDINATOR = None
+
+
+def get_coordinator():
+    global _COORDINATOR
+    if _COORDINATOR is None:
+        _COORDINATOR = coordination.get_coordinator(
+            CONF.coordination.backend_url, str.encode(CONF.host))
+    return _COORDINATOR

lower-constraints.txt

@@ -123,6 +123,7 @@ testrepository==0.0.20
 testresources==2.0.0
 testscenarios==0.4
 testtools==2.3.0
+tooz==1.64.0
 traceback2==1.4.0
 unittest2==1.1.0
 urllib3==1.22

requirements.txt

@@ -35,3 +35,4 @@ retrying!=1.3.0,>=1.2.3 # Apache-2.0
 six>=1.10.0 # MIT
 stevedore>=1.20.0 # Apache-2.0
 SQLAlchemy!=1.1.5,!=1.1.6,!=1.1.7,!=1.1.8,>=1.0.10 # MIT
+tooz>=1.64.0 # Apache-2.0