openstack/openstacksdk
Code Issues Proposed changes

Revert "Revert "Replace TaskManager with a keystoneauth concurrency""

Browse Source 
This reverts commit feae8da9bf7194d6294cf79b26311b71ed88f742.

Change-Id: I6f85cc58a404063a262cf7338c827cde61493b43
This commit is contained in:
Monty Taylor 2019-03-10 22:17:19 +00:00
parent feae8da9bf
commit e7a9b52998
11 changed files with 13 additions and 307 deletions

8
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

2
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

33
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.
Uses task_manager to run tasks rather than executing them directly.
This allows using the nodepool MultiThreaded Rate Limiting TaskManager.
"""
"""Wrapper around keystoneauth1.adapter.Adapter."""
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)
name = '.'.join([self.service_type, method] + name_parts)
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,
response = super(OpenStackSDKAdapter, self).request(
url, method,
connect_retries=connect_retries, raise_exc=False,
**kwargs)
return ret.result()
return response
def _version_matches(self, version):
api_version = self.get_api_major_version()

3
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),

1
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 += '/'

22
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.
Defaults to None which causes a direct-action Task Manager to be
used.
:type manager: :class:`~openstack.task_manager.TaskManager`
Ignored. Exists for backwards compat during transition. Rate limit
parameters should be passed directly to the `rate_limit` parameter.
: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

5
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
)

1
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,

227
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)

16
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)

2
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