Merge lp:~thomir-deactivatedaccount/core-image-publisher/trunk-upload-to-glance into lp:core-image-publisher

[Thomi Richards]

Add code to push converted images into glance.

Add code to push converted images into glance.

[lp:~thomir-deactivatedaccount/core-image-publisher/trunk-upload-to-glance]

18. By Thomi Richards on 2015-03-30

Fix typo.

19. By Thomi Richards on 2015-03-30

Further separate business logic from kombu objects.

[Celso Providelo (cprov)]

Thomi,

It is looking good, but I think we have to adjust maybe_requeue_message() behavior a bit, see the inline comments and let me know if the make sense to our plans.

[]

review: Needs Information

[lp:~thomir-deactivatedaccount/core-image-publisher/trunk-upload-to-glance]

20. By Thomi Richards on 2015-03-30

Update test cases.

[Francis Ginther (fginther)]

Just a few more comments and questions. Nothing that needs to be addressed now.

[lp:~thomir-deactivatedaccount/core-image-publisher/trunk-upload-to-glance]

21. By Thomi Richards on 2015-03-30

Fix typo.

22. By Thomi Richards on 2015-03-30

Put messages on the dead letter queue if their retry count is exceeded, rather than rejecting them.

[Thomi Richards (thomir-deactivatedaccount)]

I replied to most comments, and fixed the remaining issues. Please re-review!

[Celso Providelo (cprov)]

I am fine with landing this with a comment clarifying the use of the routing_key to get the queue_name dynamically on maybe_requeue().

I just don't see much value using this in the message producer-side if any consumers interested in that particular key will have to hardcode it.

I think we to balance the predilection for testable/readable code (modules, unittest-ing, etc) and the complexity and indirection it brings to the logic flow. It's a highly subjective topic ... let's keep talking about it while the code evolves.

review: Approve

=== modified file 'README.rst'
--- README.rst	2015-03-25 23:54:45 +0000
+++ README.rst	2015-03-30 20:05:33 +0000
@@ -57,6 +57,12 @@
   [amqp]
   uris = amqp://guest:guest@localhost:5672//
 
+  [nova]
+  os_username = foo
+  os_password = <redacted>
+  os_tenant_name = foo_project
+  os_auth_url = http://172.20.161.138:5000/v2.0/
+
 ...optionally, add a logstash section, which will turn on the logstash handler::
 
   [logstash]

=== modified file 'core-service.conf'
--- core-service.conf	2015-03-25 23:40:55 +0000
+++ core-service.conf	2015-03-30 20:05:33 +0000
@@ -1,3 +1,9 @@
 
 [amqp]
-uris = amqp://guest:guest@localhost:5672//
\ No newline at end of file
+uris = amqp://guest:guest@localhost:5672//
+
+[nova]
+os_username = foo
+os_password = <redacted>
+os_tenant_name = foo_project
+os_auth_url = http://172.20.161.138:5000/v2.0/

=== modified file 'core_image_publisher/__init__.py'
--- core_image_publisher/__init__.py	2015-03-27 04:26:08 +0000
+++ core_image_publisher/__init__.py	2015-03-30 20:05:33 +0000
@@ -52,6 +52,10 @@
     root_logger = logging.getLogger()
     root_logger.setLevel(logging.INFO)
 
+    # glanceclient & keystoneclient use requests, which logs for every
+    # new HTTP connection, so let's make it a little less chatty:
+    logging.getLogger('requests').setLevel(logging.WARNING)
+
     # If there is no ./logs directory, fallback to stderr.
     log_path = os.path.abspath(
         os.path.join(__file__, '../../logs/core-image-publisher.log'))
@@ -91,7 +95,7 @@
         with connection:
             monitor = CoreImagePublisherQueueMonitor(
                 connection,
-                worker.logging_worker
+                worker.ImagePublisherWorker(config)
             )
             monitor.run()
     except KeyboardInterrupt:

=== added file 'core_image_publisher/cloud.py'
--- core_image_publisher/cloud.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/cloud.py	2015-03-30 20:05:33 +0000
@@ -0,0 +1,40 @@
+# 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/>.
+#
+
+"""Functions for cloud interactions."""
+
+import glanceclient
+import keystoneclient.v2_0.client
+
+
+def get_glance_client(config):
+    keystone = keystoneclient.v2_0.client.Client(
+        username=config.get('nova', 'os_username'),
+        password=config.get('nova', 'os_password'),
+        tenant_name=config.get('nova', 'os_tenant_name'),
+        auth_url=config.get('nova', 'os_auth_url')
+    )
+    glance_endpoints = keystone.service_catalog.get_endpoints(
+        service_type='image'
+    )
+    glance_endpoint_url = glance_endpoints['image'][0]['internalURL']
+    glance_client = glanceclient.Client(
+        version='1',
+        endpoint=glance_endpoint_url,
+        token=keystone.auth_token
+    )
+    return glance_client

