From e02166f3553c60350ee440733915bdcbac6d1655 Mon Sep 17 00:00:00 2001 From: Ruslan Kamaldinov Date: Wed, 26 Mar 2014 18:38:45 +0400 Subject: [PATCH] Merged in murano-common In attempt to reduce number of Murano repositories to a sane minumum Murano community decided to merge in murano-common repo to projects dependent on it: * murano-api * murano-agent We hope that we will completely get rid of this code in Juno. Change-Id: Id203e62b040ddc2fe67399b2b487eaddca5f564f Partially-Implements: blueprint repository-reorganization --- python-agent/muranoagent/app.py | 2 +- python-agent/muranoagent/common/__init__.py | 0 .../muranoagent/common/messaging/__init__.py | 20 ++++ .../muranoagent/common/messaging/message.py | 58 ++++++++++ .../muranoagent/common/messaging/mqclient.py | 108 ++++++++++++++++++ .../common/messaging/subscription.py | 65 +++++++++++ python-agent/requirements.txt | 6 +- 7 files changed, 256 insertions(+), 3 deletions(-) create mode 100644 python-agent/muranoagent/common/__init__.py create mode 100644 python-agent/muranoagent/common/messaging/__init__.py create mode 100644 python-agent/muranoagent/common/messaging/message.py create mode 100644 python-agent/muranoagent/common/messaging/mqclient.py create mode 100644 python-agent/muranoagent/common/messaging/subscription.py diff --git a/python-agent/muranoagent/app.py b/python-agent/muranoagent/app.py index e240afb2..2d600b37 100644 --- a/python-agent/muranoagent/app.py +++ b/python-agent/muranoagent/app.py @@ -22,7 +22,7 @@ from execution_result import ExecutionResult from openstack.common import log as logging from openstack.common import service from config import CONF -from muranocommon.messaging import MqClient, Message +from muranoagent.common.messaging import MqClient, Message from exceptions import AgentException from time import sleep from bunch import Bunch diff --git a/python-agent/muranoagent/common/__init__.py b/python-agent/muranoagent/common/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python-agent/muranoagent/common/messaging/__init__.py b/python-agent/muranoagent/common/messaging/__init__.py new file mode 100644 index 00000000..2f85c16c --- /dev/null +++ b/python-agent/muranoagent/common/messaging/__init__.py @@ -0,0 +1,20 @@ +# Copyright (c) 2013 Mirantis Inc. +# +# 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 message import Message +from subscription import Subscription +from mqclient import MqClient + +__all__ = ['Message', 'Subscription', 'MqClient'] diff --git a/python-agent/muranoagent/common/messaging/message.py b/python-agent/muranoagent/common/messaging/message.py new file mode 100644 index 00000000..cec689c1 --- /dev/null +++ b/python-agent/muranoagent/common/messaging/message.py @@ -0,0 +1,58 @@ +# Copyright (c) 2013 Mirantis Inc. +# +# 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 anyjson +import logging + +LOG = logging.getLogger("murano-common.messaging") + + +class Message(object): + def __init__(self, connection=None, message_handle=None): + self._body = None + self._connection = connection + self._message_handle = message_handle + if message_handle: + self.id = message_handle.header.get('message_id') + else: + self.id = None + + try: + if message_handle: + self.body = anyjson.loads(message_handle.body) + else + self.body = None + except ValueError as e: + self.body = None + LOG.exception(e) + + @property + def body(self): + return self._body + + @body.setter + def body(self, value): + self._body = value + + @property + def id(self): + return self._id + + @id.setter + def id(self, value): + self._id = value or '' + + def ack(self): + self._message_handle.ack() diff --git a/python-agent/muranoagent/common/messaging/mqclient.py b/python-agent/muranoagent/common/messaging/mqclient.py new file mode 100644 index 00000000..6a701749 --- /dev/null +++ b/python-agent/muranoagent/common/messaging/mqclient.py @@ -0,0 +1,108 @@ +# Copyright (c) 2013 Mirantis Inc. +# +# 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 anyjson +import logging +import ssl as ssl_module + +from eventlet import patcher +kombu = patcher.import_patched('kombu') +from subscription import Subscription + + +log = logging.getLogger("murano-common.messaging") + + +class MqClient(object): + def __init__(self, login, password, host, port, virtual_host, + ssl=False, ca_certs=None): + ssl_params = None + + if ssl is True: + ssl_params = { + 'ca_certs': ca_certs, + 'cert_reqs': ssl_module.CERT_REQUIRED + } + + self._connection = kombu.Connection( + 'amqp://{0}:{1}@{2}:{3}/{4}'.format( + login, + password, + host, + port, + virtual_host + ), ssl=ssl_params + ) + self._channel = None + self._connected = False + + def __enter__(self): + self.connect() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + return False + + def connect(self): + self._connection.connect() + self._channel = self._connection.channel() + self._connected = True + + def close(self): + self._connection.close() + self._connected = False + + def declare(self, queue, exchange='', enable_ha=False, ttl=0): + if not self._connected: + raise RuntimeError('Not connected to RabbitMQ') + + queue_arguments = {} + if enable_ha is True: + # To use mirrored queues feature in RabbitMQ 2.x + # we need to declare this policy on the queue itself. + # + # Warning: this option has no effect on RabbitMQ 3.X, + # to enable mirrored queues feature in RabbitMQ 3.X, please + # configure RabbitMQ. + queue_arguments['x-ha-policy'] = 'all' + if ttl > 0: + queue_arguments['x-expires'] = ttl + + exchange = kombu.Exchange(exchange, type='direct', durable=True) + queue = kombu.Queue(queue, exchange, queue, durable=True, + queue_arguments=queue_arguments) + bound_queue = queue(self._connection) + bound_queue.declare() + + def send(self, message, key, exchange=''): + if not self._connected: + raise RuntimeError('Not connected to RabbitMQ') + + headers = {'message_id': str(message.id)} + + producer = kombu.Producer(self._connection) + producer.publish( + exchange=str(exchange), + routing_key=str(key), + body=anyjson.dumps(message.body), + headers=headers + ) + + def open(self, queue, prefetch_count=1): + if not self._connected: + raise RuntimeError('Not connected to RabbitMQ') + + return Subscription(self._connection, queue, prefetch_count) diff --git a/python-agent/muranoagent/common/messaging/subscription.py b/python-agent/muranoagent/common/messaging/subscription.py new file mode 100644 index 00000000..b42c0aeb --- /dev/null +++ b/python-agent/muranoagent/common/messaging/subscription.py @@ -0,0 +1,65 @@ +# Copyright (c) 2013 Mirantis Inc. +# +# 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 collections +import socket + +from eventlet import patcher +kombu = patcher.import_patched('kombu') +five = patcher.import_patched('kombu.five') +from . import message + + +class Subscription(object): + def __init__(self, connection, queue, prefetch_count=1): + self._buffer = collections.deque() + self._connection = connection + self._queue = kombu.Queue(name=queue, exchange=None) + self._consumer = kombu.Consumer(self._connection, auto_declare=False) + self._consumer.register_callback(self._receive) + self._consumer.qos(prefetch_count=prefetch_count) + + def __enter__(self): + self._consumer.add_queue(self._queue) + self._consumer.consume() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self._consumer is not None: + self._consumer.cancel() + return False + + def get_message(self, timeout=None): + msg_handle = self._get(timeout=timeout) + if msg_handle is None: + return None + return message.Message(self._connection, msg_handle) + + def _get(self, timeout=None): + elapsed = 0.0 + remaining = timeout + while True: + time_start = five.monotonic() + if self._buffer: + return self._buffer.pop() + try: + self._connection.drain_events(timeout=timeout and remaining) + except socket.timeout: + return None + elapsed += five.monotonic() - time_start + remaining = timeout and timeout - elapsed or None + + def _receive(self, message_data, message): + self._buffer.append(message) diff --git a/python-agent/requirements.txt b/python-agent/requirements.txt index dedd0651..e30fa2c6 100644 --- a/python-agent/requirements.txt +++ b/python-agent/requirements.txt @@ -1,7 +1,9 @@ -pbr>=0.5.21,<1.0 +pbr>=0.6,<1.0 semver bunch oslo.config>=1.2.0 -http://tarballs.openstack.org/murano-common/murano-common-release-0.3.tar.gz#egg=murano-common-release-0.3 iso8601 babel +anyjson>=0.3.3 +eventlet>=0.13.0 +kombu>=2.4.8