Merge lp:~cprov/adt-cloud-worker/adt-runs into lp:adt-cloud-worker

[Celso Providelo]

Supporting nova.extra_args configuration for setting bootstack net-id (and other possible stack-dependant options) and also took the opportunity for refactoring some logic inside AdtNovaWorker consumer by passing the configuration object in. Now process & run_adt are more concise and easier to understand and test.

`adt-cloud-worker` works for libpng on trusty (using the test producer from lp:adt-request-proxy), now we have to store results on swift and call it feature complete.

Yes, you might have noticed we still have no unittests :-/

[Francis Ginther (fginther)]

Looks good. I first thought it a mistake to leave out "# TODO: do sensible things with the exit code:", but I believe this is really a task for the result-checker.

review: Approve

=== modified file '.adt-service.conf'
--- .adt-service.conf	2015-03-03 02:15:49 +0000
+++ .adt-service.conf	2015-03-03 16:37:14 +0000
@@ -12,5 +12,6 @@
 os_username = foo
 os_tenant_name = foo_project
 os_password = <redacted>
-os_auth_url = https://keystone.canonistack.canonical.com:443/v2.0/
-os_region_name = lcy01
+os_auth_url = http://172.20.161.138:5000/v2.0/
+os_region_name = bot-prototype
+extra_args = --net-id=415a0839-eb05-4e7a-907c-413c657f4bf5
 
=== modified file 'adt-cloud-worker.py'
--- adt-cloud-worker.py	2015-03-03 12:16:42 +0000
+++ adt-cloud-worker.py	2015-03-03 16:37:14 +0000
@@ -36,95 +36,102 @@
 logger = get_logger(__name__)
 
 
-class Worker(ConsumerMixin):
+class AdtNovaWorker(ConsumerMixin):
 
-    def __init__(self, name, connection, queues):
-        self.name = name
+    def __init__(self, connection, config):
         self.connection = connection
-        self.queues = queues
+        self.config = config
+        self.name = self.config.get('adt', 'name')
 
     def get_consumers(self, Consumer, channel):
