Merge "Replace TaskManager with a keystoneauth concurrency"

doc/source/user/guides/logging.rst

@@ -65,14 +65,6 @@ openstack.config
   Issues pertaining to configuration are logged to the ``openstack.config``
   logger.
 
-openstack.task_manager
-  `openstacksdk` uses a Task Manager to perform remote calls. The
-  ``openstack.task_manager`` logger emits messages at the start and end
-  of each Task announcing what it is going to run and then what it ran and
-  how long it took. Logging ``openstack.task_manager`` is a good way to
-  get a trace of external actions `openstacksdk` is taking without full
-  `HTTP Tracing`_.
-
 openstack.iterate_timeout
   When `openstacksdk` needs to poll a resource, it does so in a loop that waits
   between iterations and ultimately times out. The

lower-constraints.txt

@@ -13,7 +13,7 @@ jmespath==0.9.0
 jsonpatch==1.16
 jsonpointer==1.13
 jsonschema==2.6.0
-keystoneauth1==3.11.0
+keystoneauth1==3.13.0
 linecache2==1.0.0
 mock==2.0.0
 mox3==0.20.0

openstack/_adapter.py

@@ -14,7 +14,6 @@
 
 ''' Wrapper around keystoneauth Adapter to wrap calls in TaskManager '''
 
-import functools
 try:
     import simplejson
     JSONDecodeError = simplejson.scanner.JSONDecodeError
@@ -25,7 +24,6 @@ from six.moves import urllib
 from keystoneauth1 import adapter
 
 from openstack import exceptions
-from openstack import task_manager as _task_manager
 
 
 def _extract_name(url, service_type=None):
@@ -109,43 +107,22 @@ def _json_response(response, result_key=None, error_message=None):
 
 
 class OpenStackSDKAdapter(adapter.Adapter):
-    """Wrapper around keystoneauth1.adapter.Adapter.
+    """Wrapper around keystoneauth1.adapter.Adapter."""
-
-    Uses task_manager to run tasks rather than executing them directly.
-    This allows using the nodepool MultiThreaded Rate Limiting TaskManager.
-    """
 
     def __init__(
             self, session=None,
-            task_manager=None,
-            rate_limit=None, concurrency=None,
             *args, **kwargs):
         super(OpenStackSDKAdapter, self).__init__(
             session=session, *args, **kwargs)
-        if not task_manager:
-            task_manager = _task_manager.TaskManager(
-                name=self.service_type,
-                rate=rate_limit,
-                workers=concurrency)
-            task_manager.start()
-
-        self.task_manager = task_manager
 
     def request(
             self, url, method, error_message=None,
             raise_exc=False, connect_retries=1, *args, **kwargs):
