Merge lp:~thomir-deactivatedaccount/core-result-checker/trunk-detect-adt-status-and-republish-maybe into lp:core-result-checker

[Thomi Richards]

Requeue tests if infrastructure failed.

Re-queue test payloads into core.tess.v1 if the adt exit_code indicates an infrastructure error.

[Celso Providelo (cprov)]

review: Approve

[Francis Ginther (fginther)]

This looks wrong to me:

56 + q = self.connection.SimpleQueue(
57 + "core.tests.{}".format(constants.API_VERSION)
58 + )

Shouldn't the core.tests queue match the version that the previously ran this payload? Using the core-result-checker's API_VERSION would cause us to send messages to the wrong queue should the core-image-tester's API_VERSION get out of sync.

Do we actually need to have the name of the core-image-tester queue added to the payload so that the core-result-checker knows where to resend it?

review: Needs Information

[Celso Providelo (cprov)]

Francis,

result-checker only binds core.results.v1 queue, i.e. the v1 contract is common to all processed messages, thus re-posting on core.tests.v1 makes sense.

Versioned queues will only allow us to deploy new isolated solution pipelines (watcher->publisher->tester->checker) in the same amqp server, not necessarily mix them.

=== modified file 'core_result_checker/__init__.py'
--- core_result_checker/__init__.py	2015-04-06 23:23:22 +0000
+++ core_result_checker/__init__.py	2015-04-07 00:08:11 +0000
@@ -37,9 +37,10 @@
 
 class Worker(object):
 
-    def __init__(self, swift_publisher):
+    def __init__(self, swift_publisher, retry_publisher):
         get_logger(__name__).info("Service Started.")
         self.swift_publisher = swift_publisher
+        self.retry_publisher = retry_publisher
 
     def __call__(self, message):
         logger = get_logger(__name__, message)
@@ -48,10 +49,18 @@
             channel = message['channel']
             device = message['device']
             image_name = message['image_name']
+            exit_code = message['exit_code']
         except KeyError as e:
             logger.error("Unable to unpack incoming message: %s", str(e))
             return MessageActions.Retry
 
+        if int(exit_code) in (16, 20, 100):
+            logger.info(
+                "Test run infrastructure failed (exit code %s), retrying.",
+                exit_code
+            )
+            self.retry_publisher.retry_test_run(message)
+            return MessageActions.Acknowledge
         container_name = "core-{}-{}-{}".format(
             channel,
             device,
@@ -123,6 +132,42 @@
             )
 
 
+class RetryPublisher(object):
+
+    """A class that knows how to retry a test payload."""
+
+    def __init__(self, connection, max_retries):
+        self.connection = connection
+        self.max_retries = max_retries
+
+    def retry_test_run(self, payload):
+        """Maybe retry a test payload.
+
+        This will look at it's retry count, and either requeue it in the
+        test queue, or insert it into the dead letter queue.
+
+        """
+        retry_count = int(payload.get('test_run_retry_count', '0'))
+        if retry_count < self.max_retries:
+            q = self.connection.SimpleQueue(
+                "core.tests.{}".format(constants.API_VERSION)
+            )
+            payload['test_run_retry_count'] = retry_count + 1
+            q.put(payload)
+            q.close()
+        else:
+            logger = get_logger(__name__, payload)
+            logger.error(
+                "Test retry count exceeds maximum. Inserting into dead "
+                "letter queue"
+            )
+            q = self.connection.SimpleQueue(
+                "core.deadletters.{}".format(constants.API_VERSION)
+            )
+            q.put(payload)
+            q.close()
+
+
 def main():
     config = read_config()
     log_path = os.path.abspath(
@@ -133,14 +178,15 @@
     )
 
     amqp_uris = config.get('amqp', 'uris').split()
-    swift_publisher = SwiftPublisher(config['nova'])
-    worker = Worker(swift_publisher)
     retry_policy = DefaultRetryPolicy(
         max_retries=3,
         dead_queue='core.deadletters.{}'.format(constants.API_VERSION)
     )
     try:
         with kombu.Connection(amqp_uris) as connection:
+            swift_publisher = SwiftPublisher(config['nova'])
+            retry_publisher = RetryPublisher(connection, 3)
+            worker = Worker(swift_publisher, retry_publisher)
             queue_monitor = SimpleRabbitQueueWorker(
                 connection,
                 "core.result.{}".format(constants.API_VERSION),