Merge lp:~pwlars/snappy-proposed-selftest-agent/selftest-agent-retry into lp:snappy-proposed-selftest-agent

[Paul Larson]

Retry incoming messages if they have a bad payload a few times, but deadletter them if they still fail after a few attempts

Retry incoming messages if they have a bad payload a few times, but deadletter them if they still fail after a few attempts

[Joe Talbott (joetalbott)]

Take a look at my inline suggestion and let me know what you think.

Thanks for doing this work.

review: Needs Information

[lp:~pwlars/snappy-proposed-selftest-agent/selftest-agent-retry]

16. By Paul Larson on 2015-06-04

restructure the try block to make things more clear

[Paul Larson (pwlars)]

Good idea, I think it's easier to follow like that. Fixed.

review: Needs Resubmitting

[Joe Talbott (joetalbott)]

Thanks, looks great.

review: Approve

=== modified file 'snappy_proposed_selftest_agent/tests/test_worker.py'
--- snappy_proposed_selftest_agent/tests/test_worker.py	2015-05-29 13:41:30 +0000
+++ snappy_proposed_selftest_agent/tests/test_worker.py	2015-06-04 18:36:09 +0000
@@ -70,6 +70,27 @@
             ['foo', 'bar'],
             build_worker.messages_seen[0].get('image_binaries'))
 
+    def test_bad_candidates_message(self):
+        """ Insert a bad message and make sure it fails properly """
+        conn = kombu.Connection('memory:///')
+        queue_message(conn, {'test': 'value'})
+
+        build_worker = LoggingConsumer()
+        result_worker = LoggingConsumer()
+        config = read_config()
+        q = CoreSelftestAgentWorker(conn, config,
+                                    build_worker, result_worker)
+        # The message will be retried 3 times, and the 4th time we call
+        # consume it will go to the dead letter queue
+        max_retries = 3
+        for retry in range(max_retries+1):
+            next(q.consume(limit=1, timeout=5.0))
+            self.assertEqual(build_worker.messages_seen, [])
+            self.assertEqual(result_worker.messages_seen, [])
+        with conn.SimpleQueue(constants.DEAD_LETTER_QUEUE) as q:
+            message = q.get(block=False, timeout=5.0)
+            self.assertEqual(message.payload.get('test'), 'value')
+
 
 class LoggingConsumer(object):
     """A consumer callback object that acks and logs all received payloads."""

=== modified file 'snappy_proposed_selftest_agent/worker.py'
--- snappy_proposed_selftest_agent/worker.py	2015-05-29 13:41:30 +0000
+++ snappy_proposed_selftest_agent/worker.py	2015-06-04 18:36:09 +0000
@@ -37,6 +37,7 @@
         exchange = kombu.Exchange(
             constants.INPUT_EXCHANGE, type='fanout')
         self.queue = kombu.Queue(constants.INPUT_QUEUE, exchange)
+        self.max_retries = 3
 
     def get_consumers(self, Consumer, channel):
         """Return consumers instances for all configured queues."""
@@ -71,10 +72,29 @@
 
     def process(self, body, message):
         """ Process incoming messages about candidate packages """
+        try:
+            series = body['series']
+            source_binaries = body['source_binaries']
+        except KeyError:
+            retry_count = int(body.get('candidates_retry_count', '0'))
+            if retry_count < self.max_retries:
+                body['candidates_retry_count'] = retry_count + 1
+                with self.connection.SimpleQueue(self.queue) as q:
+                    q.put(body)
+            else:
+                logger.error(
+                    "Rejecting message due to retry count exceeding maximum. "
+                    "Message will reside on the dead letter queue",
+                )
+                with self.connection.SimpleQueue(
+                        constants.DEAD_LETTER_QUEUE) as q:
+                    q.put(body)
+            message.ack()
+            return
         # Dismiss if it's not the series (derived from ubuntu 'release')
         # this agent cares about.
         context_series = constants.RELEASE_MAP[constants.RELEASE]
-        if context_series != body['series']:
+        if context_series != series:
             message.ack()
             return
 
@@ -82,8 +102,9 @@
         # in the image manifest, it does not need testing.
         image_binaries = list(
             self.get_manifest_binaries(
-                body['source_binaries'], context_series, constants.ARCH)
+                source_binaries, context_series, constants.ARCH)
         )
+
         if not image_binaries:
             message.ack()
             return