Merge lp:~thomir-deactivatedaccount/core-image-publisher/trunk-use-uservice-queue into lp:core-image-publisher

[Thomi Richards]

Use the queue classes from uservice_utils.

Use the queue classes from uservice_utils, rather than the local copy. The version in uservice-utils has better debugging facilities and is better tested.

yay for deleting code!

[Joe Talbott (joetalbott)]

Doesn't the version of the uservice_utils library need to be updated someplace?

review: Needs Information

[Joe Talbott (joetalbott)]

Was done previously (Thomi informed me via IRC).

review: Approve

=== modified file 'core_image_publisher/__init__.py'
--- core_image_publisher/__init__.py	2015-04-03 02:06:10 +0000
+++ core_image_publisher/__init__.py	2015-04-08 20:17:07 +0000
@@ -21,10 +21,13 @@
 import os
 
 from uservice_utils.logging import configure_service_logging
+from uservice_utils.queue import (
+    SimpleRabbitQueueWorker,
+    DefaultRetryPolicy,
+)
 
 from core_image_publisher.queue import (
     create_connection_from_config,
-    CoreImagePublisherQueueMonitor,
     CoreImageResultPublisher,
 )
 from core_image_publisher import (
@@ -65,9 +68,15 @@
     try:
         with connection:
             publisher = CoreImageResultPublisher(connection)
-            monitor = CoreImagePublisherQueueMonitor(
+            retry_policy = DefaultRetryPolicy(
+                max_retries=3,
+                dead_queue='core.deadletters.{}'.format(constants.API_VERSION)
+            )
+            monitor = SimpleRabbitQueueWorker(
                 connection,
-                worker.ImagePublisherWorker(config, publisher)
+                'core.image.{}'.format(constants.API_VERSION),
+                worker.ImagePublisherWorker(config, publisher),
+                retry_policy,
             )
             monitor.run()
     except KeyboardInterrupt:

=== modified file 'core_image_publisher/queue.py'
--- core_image_publisher/queue.py	2015-03-31 16:45:10 +0000
+++ core_image_publisher/queue.py	2015-04-08 20:17:07 +0000
@@ -15,138 +15,20 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 
-import enum
 import logging
-from pdb import bdb
 
 import kombu
-from kombu.mixins import ConsumerMixin
 
 from core_image_publisher import constants
 
 logger = logging.getLogger(__name__)
 
 __all__ = [
-    'CoreImagePublisherQueueMonitor',
     'CoreImageResultPublisher',
     'create_connection_from_config',
 ]
 
 
-class MessageActions(enum.Enum):
-
-    """Actions that the message processing CB can request.
-
-    These enum values can be returned from the message processing callback,
-    and the CoreImagePublisherQueueMonitor will take appropriate action.
-
-    """
-
-    # The message needs to be retried. This might cause the message to be
-    # retried, or it might reject the message if it's retry count has been
-    # exceeded.
-    Retry = 1
-
-    # The message was processed properly, and should be acknowledged.
-    Acknowledge = 2
-
-
-class CoreImagePublisherQueueMonitor(ConsumerMixin):
-
-    """A class that watches for incoming messages from a queue, and calls
-    a callback to process them.
-
-    By using an external callback we can separate the stateful logic of
-    connecting to amqp from our business logic of dealing with the messages.
-
-    """
-
-    def __init__(self, connection, on_message_cb):
-        """Construct a new CoreImagePublisherQueueMonitor.
-
-        :param connection: A kombu.Connection instance.
-        :param on_message_db: A callable that will be called for every new
-            messages from the rabbitMQ queue(s). It must accept a single
-            parameter: the message object. The callback MUST either acknowledge
-            or reject or requeue the messsage, and MUST NOT raise any unhandled
-            exceptions.
-
-        """
-        self.connection = connection
-        self.on_message_cb = on_message_cb
-
-    def get_consumers(self, Consumer, channel):
-        """Return consumers instances for all configured queues."""
-        queues = [
-            kombu.Queue('core.image.{}'.format(constants.API_VERSION))
-        ]
-        return [Consumer(queues=queues, callbacks=[self.process])]
-
-    def process(self, body, message):
-        """Process incomming test request.
-
-        Run requested tests and posts results to the 'adt_results' queue
-        for later checking.
-        """
-        extra = constants.LOGGING_EXTRA.copy()
-        extra.update(body)
-        logger.info('Got: %r', body, extra=extra)
-        try:
-            requested_action = self.on_message_cb(body)
-            if requested_action == MessageActions.Retry:
-                self.maybe_requeue_message(message)
-            elif requested_action == MessageActions.Acknowledge:
-                message.ack()
-        # Make it possible to use a debugger inside the worker callback:
-        except bdb.BdbQuit:
-            raise
-        except Exception as e:
-            logger.error(
-                "Caught unhandled exception while processing message: %s",
-                e,
-                extra=extra
-            )
-            message.requeue()
-
-    def maybe_requeue_message(self, message):
-        """Maybe requeue the message, based on it's retry count.
-
-        Note: kombu (or maybe rabbitmq) does not allow you to alter a message's
-        payload and then call requeue(), which would be super helpful here.
-        Instead we explicitly queue it (using kombu.SimpleQueue) to the last
-        queue the message was on.
-
-        The key we use to store the retry count is based on the message's
-        routing key, which allows us to have a separate retry count for each
-        stage of the system.
-
-        """
-        extra = constants.LOGGING_EXTRA.copy()
-        extra.update(message.payload)
-        retry_key = "{}_retry_count".format(
-            message.delivery_info.get('routing_key', '')
-        )
-        retry_count = int(message.payload.get(retry_key, '0'))
-        if retry_count < constants.RETRY_COUNT:
-            message.payload[retry_key] = retry_count + 1
-            queue = self.connection.SimpleQueue(
-                'core.image.{}'.format(constants.API_VERSION))
-            queue.put(message.payload)
-            queue.close()
-            message.ack()
-        else:
-            logger.error(
-                "Rejecting message due to retry count exceeding maximum. "
-                "Message will reside on the dead letter queue",
-                extra=extra
-            )
-            queue = self.connection.SimpleQueue(
-                'core.deadletters.{}'.format(constants.API_VERSION))
-            queue.put(message.payload)
-            queue.close()
-            message.ack()
-
-
 def create_connection_from_config(config):
     """Create a connection given a config object.
 

=== removed file 'core_image_publisher/tests/test_queue_integration.py'
--- core_image_publisher/tests/test_queue_integration.py	2015-03-30 06:17:13 +0000
+++ core_image_publisher/tests/test_queue_integration.py	1970-01-01 00:00:00 +0000
@@ -1,73 +0,0 @@
-# core-image-publisher
-# Copyright (C) 2015 Canonical
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-#
-
-"""Integration tests for the kombu queue code."""
-
-import kombu
-import kombu.simple
-from testtools import TestCase
-
-
-from core_image_publisher.queue import CoreImagePublisherQueueMonitor
-
-
-class KombuQueueIntegrationTests(TestCase):
-
-    def test_can_read_message(self):
-        conn = kombu.Connection('memory:///')
-        queue_message(conn, {'test': 'value'})
-
-        consumer = LoggingConsumer()
-        q = CoreImagePublisherQueueMonitor(conn, consumer)
-        # pump the queue to get the enqueued message:
-        next(q.consume(limit=1, timeout=5.0))
-
-        self.assertEqual(consumer.got_messages, [dict(test='value')])
-
-    def test_uncaught_exceptions_cause_message_requeue(self):
-        conn = kombu.Connection('memory:///')
-        queue_message(conn, {'test': 'value'})
-
-        def consumer_with_bug(message):
-            raise RuntimeError("I am a bug, all up in ur bizniz!")
-
-        logging_consumer = LoggingConsumer()
-
-        queue_bugged = CoreImagePublisherQueueMonitor(conn, consumer_with_bug)
-        queue_ok = CoreImagePublisherQueueMonitor(conn, logging_consumer)
-
-        # pump the queue to get the enqueued message:
-        next(queue_bugged.consume(limit=1, timeout=5.0))
-        next(queue_ok.consume(limit=1, timeout=5.0))
-
-        self.assertEqual(logging_consumer.got_messages, [dict(test='value')])
-
-
-class LoggingConsumer(object):
-
-    """A consumer callback object that acks and logs all received payloads."""
-
-    def __init__(self):
-        self.got_messages = []
-
-    def __call__(self, payload):
-        self.got_messages.append(payload)
-
-
-def queue_message(conn, message):
-    q = kombu.simple.SimpleQueue(conn, 'core.image.v1')
-    q.put(message)

=== modified file 'core_image_publisher/worker.py'
--- core_image_publisher/worker.py	2015-04-02 14:10:24 +0000
+++ core_image_publisher/worker.py	2015-04-08 20:17:07 +0000
@@ -22,9 +22,10 @@
 import subprocess
 import tempfile
 
+from uservice_utils.queue import MessageActions
+
 from core_image_publisher.constants import LOGGING_EXTRA
 from core_image_publisher.cloud import get_glance_client
-from core_image_publisher.queue import MessageActions
 from core_image_publisher.utils import check_call