=== modified file 'core_image_publisher/constants.py'
--- core_image_publisher/constants.py	2015-03-25 21:36:10 +0000
+++ core_image_publisher/constants.py	2015-03-30 20:05:33 +0000
@@ -18,3 +18,5 @@
 """Constants for this service."""
 
 API_VERSION="v1"
+
+RETRY_COUNT = 3

=== modified file 'core_image_publisher/queue.py'
--- core_image_publisher/queue.py	2015-03-26 21:35:31 +0000
+++ core_image_publisher/queue.py	2015-03-30 20:05:33 +0000
@@ -15,7 +15,9 @@
 # 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
@@ -31,6 +33,24 @@
 ]
 
 
+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
@@ -70,7 +90,14 @@
         """
         logger.info('Got: %r', body, extra=body)
         try:
-            self.on_message_cb(message)
+            requested_action = self.on_message_cb(message.payload)
+            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",
@@ -79,6 +106,42 @@
             )
             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 requue(), 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.
+
+        """
+        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=message.payload
+            )
+            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.

=== modified file 'core_image_publisher/tests/test_queue_integration.py'
--- core_image_publisher/tests/test_queue_integration.py	2015-03-25 21:36:10 +0000
+++ core_image_publisher/tests/test_queue_integration.py	2015-03-30 20:05:33 +0000
@@ -65,9 +65,8 @@
     def __init__(self):
         self.got_messages = []
 
-    def __call__(self, message):
-        self.got_messages.append(message.payload)
-        message.ack()
+    def __call__(self, payload):
+        self.got_messages.append(payload)
 
 
 def queue_message(conn, message):

=== added file 'core_image_publisher/utils.py'
--- core_image_publisher/utils.py	1970-01-01 00:00:00 +0000
+++ core_image_publisher/utils.py	2015-03-30 20:05:33 +0000
@@ -0,0 +1,64 @@
+# 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/>.
+#
+
+"Utilities that might be useful elsewhere."
+
+import subprocess
+import textwrap
+
+
+def check_call(*popenargs, **kwargs):
+    """A better version of subprocess.check_call that gives better errors.
+
+    If a command invoked via 'check_call' fails, the error looks like::
+
+        Command '['juju', 'bootstrap']' returned non-zero exit status 2
+
+    While concise, that doesn't help the user of this script see what went
+    wrong. This function invokes a command, but captures stdout and stderr,
+    and includes that information in the exception.
+
+    """
+    if 'stdout' in kwargs or 'stderr' in kwargs:
+        # user did something with stdout or stderr already, so just call
+        # through to check_call directly:
+        return subprocess.check_call(*popenargs, **kwargs)
+    kwargs['stdout'] = kwargs['stderr'] = subprocess.PIPE
+    process = subprocess.Popen(*popenargs, **kwargs)
+    stdout, stderr = process.communicate()
+    if process.returncode != 0:
+        raise BetterCalledProcessError(
+            process.returncode,
+            popenargs[0],
+            textwrap.dedent(
+                '''
+                stdout:
+                """\\
+                {}"""
+                stderr:
+                """\\
+                {}"""
+                ''').format(stdout.decode(), stderr.decode())
+        )
+
+
+class BetterCalledProcessError(subprocess.CalledProcessError):
+
+    """Extend CalledProcessError to show failed command output as well."""
+
+    def __str__(self):
+        return super().__str__() + self.output

=== modified file 'core_image_publisher/worker.py'
--- core_image_publisher/worker.py	2015-03-29 21:12:41 +0000
+++ core_image_publisher/worker.py	2015-03-30 20:05:33 +0000
@@ -21,46 +21,71 @@
 import os
 import subprocess
 import tempfile
-from core_image_publisher.queue import enqueue_message
+
+from core_image_publisher.queue import (
+    enqueue_message,
+    MessageActions,
+)
+from core_image_publisher.cloud import get_glance_client
+from core_image_publisher.utils import check_call
 
 logger = logging.getLogger(__name__)
 
 