-        return [Consumer(queues=self.queues, callbacks=[self.process])]
+        """Return consumers instances for all configured queues.""" 
+        exchange = kombu.Exchange("adt.exchange", type="topic")
+        queues = [
+            kombu.Queue(
+                'adt.requests.{}'.format(tag), exchange, routing_key=tag)
+            for tag in self.config.get('adt', 'tags').split()
+        ]
+        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.
+        """
         logger.info('Got: {}'.format(body))
         # TODO: body validation
-        # Ack message once it's valid or message.reject()
         message.ack()
-
-        result_dir = tempfile.mkdtemp(
-            prefix='adt-{}'.format(body['request_id']))
-
+        body['worker'] = self.name
+        # Run requested tests safely.
         try:
-            # what information do we need from the message?
-            adt_run_args = [
-                '--apt-source', body['package_name'],
-                '--output-dir', result_dir, # TODO: replace me
-            ]
-            if 'apt_pocket' in body:
-                adt_run_args += [
-                    '--apt-pocket', body['apt_pocket'],
-                    '--apt-upgrade',
-                ]
-            adt_ssh_nova_args = [
-                '--', 
-                '--flavor', body['nova_flavor'],
-                '--image', body['nova_image'],
-            ]
-
-            exit_code = run_adt(
-                adt_run_args +
-                [
-                    '---',
-                    'ssh',
-                    '-s', 'nova',
-                ] +
-                adt_ssh_nova_args
-            )
-
-            # TODO: do sensible things with the exit code:
-            # 0    all tests passed
-            # 2    at least one test skipped
-            # 4    at least one test failed
-            # 6    at least one test failed and at least one test skipped
-            # 8    no tests in this package
-            # 12   erroneous package
-            # 16   testbed failure
-            # 20   other unexpected failures including bad usage
-
+            # Create and temporary directory for storing adt results.
+            result_dir = tempfile.mkdtemp(
+                prefix='adt-{}-{}'.format(self.name, body['request_id']))
+            adt_kwargs = body.copy()
+            adt_kwargs['result_dir'] = result_dir
+            body['exit_code'] = self.run_adt(**adt_kwargs)
             # TODO: tar/gzip the output directory.
             # TODO: upload to swift instance at 'swift_container'
-
         except Exception as e:
+            # Unexpected failures ... 
             logger.error(e, exc_info=True)
+            body['exit_code'] = '100'
         finally:
-            # build the result message
-            body['worker'] = self.name
-            body['exit_code'] = exit_code
+            # Drop the result directory and post results
+            shutil.rmtree(result_dir)
             # TODO: send error logging to result-checker in the message
-
+            # TODO: resilience agains rabbit hiccups, otherwise the request
+            # will be lost.
             queue = self.connection.SimpleQueue('adt.results')
             queue.put(body)
             queue.close()
-            # Drop the result directory
-            # TODO: it could be possibly useful for clients.
-            shutil.rmtree(result_dir)
-
-
-def run_adt(arguments):
-    """Run adt-run with the given arguments.
-
-    Returns the exit code from adt-run.
-
-    """
-    # TODO: We probably want something more clever here:
-    try:
-        subprocess.check_call(['adt-run'] + arguments)
-    except subprocess.CalledProcessError as e:
-        # log?
-        # TODO: filter log content to avoid leaking cloud credentials.
-        return e.returncode
-    return 0
+
+    def run_adt(self, **kwargs):
+        """Run adt-run according to the given request parameters.
+
+        Always resets the environment with configured nova variables.
+
+        Returns the exit code from adt-run.
+        """
+        # TODO: We probably want something more clever here:
+
+        # what information do we need from the message?
+        adt_run_args = [
+            '--apt-source', kwargs['package_name'],
+            '--output-dir', kwargs['result_dir'],
+        ]
+        if 'apt_pocket' in kwargs:
+            adt_run_args += [
+                '--apt-pocket', kwargs['apt_pocket'],
+                '--apt-upgrade',
+        ]
+        adt_ssh_nova_args = [
+            '--',
+            '--flavor', kwargs['nova_flavor'],
+            '--image', kwargs['nova_image'],
+        ]
+
+        arguments = (
+            adt_run_args +
+            ['---', 'ssh', '-s', 'nova'] +
+            adt_ssh_nova_args +
+            self.config.get('nova', 'extra_args').split()
+        )
+
+        # Setup nova environment variables based on the configuration file.
+        for k, v in self.config.items('nova'):
+            if not k.startswith('os_'):
+                continue
+            os.environ[k.upper()] = str(v)
+
+        try:
+            subprocess.check_call(['adt-run'] + arguments)
+        except subprocess.CalledProcessError as e:
+            # log?
+            # TODO: filter log content to avoid leaking cloud credentials.
+            return e.returncode
+        return 0
 
 
 def main():
@@ -139,25 +146,11 @@
     # Load configuration options.
     config = configparser.ConfigParser()
     config.read(args.conf)
-    worker_name = config.get('adt', 'name')
-    routing_keys = config.get('adt', 'tags').split()
     amqp_uris = config.get('amqp', 'uris').split()
 
-    # Setup nova environment variables based on the configuration file.
-    for k, v in config.items('nova'):
-        os.environ[k.upper()] = str(v)
-
-    adt_exchange = kombu.Exchange("adt.exchange", type="topic")
-    queues = []
-    for routing_key in routing_keys:
-        queues.append(
-            kombu.Queue('adt.requests.{}'.format(routing_key) ,
-                        adt_exchange, routing_key=routing_key)
-        )
-
     with kombu.Connection(amqp_uris) as conn:
         try:
-            worker = Worker(worker_name, conn, queues)
+            worker = AdtNovaWorker(conn, config)
             worker.run()
         except KeyboardInterrupt:
             print('Bye!')