-        name_parts = _extract_name(url, self.service_type)
+        response = super(OpenStackSDKAdapter, self).request(
-        name = '.'.join([self.service_type, method] + name_parts)
+            url, method,
-
+            connect_retries=connect_retries, raise_exc=False,
-        request_method = functools.partial(
-            super(OpenStackSDKAdapter, self).request, url, method)
-
-        ret = self.task_manager.submit_function(
-            request_method, run_async=True, name=name,
-            connect_retries=connect_retries, raise_exc=raise_exc,
-            tag=self.service_type,
             **kwargs)
-        return ret.result()
+        return response
 
     def _version_matches(self, version):
         api_version = self.get_api_major_version()

openstack/cloud/openstackcloud.py

@@ -452,7 +452,6 @@ class _OpenStackCloudMixin(_normalize.Normalizer):
                 version=config_major)
             adapter = _adapter.ShadeAdapter(
                 session=self.session,
-                task_manager=self.task_manager,
                 service_type=self.config.get_service_type(service_type),
                 service_name=self.config.get_service_name(service_type),
                 interface=self.config.get_interface(service_type),
@@ -465,7 +464,6 @@ class _OpenStackCloudMixin(_normalize.Normalizer):
 
         adapter = _adapter.ShadeAdapter(
             session=self.session,
-            task_manager=self.task_manager,
             service_type=self.config.get_service_type(service_type),
             service_name=self.config.get_service_name(service_type),
             interface=self.config.get_interface(service_type),
@@ -504,7 +502,6 @@ class _OpenStackCloudMixin(_normalize.Normalizer):
             self, service_type, api_version=None, endpoint_override=None):
         return _adapter.ShadeAdapter(
             session=self.session,
-            task_manager=self.task_manager,
             service_type=self.config.get_service_type(service_type),
             service_name=self.config.get_service_name(service_type),
             interface=self.config.get_interface(service_type),

openstack/config/cloud_region.py

@@ -492,7 +492,6 @@ class CloudRegion(object):
                     region_name=self.region_name,
                 )
                 network_endpoint = network_adapter.get_endpoint()
-                network_adapter.task_manager.stop()
                 if not network_endpoint.rstrip().rsplit('/')[-1] == 'v2.0':
                     if not network_endpoint.endswith('/'):
                         network_endpoint += '/'

openstack/connection.py

@@ -168,7 +168,6 @@ from openstack import config as _config
 from openstack.config import cloud_region
 from openstack import exceptions
 from openstack import service_description
-from openstack import task_manager as _task_manager
 
 __all__ = [
     'from_config',
@@ -259,10 +258,8 @@ class Connection(six.with_metaclass(_meta.ConnectionMeta,
             filtering instead of making list calls and filtering client-side.
             Default false.
         :param task_manager:
-            Task Manager to handle the execution of remote REST calls.
+            Ignored. Exists for backwards compat during transition. Rate limit
-            Defaults to None which causes a direct-action Task Manager to be
+            parameters should be passed directly to the `rate_limit` parameter.
-            used.
-        :type manager: :class:`~openstack.task_manager.TaskManager`
         :param rate_limit:
             Client-side rate limit, expressed in calls per second. The
             parameter can either be a single float, or it can be a dict with
@@ -286,6 +283,7 @@ class Connection(six.with_metaclass(_meta.ConnectionMeta,
                     app_name=app_name, app_version=app_version,
                     load_yaml_config=False,
                     load_envvars=False,
+                    rate_limit=rate_limit,
                     **kwargs)
             else:
                 self.config = _config.get_cloud_region(
@@ -293,18 +291,9 @@ class Connection(six.with_metaclass(_meta.ConnectionMeta,
                     app_name=app_name, app_version=app_version,
                     load_yaml_config=cloud is not None,
                     load_envvars=cloud is not None,
+                    rate_limit=rate_limit,
                     **kwargs)
 
-        if task_manager:
-            # If a TaskManager was passed in, don't start it, assume it's
-            # under the control of the calling context.
-            self.task_manager = task_manager
-        else:
-            self.task_manager = _task_manager.TaskManager(
-                self.config.full_name,
-                rate=rate_limit)
-            self.task_manager.start()
-
         self._session = None
         self._proxies = {}
         self.use_direct_get = use_direct_get
@@ -371,7 +360,8 @@ class Connection(six.with_metaclass(_meta.ConnectionMeta,
 
     def close(self):
         """Release any resources held open."""
-        self.task_manager.stop()
+        if self.__pool_executor:
+            self.__pool_executor.shutdown()
 
     def __enter__(self):
         return self

openstack/service_description.py

@@ -108,7 +108,6 @@ class ServiceDescription(object):
                 proxy_obj = config.get_session_client(
                     self.service_type,
                     constructor=proxy_class,
-                    task_manager=instance.task_manager,
                 )
             else:
                 warnings.warn(
@@ -129,7 +128,6 @@ class ServiceDescription(object):
                 proxy_obj = config.get_session_client(
                     self.service_type,
                     constructor=proxy_class,
-                    task_manager=instance.task_manager,
                 )
             else:
                 warnings.warn(
@@ -165,7 +163,6 @@ class ServiceDescription(object):
                 proxy_obj = config.get_session_client(
                     self.service_type,
                     constructor=proxy_class,
-                    task_manager=instance.task_manager,
                 )
             return proxy_obj
 
@@ -199,14 +196,12 @@ class ServiceDescription(object):
                     service_type=self.service_type),
                 category=exceptions.UnsupportedServiceVersion)
             return temp_adapter
-        temp_adapter.task_manager.stop()
         proxy_class = self.supported_versions.get(str(found_version[0]))
         if not proxy_class:
             proxy_class = proxy.Proxy
         return config.get_session_client(
             self.service_type,
             constructor=proxy_class,
-            task_manager=instance.task_manager,
             allow_version_hack=True,
             **version_kwargs
         )

openstack/tests/unit/base.py

@@ -438,7 +438,6 @@ class TestCase(base.TestCase):
             cloud=test_cloud, validate=True, **kwargs)
         self.cloud = openstack.connection.Connection(
             config=self.cloud_config, strict=self.strict_cloud)
-        self.addCleanup(self.cloud.task_manager.stop)
 
     def get_cinder_discovery_mock_dict(
             self,

openstack/tests/unit/cloud/test_task_manager.py

@@ -1,227 +0,0 @@
-# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
-#
-# 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 concurrent.futures
-import fixtures
-import mock
-import threading
-import time
-
-from six.moves import queue
-
-from openstack import task_manager
-from openstack.tests.unit import base
-
-
-class TestException(Exception):
-    pass
-
-
-class TaskTest(task_manager.Task):
-    def main(self):
-        raise TestException("This is a test exception")
-
-
-class TaskTestGenerator(task_manager.Task):
-    def main(self):
-        yield 1
-
-
-class TaskTestInt(task_manager.Task):
-    def main(self):
-        return int(1)
-
-
-class TaskTestFloat(task_manager.Task):
-    def main(self):
-        return float(2.0)
-
-
-class TaskTestStr(task_manager.Task):
-    def main(self):
-        return "test"
-
-
-class TaskTestBool(task_manager.Task):
-    def main(self):
-        return True
-
-
-class TaskTestSet(task_manager.Task):
-    def main(self):
-        return set([1, 2])
-
-
-class TestRateTransforms(base.TestCase):
-
-    def test_rate_parameter_scalar(self):
-        manager = task_manager.TaskManager(name='test', rate=0.1234)
-        self.assertEqual(1 / 0.1234, manager._get_wait('compute'))
-        self.assertEqual(1 / 0.1234, manager._get_wait(None))
-
-    def test_rate_parameter_dict(self):
-        manager = task_manager.TaskManager(
-            name='test',
-            rate={
-                'compute': 20,
-                'network': 10,
-            })
-        self.assertEqual(1.0 / 20, manager._get_wait('compute'))
-        self.assertEqual(1.0 / 10, manager._get_wait('network'))
-        self.assertIsNone(manager._get_wait('object-store'))
-
-
-class TestTaskManager(base.TestCase):
-
-    def setUp(self):
-        super(TestTaskManager, self).setUp()
-        self.manager = task_manager.TaskManager(name='test')
-        self.manager.start()
-
-    def test_wait_re_raise(self):
-        """Test that Exceptions thrown in a Task is reraised correctly
-
-        This test is aimed to six.reraise(), called in Task::wait().
-        Specifically, we test if we get the same behaviour with all the
-        configured interpreters (e.g. py27, p35, ...)
-        """
-        self.assertRaises(TestException, self.manager.submit_task, TaskTest())
-
-    def test_dont_munchify_int(self):
-        ret = self.manager.submit_task(TaskTestInt())
-        self.assertIsInstance(ret, int)
-
-    def test_dont_munchify_float(self):
-        ret = self.manager.submit_task(TaskTestFloat())
-        self.assertIsInstance(ret, float)
-
-    def test_dont_munchify_str(self):
-        ret = self.manager.submit_task(TaskTestStr())
-        self.assertIsInstance(ret, str)
-
-    def test_dont_munchify_bool(self):
-        ret = self.manager.submit_task(TaskTestBool())
-        self.assertIsInstance(ret, bool)
-
-    def test_dont_munchify_set(self):
-        ret = self.manager.submit_task(TaskTestSet())
-        self.assertIsInstance(ret, set)
-
-    @mock.patch.object(concurrent.futures.ThreadPoolExecutor, 'submit')
-    def test_async(self, mock_submit):
-
-        self.manager.submit_function(set, run_async=True)
-        self.assertTrue(mock_submit.called)
-
-    @mock.patch.object(task_manager.TaskManager, 'post_run_task')
-    @mock.patch.object(task_manager.TaskManager, 'pre_run_task')
-    def test_pre_post_calls(self, mock_pre, mock_post):
-        self.manager.submit_function(lambda: None)
-        mock_pre.assert_called_once()
-        mock_post.assert_called_once()
-
-    @mock.patch.object(task_manager.TaskManager, 'post_run_task')
-    @mock.patch.object(task_manager.TaskManager, 'pre_run_task')
-    def test_validate_timing(self, mock_pre, mock_post):
-        # Note the unit test setup has mocked out time.sleep() and
-        # done a * 0.0001, and the test should be under the 5
-        # second timeout.  Thus with below, we should see at
-        # *least* a 1 second pause running the task.
-        self.manager.submit_function(lambda: time.sleep(10000))
-
-        mock_pre.assert_called_once()
-        mock_post.assert_called_once()
-
-        args, kwargs = mock_post.call_args_list[0]
-        self.assertTrue(args[0] > 1.0)
-
-
-class ThreadingTaskManager(task_manager.TaskManager):
-    """A subclass of TaskManager which exercises the thread-shifting
-       exception handling behavior."""
-
-    def __init__(self, *args, **kw):
-        super(ThreadingTaskManager, self).__init__(
-            *args, **kw)
-        self.queue = queue.Queue()
-        self._running = True
-        self._thread = threading.Thread(name=self.name, target=self.run)
-        self._thread.daemon = True
-        self.failed = False
-
-    def start(self):
-        self._thread.start()
-
-    def stop(self):
-        self._running = False
-        self.queue.put(None)
-
-    def join(self):
-        self._thread.join()
-
-    def run(self):
-        # No exception should ever cause this method to hit its
-        # exception handler.
-        try:
-            while True:
-                task = self.queue.get()
-                if not task:
-                    if not self._running:
-                        break
-                    continue
-                self.run_task(task)
-                self.queue.task_done()
-        except Exception:
-            self.failed = True
-            raise
-
-    def submit_task(self, task, raw=False):
-        # An important part of the exception-shifting feature is that
-        # this method should raise the exception.
-        self.queue.put(task)
-        return task.wait()
-
-
-class ThreadingTaskManagerFixture(fixtures.Fixture):
-    def _setUp(self):
-        self.manager = ThreadingTaskManager(name='threading test')
-        self.manager.start()
-        self.addCleanup(self._cleanup)
-
-    def _cleanup(self):
-        self.manager.stop()
-        self.manager.join()
-
-
-class TestThreadingTaskManager(base.TestCase):
-
-    def setUp(self):
-        super(TestThreadingTaskManager, self).setUp()
-        f = self.useFixture(ThreadingTaskManagerFixture())
-        self.manager = f.manager
-
-    def test_wait_re_raise(self):
-        """Test that Exceptions thrown in a Task is reraised correctly
-
-        This test is aimed to six.reraise(), called in Task::wait().
-        Specifically, we test if we get the same behaviour with all the
-        configured interpreters (e.g. py27, p35, ...)
-        """
-        self.assertRaises(TestException, self.manager.submit_task, TaskTest())
-        # Stop the manager and join the run thread to ensure the
-        # exception handler has run.
-        self.manager.stop()
-        self.manager.join()
-        self.assertFalse(self.manager.failed)

openstack/tests/unit/test_connection.py

@@ -91,22 +91,6 @@ class TestConnection(base.TestCase):
         self.assertEqual(mock_session, conn.session)
         self.assertEqual('auth.example.com', conn.config.name)
 
-    def test_task_manager_rate_scalar(self):
-        conn = connection.Connection(cloud='sample-cloud', rate_limit=20)
-        self.assertEqual(1.0 / 20, conn.task_manager._get_wait('object-store'))
-        self.assertEqual(1.0 / 20, conn.task_manager._get_wait(None))
-
-    def test_task_manager_rate_dict(self):
-        conn = connection.Connection(
-            cloud='sample-cloud',
-            rate_limit={
-                'compute': 20,
-                'network': 10,
-            })
-        self.assertEqual(1.0 / 20, conn.task_manager._get_wait('compute'))
-        self.assertEqual(1.0 / 10, conn.task_manager._get_wait('network'))
-        self.assertIsNone(conn.task_manager._get_wait('object-store'))
-
     def test_create_session(self):
         conn = connection.Connection(cloud='sample-cloud')
         self.assertIsNotNone(conn)

requirements.txt

@@ -8,7 +8,7 @@ requestsexceptions>=1.2.0 # Apache-2.0
 jsonpatch!=1.20,>=1.16 # BSD
 six>=1.10.0 # MIT
 os-service-types>=1.2.0 # Apache-2.0
-keystoneauth1>=3.11.0 # Apache-2.0
+keystoneauth1>=3.13.0 # Apache-2.0
 
 munch>=2.1.0 # MIT
 decorator>=3.4.0 # BSD