oslo.messaging/oslo_messaging/_drivers/impl_pika.py

#    Copyright 2011 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 time

from oslo_config import cfg
from oslo_log import log as logging
import pika_pool
import retrying

from oslo_messaging._drivers import base
from oslo_messaging._drivers.pika_driver import pika_engine as pika_drv_engine
from oslo_messaging._drivers.pika_driver import pika_exceptions as pika_drv_exc
from oslo_messaging._drivers.pika_driver import pika_listener as pika_drv_lstnr
from oslo_messaging._drivers.pika_driver import pika_message as pika_drv_msg
from oslo_messaging._drivers.pika_driver import pika_poller as pika_drv_poller
from oslo_messaging import exceptions

LOG = logging.getLogger(__name__)

pika_opts = [
    cfg.IntOpt('channel_max', default=None,
               help='Maximum number of channels to allow'),
    cfg.IntOpt('frame_max', default=None,
               help='The maximum byte size for an AMQP frame'),
    cfg.IntOpt('heartbeat_interval', default=3,
               help="How often to send heartbeats for consumer's connections"),
    cfg.BoolOpt('ssl', default=None,
                help='Enable SSL'),
    cfg.DictOpt('ssl_options', default=None,
                help='Arguments passed to ssl.wrap_socket'),
    cfg.FloatOpt('socket_timeout', default=0.25,
                 help="Set socket timeout in seconds for connection's socket"),
    cfg.FloatOpt('tcp_user_timeout', default=0.25,
                 help="Set TCP_USER_TIMEOUT in seconds for connection's "
                      "socket"),
    cfg.FloatOpt('host_connection_reconnect_delay', default=0.25,
                 help="Set delay for reconnection to some host which has "
                      "connection error")
]

pika_pool_opts = [
    cfg.IntOpt('pool_max_size', default=30,
               help="Maximum number of connections to keep queued."),
    cfg.IntOpt('pool_max_overflow', default=0,
               help="Maximum number of connections to create above "
                    "`pool_max_size`."),
    cfg.IntOpt('pool_timeout', default=30,
               help="Default number of seconds to wait for a connections to "
                    "available"),
    cfg.IntOpt('pool_recycle', default=600,
               help="Lifetime of a connection (since creation) in seconds "
                    "or None for no recycling. Expired connections are "
                    "closed on acquire."),
    cfg.IntOpt('pool_stale', default=60,
               help="Threshold at which inactive (since release) connections "
                    "are considered stale in seconds or None for no "
                    "staleness. Stale connections are closed on acquire.")
]

notification_opts = [
    cfg.BoolOpt('notification_persistence', default=False,
                help="Persist notification messages."),
    cfg.StrOpt('default_notification_exchange',
               default="${control_exchange}_notification",
               help="Exchange name for for sending notifications"),
    cfg.IntOpt(
        'notification_listener_prefetch_count', default=100,
        help="Max number of not acknowledged message which RabbitMQ can send "
             "to notification listener."
    ),
    cfg.IntOpt(
        'default_notification_retry_attempts', default=-1,
        help="Reconnecting retry count in case of connectivity problem during "
             "sending notification, -1 means infinite retry."
    ),
    cfg.FloatOpt(
        'notification_retry_delay', default=0.25,
        help="Reconnecting retry delay in case of connectivity problem during "
             "sending notification message"
    )
]

