Replace remaining eventlet usage with AsyncIO

This also moves to use Threading backend for oslo.service and
oslo.messaging replacing the deprecated Eventlet backend.

Change-Id: I0d74e02a63cfa479c5a75457edb046e0b9824dd5
Signed-off-by: Matt Crees <mattc@stackhpc.com>

blazar/__init__.py

@@ -0,0 +1,22 @@
+# 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 warnings
+
+from oslo_service import backend as oslo_service_backend
+
+try:
+    oslo_service_backend.init_backend(
+        oslo_service_backend.BackendType.THREADING)
+except oslo_service_backend.exceptions.BackendAlreadySelected:
+    warnings.warn('The selected oslo_service backend is "eventlet"')

blazar/cmd/manager.py

@@ -13,9 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import eventlet
-eventlet.monkey_patch()
-
 import gettext
 import sys
 

blazar/manager/service.py

@@ -13,11 +13,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 from collections import defaultdict
 import datetime
 from operator import itemgetter
 
-import eventlet
 from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils.excutils import save_and_reraise_exception
@@ -146,7 +146,7 @@ class ManagerService(service_utils.RPCServer):
         return actions
 
     @service_utils.with_empty_context
-    def _process_events_concurrently(self, events):
+    async def _process_events_concurrently(self, events):
         if not events:
             return
 
@@ -160,9 +160,7 @@ class ManagerService(service_utils.RPCServer):
             db_api.event_update(event['id'],
                                 {'status': status.event.IN_PROGRESS})
             try:
-                event_thread = eventlet.spawn(
-                    service_utils.with_empty_context(self._exec_event),
-                    event)
+                event_thread = asyncio.create_task(self._exec_event(event))
                 event_threads[event['id']] = event_thread
             except Exception:
                 db_api.event_update(event['id'],
@@ -172,7 +170,7 @@ class ManagerService(service_utils.RPCServer):
 
         for event_id, event_thread in event_threads.items():
             try:
-                event_thread.wait()
+                await event_thread
             except Exception:
                 db_api.event_update(event['id'],
                                     {'status': status.event.ERROR})
@@ -241,9 +239,10 @@ class ManagerService(service_utils.RPCServer):
         )
 
         for batch in self._select_for_execution(events):
-            self._process_events_concurrently(batch)
+            asyncio.run(self._process_events_concurrently(batch))
 
-    def _exec_event(self, event):
+    @service_utils.with_empty_context
+    async def _exec_event(self, event):
         """Execute an event function"""
         event_fn = getattr(self, event['event_type'], None)
         if event_fn is None:

blazar/monitor/notification_monitor.py

@@ -35,7 +35,7 @@ class NotificationMonitor(base.BaseMonitor):
                 oslo_messaging.get_notification_transport(cfg.CONF),
                 self._get_targets(monitor_plugins),
                 self._get_endpoints(monitor_plugins),
-                executor='eventlet'
+                executor='threading'
             )
             LOG.debug('Notification listener is successfully created.')
         except Exception as e:

blazar/rpc.py

@@ -43,7 +43,7 @@ def get_server(target, endpoints):
         TRANSPORT,
         target,
         endpoints,
-        executor='eventlet',
+        executor='threading',
     )
 
 

blazar/tests/manager/test_service.py

@@ -13,13 +13,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import copy
 import datetime
+import importlib
 from unittest import mock
 
 import ddt
-import eventlet
-import importlib
 from oslo_config import cfg
 import oslo_messaging as messaging
 from oslo_utils import timeutils
@@ -121,7 +121,7 @@ class ServiceTestCase(tests.DBTestCase):
         self.context = context
         self.enabled = enabled
         self.exceptions = exceptions
-        self.eventlet = eventlet
+        self.asyncio = asyncio
         self.datetime = datetime
         self.db_api = db_api
         self.dummy_plugin = dummy_vm_plugin
@@ -284,7 +284,7 @@ class ServiceTestCase(tests.DBTestCase):
                                {'id': '444-555-666', 'time': self.good_date,
                                 'lease_id': 'bbb-ccc-ddd',
                                 'event_type': 'start_lease'}]
-        self.patch(eventlet, 'spawn')
+        self.patch(asyncio, 'create_task')
 
         self.manager._process_events()
 
@@ -340,7 +340,7 @@ class ServiceTestCase(tests.DBTestCase):
             # start_lease event
             mock.call([events_values[4]])])
 
-    def test_process_events_concurrently(self):
+    async def test_process_events_concurrently(self):
         events = [{'id': '111-222-333', 'time': self.good_date,
                    'lease_id': 'aaa-bbb-ccc',
                    'event_type': 'start_lease'},
@@ -350,18 +350,18 @@ class ServiceTestCase(tests.DBTestCase):
                   {'id': '333-444-555', 'time': self.good_date,
                    'lease_id': 'ccc-ddd-eee',
                    'event_type': 'start_lease'}]
-        spawn = self.patch(eventlet, 'spawn')
+        create_task = self.patch(asyncio, 'create_task')
 
