Merge "Remove the task manager"

This commit is contained in:
Zuul 2018-09-26 15:07:09 +00:00 committed by Gerrit Code Review
commit 6e733e77d5
5 changed files with 6 additions and 565 deletions

View File

@ -39,13 +39,6 @@ Most of the logging is set up to log to the root `shade` logger. There are
additional sub-loggers that are used at times, primarily so that a user can
decide to turn on or off a specific type of logging. They are listed below.
shade.task_manager
`shade` uses a Task Manager to perform remote calls. The `shade.task_manager`
logger emits messages at the start and end of each Task announging what
it is going to run and then what it ran and how long it took. Logging
`shade.task_manager` is a good way to get a trace of external actions shade
is taking without full `HTTP Tracing`_.
shade.request_ids
The `shade.request_ids` logger emits a log line at the end of each HTTP
interaction with the OpenStack Request ID associated with the interaction.

View File

@ -0,0 +1,6 @@
---
upgrade:
- |
The ``manager`` parameter is no longer meaningful. This should have no
impact as the only known consumer of the feature is nodepool which
no longer uses shade.

View File

@ -11,7 +11,6 @@
# limitations under the License.
import base64
import collections
import copy
import datetime
import functools
@ -35,7 +34,6 @@ import os
from openstack.cloud import _utils
from openstack.config import loader
from openstack import connection
from openstack import task_manager
from openstack import utils
from shade import exc
@ -135,7 +133,6 @@ class OpenStackCloud(
super(OpenStackCloud, self).__init__(
config=cloud_config,
strict=strict,
task_manager=manager,
app_name=app_name,
app_version=app_version,
use_direct_get=use_direct_get,
@ -7270,118 +7267,6 @@ class OpenStackCloud(
endpoint, filename, headers,
file_size, segment_size, use_slo)
def _upload_object(self, endpoint, filename, headers):
return self._object_store_client.put(
endpoint, headers=headers, data=open(filename, 'r'))
def _get_file_segments(self, endpoint, filename, file_size, segment_size):
# Use an ordered dict here so that testing can replicate things
segments = collections.OrderedDict()
for (index, offset) in enumerate(range(0, file_size, segment_size)):
remaining = file_size - (index * segment_size)
segment = _utils.FileSegment(
filename, offset,
segment_size if segment_size < remaining else remaining)
name = '{endpoint}/{index:0>6}'.format(
endpoint=endpoint, index=index)
segments[name] = segment
return segments
def _object_name_from_url(self, url):
'''Get container_name/object_name from the full URL called.
Remove the Swift endpoint from the front of the URL, and remove
the leaving / that will leave behind.'''
endpoint = self._object_store_client.get_endpoint()
object_name = url.replace(endpoint, '')
if object_name.startswith('/'):
object_name = object_name[1:]
return object_name
def _add_etag_to_manifest(self, segment_results, manifest):
for result in segment_results:
if 'Etag' not in result.headers:
continue
name = self._object_name_from_url(result.url)
for entry in manifest:
if entry['path'] == '/{name}'.format(name=name):
entry['etag'] = result.headers['Etag']
def _upload_large_object(
self, endpoint, filename,
headers, file_size, segment_size, use_slo):
# If the object is big, we need to break it up into segments that
# are no larger than segment_size, upload each of them individually
# and then upload a manifest object. The segments can be uploaded in
# parallel, so we'll use the async feature of the TaskManager.
segment_futures = []
segment_results = []
retry_results = []
retry_futures = []
manifest = []
# Get an OrderedDict with keys being the swift location for the
# segment, the value a FileSegment file-like object that is a
# slice of the data for the segment.
segments = self._get_file_segments(
endpoint, filename, file_size, segment_size)
# Schedule the segments for upload
for name, segment in segments.items():
# Async call to put - schedules execution and returns a future
segment_future = self._object_store_client.put(
name, headers=headers, data=segment, run_async=True)
segment_futures.append(segment_future)
# TODO(mordred) Collect etags from results to add to this manifest
# dict. Then sort the list of dicts by path.
manifest.append(dict(
path='/{name}'.format(name=name),
size_bytes=segment.length))
# Try once and collect failed results to retry
segment_results, retry_results = task_manager.wait_for_futures(
segment_futures, raise_on_error=False)
self._add_etag_to_manifest(segment_results, manifest)
for result in retry_results:
# Grab the FileSegment for the failed upload so we can retry
name = self._object_name_from_url(result.url)
segment = segments[name]
segment.seek(0)
# Async call to put - schedules execution and returns a future
segment_future = self._object_store_client.put(
name, headers=headers, data=segment, run_async=True)
# TODO(mordred) Collect etags from results to add to this manifest
# dict. Then sort the list of dicts by path.
retry_futures.append(segment_future)
# If any segments fail the second time, just throw the error
segment_results, retry_results = task_manager.wait_for_futures(
retry_futures, raise_on_error=True)
self._add_etag_to_manifest(segment_results, manifest)
if use_slo:
return self._finish_large_object_slo(endpoint, headers, manifest)
else:
return self._finish_large_object_dlo(endpoint, headers)
def _finish_large_object_slo(self, endpoint, headers, manifest):
# TODO(mordred) send an etag of the manifest, which is the md5sum
# of the concatenation of the etags of the results
headers = headers.copy()
return self._object_store_client.put(
endpoint,
params={'multipart-manifest': 'put'},
headers=headers, data=json.dumps(manifest))
def _finish_large_object_dlo(self, endpoint, headers):
headers = headers.copy()
headers['X-Object-Manifest'] = endpoint
return self._object_store_client.put(endpoint, headers=headers)
def update_object(self, container, name, metadata=None, **headers):
"""Update the metadata of an object

View File

@ -1,334 +0,0 @@
# Copyright (C) 2011-2013 OpenStack Foundation
#
# 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 abc
import concurrent.futures
import sys
import threading
import time
import types
import keystoneauth1.exceptions
import six
from shade import _log
from shade import exc
from shade import meta
def _is_listlike(obj):
# NOTE(Shrews): Since the client API might decide to subclass one
# of these result types, we use isinstance() here instead of type().
return (
isinstance(obj, list) or
isinstance(obj, types.GeneratorType))
def _is_objlike(obj):
# NOTE(Shrews): Since the client API might decide to subclass one
# of these result types, we use isinstance() here instead of type().
return (
not isinstance(obj, bool) and
not isinstance(obj, int) and
not isinstance(obj, float) and
not isinstance(obj, six.string_types) and
not isinstance(obj, set) and
not isinstance(obj, tuple))
@six.add_metaclass(abc.ABCMeta)
class BaseTask(object):
"""Represent a task to be performed on an OpenStack Cloud.
Some consumers need to inject things like rate-limiting or auditing
around each external REST interaction. Task provides an interface
to encapsulate each such interaction. Also, although shade itself
operates normally in a single-threaded direct action manner, consuming
programs may provide a multi-threaded TaskManager themselves. For that
reason, Task uses threading events to ensure appropriate wait conditions.
These should be a no-op in single-threaded applications.
A consumer is expected to overload the main method.
:param dict kw: Any args that are expected to be passed to something in
the main payload at execution time.
"""
def __init__(self, **kw):
self._exception = None
self._traceback = None
self._result = None
self._response = None
self._finished = threading.Event()
self.run_async = False
self.args = kw
self.name = type(self).__name__
@abc.abstractmethod
def main(self, client):
""" Override this method with the actual workload to be performed """
def done(self, result):
self._result = result
self._finished.set()
def exception(self, e, tb):
self._exception = e
self._traceback = tb
self._finished.set()
def wait(self, raw=False):
self._finished.wait()
if self._exception:
six.reraise(type(self._exception), self._exception,
self._traceback)
return self._result
def run(self, client):
self._client = client
try:
# Retry one time if we get a retriable connection failure
try:
# Keep time for connection retrying logging
start = time.time()
self.done(self.main(client))
except keystoneauth1.exceptions.RetriableConnectionFailure as e:
end = time.time()
dt = end - start
if client.region_name:
client.log.debug(str(e))
client.log.debug(
"Connection failure on %(cloud)s:%(region)s"
" for %(name)s after %(secs)s seconds, retrying",
{'cloud': client.name,
'region': client.region_name,
'secs': dt,
'name': self.name})
else:
client.log.debug(
"Connection failure on %(cloud)s for %(name)s after"
" %(secs)s seconds, retrying",
{'cloud': client.name, 'name': self.name, 'secs': dt})
self.done(self.main(client))
except Exception:
raise
except Exception as e:
self.exception(e, sys.exc_info()[2])
class Task(BaseTask):
""" Shade specific additions to the BaseTask Interface. """
def wait(self, raw=False):
super(Task, self).wait()
if raw:
# Do NOT convert the result.
return self._result
if _is_listlike(self._result):
return meta.obj_list_to_munch(self._result)
elif _is_objlike(self._result):
return meta.obj_to_munch(self._result)
else:
return self._result
class RequestTask(BaseTask):
""" Extensions to the Shade Tasks to handle raw requests """
# It's totally legit for calls to not return things
result_key = None
# keystoneauth1 throws keystoneauth1.exceptions.http.HttpError on !200
def done(self, result):
self._response = result
try:
result_json = self._response.json()
except Exception as e:
result_json = self._response.text
self._client.log.debug(
'Could not decode json in response: %(e)s', {'e': str(e)})
self._client.log.debug(result_json)
if self.result_key:
self._result = result_json[self.result_key]
else:
self._result = result_json
self._request_id = self._response.headers.get('x-openstack-request-id')
self._finished.set()
def wait(self, raw=False):
super(RequestTask, self).wait()
if raw:
# Do NOT convert the result.
return self._result
if _is_listlike(self._result):
return meta.obj_list_to_munch(
self._result, request_id=self._request_id)
elif _is_objlike(self._result):
return meta.obj_to_munch(self._result, request_id=self._request_id)
return self._result
def _result_filter_cb(result):
return result
def generate_task_class(method, name, result_filter_cb):
if name is None:
if callable(method):
name = method.__name__
else:
name = method
class RunTask(Task):
def __init__(self, **kw):
super(RunTask, self).__init__(**kw)
self.name = name
self._method = method
def wait(self, raw=False):
super(RunTask, self).wait()
if raw:
# Do NOT convert the result.
return self._result
return result_filter_cb(self._result)
def main(self, client):
if callable(self._method):
return method(**self.args)
else:
meth = getattr(client, self._method)
return meth(**self.args)
return RunTask
class TaskManager(object):
log = _log.setup_logging('shade.task_manager')
def __init__(
self, client, name, result_filter_cb=None, workers=5, **kwargs):
self.name = name
self._client = client
self._executor = concurrent.futures.ThreadPoolExecutor(
max_workers=workers)
if not result_filter_cb:
self._result_filter_cb = _result_filter_cb
else:
self._result_filter_cb = result_filter_cb
def set_client(self, client):
self._client = client
def stop(self):
""" This is a direct action passthrough TaskManager """
self._executor.shutdown(wait=True)
def run(self):
""" This is a direct action passthrough TaskManager """
pass
def submit_task(self, task, raw=False):
"""Submit and execute the given task.
:param task: The task to execute.
:param bool raw: If True, return the raw result as received from the
underlying client call.
"""
return self.run_task(task=task, raw=raw)
def _run_task_async(self, task, raw=False):
self.log.debug(
"Manager %s submitting task %s", self.name, task.name)
return self._executor.submit(self._run_task, task, raw=raw)
def run_task(self, task, raw=False):
if hasattr(task, 'run_async') and task.run_async:
return self._run_task_async(task, raw=raw)
else:
return self._run_task(task, raw=raw)
def _run_task(self, task, raw=False):
self.log.debug(
"Manager %s running task %s", self.name, task.name)
start = time.time()
task.run(self._client)
end = time.time()
dt = end - start
self.log.debug(
"Manager %s ran task %s in %ss", self.name, task.name, dt)
self.post_run_task(dt, task)
return task.wait(raw)
def post_run_task(self, elasped_time, task):
pass
# Backwards compatibility
submitTask = submit_task
def submit_function(
self, method, name=None, result_filter_cb=None, **kwargs):
""" Allows submitting an arbitrary method for work.
:param method: Method to run in the TaskManager. Can be either the
name of a method to find on self.client, or a callable.
"""
if not result_filter_cb:
result_filter_cb = self._result_filter_cb
task_class = generate_task_class(method, name, result_filter_cb)
return self._executor.submit_task(task_class(**kwargs))
def wait_for_futures(futures, raise_on_error=True, log=None):
'''Collect results or failures from a list of running future tasks.'''
results = []
retries = []
# Check on each result as its thread finishes
for completed in concurrent.futures.as_completed(futures):
try:
result = completed.result()
# We have to do this here because munch_response doesn't
# get called on async job results
exc.raise_from_response(result)
results.append(result)
except (keystoneauth1.exceptions.RetriableConnectionFailure,
exc.OpenStackCloudException) as e:
if log:
log.debug(
"Exception processing async task: {e}".format(
e=str(e)),
exc_info=True)
# If we get an exception, put the result into a list so we
# can try again
if raise_on_error:
raise
else:
retries.append(result)
return results, retries

View File

@ -1,109 +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 mock
from shade import task_manager
from shade.tests.unit import base
class TestException(Exception):
pass
class TaskTest(task_manager.Task):
def main(self, client):
raise TestException("This is a test exception")
class TaskTestGenerator(task_manager.Task):
def main(self, client):
yield 1
class TaskTestInt(task_manager.Task):
def main(self, client):
return int(1)
class TaskTestFloat(task_manager.Task):
def main(self, client):
return float(2.0)
class TaskTestStr(task_manager.Task):
def main(self, client):
return "test"
class TaskTestBool(task_manager.Task):
def main(self, client):
return True
class TaskTestSet(task_manager.Task):
def main(self, client):
return set([1, 2])
class TaskTestAsync(task_manager.Task):
def __init__(self):
super(task_manager.Task, self).__init__()
self.run_async = True
def main(self, client):
pass
class TestTaskManager(base.RequestsMockTestCase):
def setUp(self):
super(TestTaskManager, self).setUp()
self.manager = task_manager.TaskManager(name='test', client=self)
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, p34, pypy, ...)
"""
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_task(TaskTestAsync())
self.assertTrue(mock_submit.called)