diff options
| -rw-r--r-- | requirements.txt | 1 | ||||
| -rw-r--r-- | yardstick/common/exceptions.py | 5 | ||||
| -rw-r--r-- | yardstick/common/messaging/__init__.py | 36 | ||||
| -rw-r--r-- | yardstick/common/messaging/consumer.py | 85 | ||||
| -rw-r--r-- | yardstick/common/messaging/payloads.py | 53 | ||||
| -rw-r--r-- | yardstick/common/messaging/producer.py | 70 | ||||
| -rw-r--r-- | yardstick/tests/functional/common/messaging/__init__.py | 0 | ||||
| -rw-r--r-- | yardstick/tests/functional/common/messaging/test_messaging.py | 94 | ||||
| -rw-r--r-- | yardstick/tests/unit/common/messaging/__init__.py | 0 | ||||
| -rw-r--r-- | yardstick/tests/unit/common/messaging/test_consumer.py | 54 | ||||
| -rw-r--r-- | yardstick/tests/unit/common/messaging/test_payloads.py | 46 | ||||
| -rw-r--r-- | yardstick/tests/unit/common/messaging/test_producer.py | 46 |
12 files changed, 490 insertions, 0 deletions
diff --git a/requirements.txt b/requirements.txt index 02545de1d..43f0e796c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -37,6 +37,7 @@ os-client-config==1.28.0 # OSI Approved Apache Software License osc-lib==1.7.0 # OSI Approved Apache Software License oslo.config==4.11.1 # OSI Approved Apache Software License oslo.i18n==3.17.0 # OSI Approved Apache Software License +oslo.messaging===5.30.2 # OSI Approved Apache Software License oslo.privsep===1.22.1 # OSI Approved Apache Software License oslo.serialization==2.20.1 # OSI Approved Apache Software License oslo.utils==3.28.0 # OSI Approved Apache Software License diff --git a/yardstick/common/exceptions.py b/yardstick/common/exceptions.py index 439b9cb1b..65e444071 100644 --- a/yardstick/common/exceptions.py +++ b/yardstick/common/exceptions.py @@ -64,6 +64,11 @@ class YardstickBannedModuleImported(YardstickException): message = 'Module "%(module)s" cannnot be imported. Reason: "%(reason)s"' +class PayloadMissingAttributes(YardstickException): + message = ('Error instantiating a Payload class, missing attributes: ' + '%(missing_attributes)s') + + class HeatTemplateError(YardstickException): """Error in Heat during the stack deployment""" message = ('Error in Heat during the creation of the OpenStack stack ' diff --git a/yardstick/common/messaging/__init__.py b/yardstick/common/messaging/__init__.py new file mode 100644 index 000000000..f0f012ec3 --- /dev/null +++ b/yardstick/common/messaging/__init__.py @@ -0,0 +1,36 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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. + +# MQ is statically configured: +# - MQ service: RabbitMQ +# - user/password: yardstick/yardstick +# - host:port: localhost:5672 +MQ_USER = 'yardstick' +MQ_PASS = 'yardstick' +MQ_SERVICE = 'rabbit' +SERVER = 'localhost' +PORT = 5672 +TRANSPORT_URL = (MQ_SERVICE + '://' + MQ_USER + ':' + MQ_PASS + '@' + SERVER + + ':' + str(PORT) + '/') + +# RPC server. +RPC_SERVER_EXECUTOR = 'threading' + +# Topics. +RUNNER = 'runner' + +# Methods. +# RUNNER methods: +RUNNER_INFO = 'runner_info' +RUNNER_LOOP = 'runner_loop' diff --git a/yardstick/common/messaging/consumer.py b/yardstick/common/messaging/consumer.py new file mode 100644 index 000000000..a0feeb300 --- /dev/null +++ b/yardstick/common/messaging/consumer.py @@ -0,0 +1,85 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 logging + +from oslo_config import cfg +import oslo_messaging +import six + +from yardstick.common import messaging + + +LOG = logging.getLogger(__name__) + + +@six.add_metaclass(abc.ABCMeta) +class NotificationHandler(object): + """Abstract class to define a endpoint object for a MessagingConsumer""" + + def __init__(self, id, ctx_pid, queue): + self._id = id + self._ctx_pid = ctx_pid + self._queue = queue + + +@six.add_metaclass(abc.ABCMeta) +class MessagingConsumer(object): + """Abstract class to implement a MQ consumer + + This abstract class allows a class implementing this interface to receive + the messages published by a `MessagingNotifier`. + """ + + def __init__(self, topic, pid, endpoints, fanout=True): + """Init function. + + :param topic: (string) MQ exchange topic + :param pid: (int) PID of the process implementing the MQ Notifier which + will be in the message context + :param endpoints: (list of class) list of classes implementing the + methods (see `MessagingNotifier.send_message) used by + the Notifier + :param fanout: (bool) MQ clients may request that a copy of the message + be delivered to all servers listening on a topic by + setting fanout to ``True``, rather than just one of them + :returns: `MessagingConsumer` class object + """ + + self._pid = pid + self._endpoints = endpoints + self._transport = oslo_messaging.get_rpc_transport( + cfg.CONF, url=messaging.TRANSPORT_URL) + self._target = oslo_messaging.Target(topic=topic, fanout=fanout, + server=messaging.SERVER) + self._server = oslo_messaging.get_rpc_server( + self._transport, self._target, self._endpoints, + executor=messaging.RPC_SERVER_EXECUTOR, + access_policy=oslo_messaging.DefaultRPCAccessPolicy) + + def start_rpc_server(self): + """Start the RPC server.""" + if self._server: + self._server.start() + + def stop_rpc_server(self): + """Stop the RPC server.""" + if self._server: + self._server.stop() + + def wait(self): + """Wait for message processing to complete (blocking).""" + if self._server: + self._server.wait() diff --git a/yardstick/common/messaging/payloads.py b/yardstick/common/messaging/payloads.py new file mode 100644 index 000000000..d29d79808 --- /dev/null +++ b/yardstick/common/messaging/payloads.py @@ -0,0 +1,53 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 six + +from yardstick.common import exceptions + + +@six.add_metaclass(abc.ABCMeta) +class Payload(object): + """Base Payload class to transfer data through the MQ service""" + + REQUIRED_FIELDS = {'version'} + + def __init__(self, **kwargs): + """Init method + + :param kwargs: (dictionary) attributes and values of the object + :returns: Payload object + """ + + if not all(req_field in kwargs for req_field in self.REQUIRED_FIELDS): + _attrs = set(kwargs) - self.REQUIRED_FIELDS + missing_attributes = ', '.join(str(_attr) for _attr in _attrs) + raise exceptions.PayloadMissingAttributes( + missing_attributes=missing_attributes) + + for name, value in kwargs.items(): + setattr(self, name, value) + + self._fields = set(kwargs.keys()) + + def obj_to_dict(self): + """Returns a dictionary with the attributes of the object""" + return {field: getattr(self, field) for field in self._fields} + + @classmethod + def dict_to_obj(cls, _dict): + """Returns a Payload object built from the dictionary elements""" + return cls(**_dict) diff --git a/yardstick/common/messaging/producer.py b/yardstick/common/messaging/producer.py new file mode 100644 index 000000000..b6adc0c17 --- /dev/null +++ b/yardstick/common/messaging/producer.py @@ -0,0 +1,70 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 logging +import os + +from oslo_config import cfg +import oslo_messaging +import six + +from yardstick.common import messaging + + +LOG = logging.getLogger(__name__) + + +@six.add_metaclass(abc.ABCMeta) +class MessagingProducer(object): + """Abstract class to implement a MQ producer + + This abstract class allows a class implementing this interface to publish + messages in a message queue. + """ + + def __init__(self, topic, pid=os.getpid(), fanout=True): + """Init function. + + :param topic: (string) MQ exchange topic + :param pid: (int) PID of the process implementing this MQ Notifier + :param fanout: (bool) MQ clients may request that a copy of the message + be delivered to all servers listening on a topic by + setting fanout to ``True``, rather than just one of them + :returns: `MessagingNotifier` class object + """ + self._topic = topic + self._pid = pid + self._fanout = fanout + self._transport = oslo_messaging.get_rpc_transport( + cfg.CONF, url=messaging.TRANSPORT_URL) + self._target = oslo_messaging.Target(topic=topic, fanout=fanout, + server=messaging.SERVER) + self._notifier = oslo_messaging.RPCClient(self._transport, + self._target) + + def send_message(self, method, payload): + """Send a cast message, that will invoke a method without blocking. + + The cast() method is used to invoke an RPC method that does not return + a value. cast() RPC requests may be broadcast to all Servers listening + on a given topic by setting the fanout Target property to ``True``. + + :param methos: (string) method name, that must be implemented in the + consumer endpoints + :param payload: (subclass `Payload`) payload content + """ + self._notifier.cast({'pid': self._pid}, + method, + **payload.obj_to_dict()) diff --git a/yardstick/tests/functional/common/messaging/__init__.py b/yardstick/tests/functional/common/messaging/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/yardstick/tests/functional/common/messaging/__init__.py diff --git a/yardstick/tests/functional/common/messaging/test_messaging.py b/yardstick/tests/functional/common/messaging/test_messaging.py new file mode 100644 index 000000000..96deeb35b --- /dev/null +++ b/yardstick/tests/functional/common/messaging/test_messaging.py @@ -0,0 +1,94 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 multiprocessing +import os +import time + +from yardstick.common.messaging import consumer +from yardstick.common.messaging import payloads +from yardstick.common.messaging import producer +from yardstick.tests.functional import base + + +TOPIC = 'topic_MQ' +METHOD_INFO = 'info' + + +class DummyPayload(payloads.Payload): + REQUIRED_FIELDS = {'version', 'data'} + + +class DummyEndpoint(consumer.NotificationHandler): + + def info(self, ctxt, **kwargs): + if ctxt['pid'] == self._ctx_pid: + self._queue.put('ID {}, data: {}'.format(self._id, kwargs['data'])) + + +class DummyConsumer(consumer.MessagingConsumer): + + def __init__(self, id, ctx_pid, queue): + self._id = id + endpoints = [DummyEndpoint(id, ctx_pid, queue)] + super(DummyConsumer, self).__init__(TOPIC, ctx_pid, endpoints) + + +class DummyProducer(producer.MessagingProducer): + pass + + +def _run_consumer(id, ctx_pid, queue): + _consumer = DummyConsumer(id, ctx_pid, queue) + _consumer.start_rpc_server() + _consumer.wait() + + +class MessagingTestCase(base.BaseFunctionalTestCase): + + @staticmethod + def _terminate_consumers(num_consumers, processes): + for i in range(num_consumers): + processes[i].terminate() + + def test_run_five_consumers(self): + output_queue = multiprocessing.Queue() + num_consumers = 10 + ctx_id = os.getpid() + producer = DummyProducer(TOPIC, pid=ctx_id) + + processes = [] + for i in range(num_consumers): + processes.append(multiprocessing.Process( + name='consumer_{}'.format(i), + target=_run_consumer, + args=(i, ctx_id, output_queue))) + processes[i].start() + self.addCleanup(self._terminate_consumers, num_consumers, processes) + + time.sleep(2) # Let consumers to create the listeners + producer.send_message(METHOD_INFO, DummyPayload(version=1, + data='message 0')) + producer.send_message(METHOD_INFO, DummyPayload(version=1, + data='message 1')) + time.sleep(2) # Let consumers attend the calls + + output = [] + while not output_queue.empty(): + output.append(output_queue.get(True, 1)) + + self.assertEqual(num_consumers * 2, len(output)) + for i in range(num_consumers): + self.assertIn('ID {}, data: {}'.format(1, 'message 0'), output) + self.assertIn('ID {}, data: {}'.format(1, 'message 1'), output) diff --git a/yardstick/tests/unit/common/messaging/__init__.py b/yardstick/tests/unit/common/messaging/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/yardstick/tests/unit/common/messaging/__init__.py diff --git a/yardstick/tests/unit/common/messaging/test_consumer.py b/yardstick/tests/unit/common/messaging/test_consumer.py new file mode 100644 index 000000000..612dcaecd --- /dev/null +++ b/yardstick/tests/unit/common/messaging/test_consumer.py @@ -0,0 +1,54 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 +import oslo_messaging + +from yardstick.common import messaging +from yardstick.common.messaging import consumer +from yardstick.tests.unit import base as ut_base + + +class TestEndPoint(object): + def action_1(self): + pass + + +class _MessagingConsumer(consumer.MessagingConsumer): + pass + + +class MessagingConsumerTestCase(ut_base.BaseUnitTestCase): + + def test__init(self): + with mock.patch.object(oslo_messaging, 'get_rpc_server') as \ + mock_get_rpc_server, \ + mock.patch.object(oslo_messaging, 'get_rpc_transport') as \ + mock_get_rpc_transport, \ + mock.patch.object(oslo_messaging, 'Target') as \ + mock_Target: + mock_get_rpc_transport.return_value = 'test_rpc_transport' + mock_Target.return_value = 'test_Target' + + _MessagingConsumer('test_topic', 'test_pid', [TestEndPoint], + fanout=True) + mock_get_rpc_transport.assert_called_once_with( + cfg.CONF, url=messaging.TRANSPORT_URL) + mock_Target.assert_called_once_with( + topic='test_topic', fanout=True, server=messaging.SERVER) + mock_get_rpc_server.assert_called_once_with( + 'test_rpc_transport', 'test_Target', [TestEndPoint], + executor=messaging.RPC_SERVER_EXECUTOR, + access_policy=oslo_messaging.DefaultRPCAccessPolicy) diff --git a/yardstick/tests/unit/common/messaging/test_payloads.py b/yardstick/tests/unit/common/messaging/test_payloads.py new file mode 100644 index 000000000..00ec220c9 --- /dev/null +++ b/yardstick/tests/unit/common/messaging/test_payloads.py @@ -0,0 +1,46 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 yardstick.common import exceptions +from yardstick.common.messaging import payloads +from yardstick.tests.unit import base as ut_base + + +class _DummyPayload(payloads.Payload): + REQUIRED_FIELDS = {'version', 'key1', 'key2'} + + +class PayloadTestCase(ut_base.BaseUnitTestCase): + + def test__init(self): + payload = _DummyPayload(version=1, key1='value1', key2='value2') + self.assertEqual(1, payload.version) + self.assertEqual('value1', payload.key1) + self.assertEqual('value2', payload.key2) + self.assertEqual(3, len(payload._fields)) + + def test__init_missing_required_fields(self): + with self.assertRaises(exceptions.PayloadMissingAttributes): + _DummyPayload(key1='value1', key2='value2') + + def test_obj_to_dict(self): + payload = _DummyPayload(version=1, key1='value1', key2='value2') + payload_dict = payload.obj_to_dict() + self.assertEqual({'version': 1, 'key1': 'value1', 'key2': 'value2'}, + payload_dict) + + def test_dict_to_obj(self): + _dict = {'version': 2, 'key1': 'value100', 'key2': 'value200'} + payload = _DummyPayload.dict_to_obj(_dict) + self.assertEqual(set(_dict.keys()), payload._fields) diff --git a/yardstick/tests/unit/common/messaging/test_producer.py b/yardstick/tests/unit/common/messaging/test_producer.py new file mode 100644 index 000000000..0289689dc --- /dev/null +++ b/yardstick/tests/unit/common/messaging/test_producer.py @@ -0,0 +1,46 @@ +# Copyright (c) 2018 Intel Corporation +# +# 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 +import oslo_messaging + +from yardstick.common import messaging +from yardstick.common.messaging import producer +from yardstick.tests.unit import base as ut_base + + +class _MessagingProducer(producer.MessagingProducer): + pass + + +class MessagingProducerTestCase(ut_base.BaseUnitTestCase): + + def test__init(self): + with mock.patch.object(oslo_messaging, 'RPCClient') as \ + mock_RPCClient, \ + mock.patch.object(oslo_messaging, 'get_rpc_transport') as \ + mock_get_rpc_transport, \ + mock.patch.object(oslo_messaging, 'Target') as \ + mock_Target: + mock_get_rpc_transport.return_value = 'test_rpc_transport' + mock_Target.return_value = 'test_Target' + + _MessagingProducer('test_topic', 'test_pid', fanout=True) + mock_get_rpc_transport.assert_called_once_with( + cfg.CONF, url=messaging.TRANSPORT_URL) + mock_Target.assert_called_once_with( + topic='test_topic', fanout=True, server=messaging.SERVER) + mock_RPCClient.assert_called_once_with('test_rpc_transport', + 'test_Target') |