-def logging_worker(message):
-    logger.info("Got %r", message.payload, extra=message.payload)
-    payload = message.payload
-    try:
-        image_name = payload['image_name']
-        channel = payload['channel']
-        device = payload['device']
-    except KeyError as e:
-        logger.error(
-            "Unable to deserialize message payload - rejecting message: %s",
-            e,
-            extra=payload
-        )
-        message.requeue()
-        return
-
-    with tempfile.TemporaryDirectory as tmpdir:
-        try:
-            image_path = download_image(image_name, channel, device, tmpdir)
-        except subprocess.CalledProcessError as e:
-            logger.error("Unable to download core image: %s", e, extra=payload)
-            message.requeue()
-            return
-        try:
-            nova_image_path = convert_nova_image(image_path)
-        except subprocess.CalledProcessError as e:
-            logger.error("Unable to convert core image to qcow2 image: %s", e,
-                extra=payload)
-            message.requeue()
-            return
-        glance_image_name = upload_image_to_glance(nova_image_path)
-
-    payload['nova_image'] = glance_image_name
-    enqueue_message(payload)
-    message.ack()
+class ImagePublisherWorker(object):
+
+    """A worker callable that contains all our main logic."""
+
+    def __init__(self, config):
+        self._config = config
+
+    def __call__(self, payload):
+        logger.info("Got %r", payload, extra=payload)
+        try:
+            image_name = payload['image_name']
+            channel = payload['channel']
+            device = payload['device']
+        except KeyError as e:
+            logger.error(
+                "Unable to deserialize message payload - rejecting message: %s",
+                e,
+                extra=payload
+            )
+            return MessageActions.Retry
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            logger.info("Beginning image download.", extra=payload)
+            try:
+                image_path = download_image(image_name, channel, device, tmpdir)
+            except subprocess.CalledProcessError as e:
+                logger.error("Unable to download core image: %s", e, extra=payload)
+                return MessageActions.Retry
+            logger.info("Ubuntu Core image downloaded OK.", extra=payload)
+
+            try:
+                nova_image_path = convert_nova_image(image_path)
+            except subprocess.CalledProcessError as e:
+                logger.error("Unable to convert core image to qcow2 image: %s", e,
+                    extra=payload)
+                return MessageActions.Retry
+            logger.info("Image converted to qcow2 OK.", extra=payload)
+
+            try:
+                logger.info("Beginning image upload to glance...")
+                glance_image_name = upload_image_to_glance(
+                    nova_image_path,
+                    self._config
+                )
+            except Exception as e:
+                logger.error("Unable to upload image to glance: %s", e,
+                    extra=payload)
+                return MessageActions.Retry
+            logger.info("Image uploaded to glance OK.", extra=payload)
+
+        payload['nova_image'] = glance_image_name
+        enqueue_message(payload)
+        logger.info("Processing completed.", extra=payload)
+        return MessageActions.Acknowledge
 
 
 def download_image(name, channel, device, tmpdir):
@@ -70,13 +95,13 @@
            'ubuntu-device-flash',
            '--revision', name,
            'core',
-           '--device', generic_amd64,
+           '--device', device,
            '--channel', channel,
            '--size', '3',
            '-o', image_path,
            '--developer-mode',
            '--cloud']
-    subprocess.check_call(cmd)
+    check_call(cmd)
     return image_path
 
 
@@ -91,11 +116,21 @@
            '-O', 'qcow2',
            image_path,
            converted_image_path]
-    subprocess.check_call(cmd)
+    check_call(cmd)
     return converted_image_path
 
 
-def upload_image_to_glance(nova_image_path):
+def upload_image_to_glance(nova_image_path, config):
     """
     Upload the nova image to glance, returning the name of the image in glance.
     """
+    glance = get_glance_client(config)
+    image_name = os.path.basename(nova_image_path)
+    image = glance.images.create(
+        name=image_name,
+        data=open(nova_image_path, 'rb'),
+        disk_format='qcow2',
+        container_format='bare',
+    )
+    return image_name
+

=== modified file 'requirements.txt'
--- requirements.txt	2015-03-25 23:51:59 +0000
+++ requirements.txt	2015-03-30 20:05:33 +0000
@@ -1,2 +1,3 @@
 kombu==3.0.24
-python-logstash==0.4.2
\ No newline at end of file
+python-logstash==0.4.2
+python-glanceclient==0.17.0
\ No newline at end of file