rpc_opts = [
    cfg.IntOpt('rpc_queue_expiration', default=60,
               help="Time to live for rpc queues without consumers in "
                    "seconds."),
    cfg.StrOpt('default_rpc_exchange', default="${control_exchange}_rpc",
               help="Exchange name for sending RPC messages"),
    cfg.StrOpt('rpc_reply_exchange', default="${control_exchange}_rpc_reply",
               help="Exchange name for receiving RPC replies"),
    cfg.IntOpt(
        'rpc_listener_prefetch_count', default=100,
        help="Max number of not acknowledged message which RabbitMQ can send "
             "to rpc listener."
    ),
    cfg.IntOpt(
        'rpc_reply_listener_prefetch_count', default=100,
        help="Max number of not acknowledged message which RabbitMQ can send "
             "to rpc reply listener."
    ),
    cfg.IntOpt(
        'rpc_reply_retry_attempts', default=-1,
        help="Reconnecting retry count in case of connectivity problem during "
             "sending reply. -1 means infinite retry during rpc_timeout"
    ),
    cfg.FloatOpt(
        'rpc_reply_retry_delay', default=0.25,
        help="Reconnecting retry delay in case of connectivity problem during "
             "sending reply."
    ),
    cfg.IntOpt(
        'default_rpc_retry_attempts', default=-1,
        help="Reconnecting retry count in case of connectivity problem during "
             "sending RPC message, -1 means infinite retry. If actual "
             "retry attempts in not 0 the rpc request could be processed more "
             "then one time"
    ),
    cfg.FloatOpt(
        'rpc_retry_delay', default=0.25,
        help="Reconnecting retry delay in case of connectivity problem during "
             "sending RPC message"
    )
]