-        self.manager._process_events_concurrently(events)
-        spawn.assert_has_calls([
+        await self.manager._process_events_concurrently(events)
+        create_task.assert_has_calls([
             mock.call(mock.ANY, events[0]),
             mock.call(mock.ANY, events[1]),
             mock.call(mock.ANY, events[2])])
 
-    def test_event_spawn_fail(self):
+    def test_event_create_task_fail(self):
         events = self.patch(self.db_api, 'event_get_all_sorted_by_filters')
         event_update = self.patch(self.db_api, 'event_update')
-        self.patch(eventlet, 'spawn').side_effect = Exception
+        self.patch(asyncio, 'create_task').side_effect = Exception
         events.return_value = [{'id': '111-222-333', 'time': self.good_date,
                                 'lease_id': 'aaa-bbb-ccc',
                                 'event_type': 'start_lease'}]
@@ -390,13 +390,13 @@ class ServiceTestCase(tests.DBTestCase):
 
         event_update.assert_not_called()
 
-    def test_exec_event_success(self):
+    async def test_exec_event_success(self):
         event = {'id': '111-222-333',
                  'event_type': 'start_lease',
                  'lease_id': self.lease_id}
         start_lease = self.patch(self.manager, 'start_lease')
 
-        self.manager._exec_event(event)
+        await self.manager._exec_event(event)
 
         start_lease.assert_called_once_with(lease_id=event['lease_id'],
                                             event_id=event['id'])
@@ -407,7 +407,7 @@ class ServiceTestCase(tests.DBTestCase):
             notifier_api.format_lease_payload(self.lease),
             'lease.event.start_lease')
 
-    def test_exec_event_invalid_event_type(self):
+    async def test_exec_event_invalid_event_type(self):
         event = {'id': '111-222-333',
                  'event_type': 'invalid',
                  'lease_id': self.lease_id}
@@ -416,7 +416,7 @@ class ServiceTestCase(tests.DBTestCase):
                           self.manager._exec_event,
                           event)
 
-    def test_exec_event_retry(self):
+    async def test_exec_event_retry(self):
         event = {'id': '111-222-333',
                  'event_type': 'start_lease',
                  'lease_id': self.lease_id,
@@ -428,7 +428,7 @@ class ServiceTestCase(tests.DBTestCase):
         with mock.patch.object(timeutils, 'utcnow') as patched:
             patched.return_value = (self.good_date + datetime.timedelta(
                 seconds=1))
-            self.manager._exec_event(event)
+            await self.manager._exec_event(event)
 
         start_lease.assert_called_once_with(lease_id=event['lease_id'],
                                             event_id=event['id'])
@@ -436,7 +436,7 @@ class ServiceTestCase(tests.DBTestCase):
             event['id'], {'status': status.event.UNDONE})
         self.lease_get.assert_not_called()
 
-    def test_exec_event_no_more_retry(self):
+    async def test_exec_event_no_more_retry(self):
         event = {'id': '111-222-333',
                  'event_type': 'start_lease',
                  'lease_id': self.lease_id,
@@ -448,7 +448,7 @@ class ServiceTestCase(tests.DBTestCase):
         with mock.patch.object(timeutils, 'utcnow') as patched:
             patched.return_value = (self.good_date + datetime.timedelta(
                 days=1))
-            self.manager._exec_event(event)
+            await self.manager._exec_event(event)
 
         start_lease.assert_called_once_with(lease_id=event['lease_id'],
                                             event_id=event['id'])
@@ -456,7 +456,7 @@ class ServiceTestCase(tests.DBTestCase):
             event['id'], {'status': status.event.ERROR})
         self.lease_get.assert_not_called()
 
-    def test_exec_event_handle_exception(self):
+    async def test_exec_event_handle_exception(self):
         event = {'id': '111-222-333',
                  'event_type': 'start_lease',
                  'lease_id': self.lease_id,
@@ -465,7 +465,7 @@ class ServiceTestCase(tests.DBTestCase):
         start_lease.side_effect = Exception
         event_update = self.patch(self.db_api, 'event_update')
 
-        self.manager._exec_event(event)
+        await self.manager._exec_event(event)
 
         start_lease.assert_called_once_with(lease_id=event['lease_id'],
                                             event_id=event['id'])

blazar/utils/service.py

@@ -86,9 +86,9 @@ class ContextEndpointHandler(object):
 
 def with_empty_context(func):
     @functools.wraps(func)
-    def decorator(*args, **kwargs):
+    async def decorator(*args, **kwargs):
         with context.BlazarContext():
-            return func(*args, **kwargs)
+            await func(*args, **kwargs)
 
     return decorator
 

requirements.txt

@@ -4,7 +4,6 @@
 pbr>=6.1.1 # Apache-2.0
 
 alembic>=0.9.6 # MIT
-eventlet!=0.18.3,!=0.20.1,!=0.21.0,!=0.23.0,!=0.25.0,>=0.18.2 # MIT
 Flask>=1.0.2 # BSD
 iso8601>=0.1.11 # MIT
 keystoneauth1>=3.13.0 # Apache-2.0
@@ -20,7 +19,7 @@ oslo.messaging>=5.29.0 # Apache-2.0
 oslo.middleware>=3.31.0 # Apache-2.0
 oslo.policy>=4.5.0 # Apache-2.0
 oslo.serialization!=2.19.1,>=2.18.0 # Apache-2.0
-oslo.service>=1.34.0 # Apache-2.0
+oslo.service[threading]>=4.2.0 # Apache-2.0
 oslo.upgradecheck>=1.3.0 # Apache-2.0
 oslo.utils>=7.0.0 # Apache-2.0
 python-neutronclient>=6.0.0 # Apache-2.0