class PikaDriver(base.BaseDriver):
    def __init__(self, conf, url, default_exchange=None,
                 allowed_remote_exmods=None):
        opt_group = cfg.OptGroup(name='oslo_messaging_pika',
                                 title='Pika driver options')
        conf.register_group(opt_group)
        conf.register_opts(pika_opts, group=opt_group)
        conf.register_opts(pika_pool_opts, group=opt_group)
        conf.register_opts(rpc_opts, group=opt_group)
        conf.register_opts(notification_opts, group=opt_group)

        self._pika_engine = pika_drv_engine.PikaEngine(
            conf, url, default_exchange, allowed_remote_exmods
        )
        self._reply_listener = pika_drv_lstnr.RpcReplyPikaListener(
            self._pika_engine
        )
        super(PikaDriver, self).__init__(conf, url, default_exchange,
                                         allowed_remote_exmods)

    def require_features(self, requeue=False):
        pass

    def _declare_rpc_exchange(self, exchange, timeout):
        with (self._pika_engine.connection_without_confirmation_pool
                .acquire(timeout=timeout)) as conn:
            try:
                self._pika_engine.declare_exchange_by_channel(
                    conn.channel,
                    self._pika_engine.get_rpc_exchange_name(
                        exchange
                    ), "direct", False
                )
            except pika_pool.Timeout as e:
                raise exceptions.MessagingTimeout(
                    "Timeout for current operation was expired. {}.".format(
                        str(e)
                    )
                )

    def send(self, target, ctxt, message, wait_for_reply=None, timeout=None,
             retry=None):
        expiration_time = None if timeout is None else time.time() + timeout

        if retry is None:
            retry = self._pika_engine.default_rpc_retry_attempts

        exchange = self._pika_engine.get_rpc_exchange_name(
            target.exchange
        )

        def on_exception(ex):
            if isinstance(ex, pika_drv_exc.ExchangeNotFoundException):
                # it is desired to create exchange because if we sent to
                # exchange which is not exists, we get ChannelClosed exception
                # and need to reconnect
                try:
                    self._declare_rpc_exchange(exchange,
                                               expiration_time - time.time())
                except pika_drv_exc.ConnectionException as e:
                    LOG.warn("Problem during declaring exchange. %", e)
                return True
            elif isinstance(ex, (pika_drv_exc.ConnectionException,
                                 exceptions.MessageDeliveryFailure)):
                LOG.warn("Problem during message sending. %s", ex)
                return True
            else:
                return False

        retrier = (
            None if retry == 0 else
            retrying.retry(
                stop_max_attempt_number=(None if retry == -1 else retry),
                retry_on_exception=on_exception,
                wait_fixed=self._pika_engine.rpc_retry_delay * 1000,
            )
        )

        if target.fanout:
            return self.cast_all_servers(
                exchange, target.topic, ctxt, message, expiration_time,
                retrier
            )

        routing_key = self._pika_engine.get_rpc_queue_name(
            target.topic, target.server, retrier is None
        )

        msg = pika_drv_msg.RpcPikaOutgoingMessage(self._pika_engine, message,
                                                  ctxt)
        try:
            reply = msg.send(
                exchange=exchange,
                routing_key=routing_key,
                reply_listener=(
                    self._reply_listener if wait_for_reply else None
                ),
                expiration_time=expiration_time,
                retrier=retrier
            )
        except pika_drv_exc.ExchangeNotFoundException as ex:
            try:
                self._declare_rpc_exchange(exchange,
                                           expiration_time - time.time())
            except pika_drv_exc.ConnectionException as e:
                LOG.warn("Problem during declaring exchange. %", e)
            raise ex

        if reply is not None:
            if reply.failure is not None:
                raise reply.failure

            return reply.result

    def cast_all_servers(self, exchange, topic, ctxt, message, expiration_time,
                         retrier=None):
        msg = pika_drv_msg.PikaOutgoingMessage(self._pika_engine, message,
                                               ctxt)
        try:
            msg.send(
                exchange=exchange,
                routing_key=self._pika_engine.get_rpc_queue_name(
                    topic, "all_servers", retrier is None
                ),
                mandatory=False,
                expiration_time=expiration_time,
                retrier=retrier
            )
        except pika_drv_exc.ExchangeNotFoundException:
            try:
                self._declare_rpc_exchange(
                    exchange, expiration_time - time.time()
                )
            except pika_drv_exc.ConnectionException as e:
                LOG.warn("Problem during declaring exchange. %", e)

    def _declare_notification_queue_binding(self, target, timeout=None):
        if timeout is not None and timeout < 0:
            raise exceptions.MessagingTimeout(
                "Timeout for current operation was expired."
            )
        try:
            with (self._pika_engine.connection_without_confirmation_pool
                    .acquire)(timeout=timeout) as conn:
                self._pika_engine.declare_queue_binding_by_channel(
                    conn.channel,
                    exchange=(
                        target.exchange or
                        self._pika_engine.default_notification_exchange
                    ),
                    queue=target.topic,
                    routing_key=target.topic,
                    exchange_type='direct',
                    queue_expiration=None,
                    durable=self._pika_engine.notification_persistence,
                )
        except pika_pool.Timeout as e:
            raise exceptions.MessagingTimeout(
                "Timeout for current operation was expired. {}.".format(str(e))
            )

    def send_notification(self, target, ctxt, message, version, retry=None):
        if retry is None:
            retry = self._pika_engine.default_notification_retry_attempts

        def on_exception(ex):
            if isinstance(ex, (pika_drv_exc.ExchangeNotFoundException,
                               pika_drv_exc.RoutingException)):
                LOG.warn("Problem during sending notification. %", ex)
                try:
                    self._declare_notification_queue_binding(target)
                except pika_drv_exc.ConnectionException as e:
                    LOG.warn("Problem during declaring notification queue "
                             "binding. %", e)
                return True
            elif isinstance(ex, (pika_drv_exc.ConnectionException,
                                 pika_drv_exc.MessageRejectedException)):
                LOG.warn("Problem during sending notification. %", ex)
                return True
            else:
                return False

        retrier = retrying.retry(
            stop_max_attempt_number=(None if retry == -1 else retry),
            retry_on_exception=on_exception,
            wait_fixed=self._pika_engine.notification_retry_delay * 1000,
        )

        msg = pika_drv_msg.PikaOutgoingMessage(self._pika_engine, message,
                                               ctxt)
        return msg.send(
            exchange=(
                target.exchange or
                self._pika_engine.default_notification_exchange
            ),
            routing_key=target.topic,
            confirm=True,
            mandatory=True,
            persistent=self._pika_engine.notification_persistence,
            retrier=retrier
        )

    def listen(self, target):
        listener = pika_drv_poller.RpcServicePikaPoller(
            self._pika_engine, target,
            prefetch_count=self._pika_engine.rpc_listener_prefetch_count
        )
        listener.start()
        return listener

    def listen_for_notifications(self, targets_and_priorities, pool):
        listener = pika_drv_poller.NotificationPikaPoller(
            self._pika_engine, targets_and_priorities,
            prefetch_count=(
                self._pika_engine.notification_listener_prefetch_count
            ),
            queue_name=pool
        )
        listener.start()
        return listener

    def cleanup(self):
        self._reply_listener.cleanup()