Merge lp:~doanac/ubuntu-ci-services-itself/runworker-basic into lp:ubuntu-ci-services-itself

[Andy Doan]

run-worker: create a new common run worker

This combines the functionality and error cases of the 3 run-workers
we have into a common base class. It also includes actual test cases
so that its much easier to work on changes.

Part 1 of a 3 part series:

In preparation for run-worker scripts that can be cancelled, i took a brief detour to create a common base class for run-workers that handles all the dirty details they were dealing with. This also has a nice side-effect of being written in a way that we can easily test the hard parts.

I'm trying to keep this in digestible chunks. The main thing I'm hoping people to see here is the "def _on_message(self, msg)" logic. I think it covers the error cases we know of for the bsbuilder, image-builder, and test-runner. And I *think* I've covered all those corners with tests.

To see what this will look like for the worker scripts, you might want to fast forward to part 3 of my series:

 <https://code.launchpad.net/~doanac/ubuntu-ci-services-itself/run-worker-conversion/+merge/210078>

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:342
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/351/
Executed test runs:

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/351/rebuild

review: Approve (continuous-integration)

[Vincent Ladeuil (vila)]

78 + if not store or not logger:
79 + return

I think we also want to not store the logger if there is nothing there. So:

  log_content = logger.getvalue()
  if not log_content:
      return

102 + amqp_cb, ret = self.handle_request(logger, params)
103 + self._store_worker_log(store, logger, ret)
104 + amqp_cb(trigger, ret)

Pfew, that's dense, took me some back and forth to understand:

ampq_cb is the callback for [n]acking the msg
ret is something I can't define by reading the MP ;-p

Can you document that in handle_request() ? That's what the AMQPWorker users
will need.

The try/except/finally in _on_message() is still big but it makes more sense now. I would love some more comments as this code is quite complex but it's not a blocker.

And that's all I have to say ?

Wow, I had to re-read a few bits but really the above are nits ;)

Nice.

+1

review: Approve

[lp:~doanac/ubuntu-ci-services-itself/runworker-basic]

343. By Andy Doan on 2014-03-10

review comments

fix some suggestions by vila

[Andy Doan (doanac)]

vila - i know you already +1'd this, but want to take a look at revno 343 to see if it helps sufficiently?

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:343
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/366/
Executed test runs:

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/uci-engine-ci/366/rebuild

review: Approve (continuous-integration)

[Vincent Ladeuil (vila)]

> vila - i know you already +1'd this, but want to take a look at revno 343 to
> see if it helps sufficiently?

Definitely.

review: Approve

=== added file 'ci-utils/ci_utils/amqp_worker.py'
--- ci-utils/ci_utils/amqp_worker.py	1970-01-01 00:00:00 +0000
+++ ci-utils/ci_utils/amqp_worker.py	2014-03-10 23:10:31 +0000
@@ -0,0 +1,124 @@
+# Ubuntu CI Engine
+# Copyright 2013, 2014 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import logging
+import os
+import StringIO
+import traceback
+import yaml
+
+from ci_utils import amqp_utils, data_store, dump_stack
+
+logging.basicConfig(level=logging.INFO)
+log = logging.getLogger(__name__)
+
+
+class AMQPWorker(object):
+    '''Base class that handles the more complex issues of a rabbit worker.
+
+    This allows a user to subclass this and provide a custom "handle_request"
+    method.
+    '''
+
+    def handle_request(self, logger, params):
+        '''To be implemented by the subclass to do their work
+
+        amqp_cb should either be one of:
+          * amqp_utils.progress_failed
+          * amqp_utils.progress_comple
+        :return: tuple(ampq_cb, return_value_dictionary)
+        '''
+        raise NotImplementedError
+
+    def main(self, queue):
+        dump_stack.install_stack_dump_signal()
+        config = amqp_utils.get_config()
+        if not config:
+            exit(1)  # the get_config code prints an error
+        amqp_utils.process_queue(config, queue, self._on_message)
+
+    def _create_worker_logger(self, trigger):
+        '''Create a logger that captures output into StingIO
+
+        This provides an easy way to let workers get free logging that will
+        get added to the ticket.
+        '''
+        log = logging.getLogger(trigger)
+        log.buffer = StringIO.StringIO()
+        logstream = logging.StreamHandler(log.buffer)
+        formatter = logging.Formatter(
+            '[%(asctime)s] %(name)s:%(levelname)s:%(message)s')
+        logstream.setFormatter(formatter)
+        logstream.setLevel(logging.INFO)
+        log.addHandler(logstream)
+        return log
+
+    def _create_data_store(self, ticket):
+        fname = os.path.join(os.path.dirname(__file__), '../../unit_config')
+        with open(fname) as f:
+            config = yaml.safe_load(f)
+            return data_store.create_for_ticket(ticket, config)
+
+    def _store_worker_log(self, store, logger, retval):
+        '''An exception safe mechanism to upload log files
+
+        This needs to catch any exception so that progress complete/fail
+        call can be made on the queue
+        '''
+        if not store or not logger:
+            return
+        content = logger.buffer.getvalue()
+        if not content:
+            log.info('no logging content from action, skipping upload')
+        try:
+            name = '{}.log'.format(logger.name)
+            url = store.put_file(name, content, 'text/plain')
+            retval.setdefault('artifacts', []).append({
+                'name': name,
+                'reference': url,
+                'type': 'LOGS',
+            })
+        except:
+            log.exception('unable to store worker log')
+
+    def _on_message(self, msg):
+        log.info('on_message: %s', msg.body)
+        ret = {}
+        store = logger = trigger = None
+        try:
+            params = json.loads(msg.body)
+            trigger = params['progress_trigger']
+            amqp_utils.progress_update(trigger, params)
+            store = self._create_data_store(params['ticket_id'])
+
+            logger = self._create_worker_logger(trigger)
+            amqp_cb, ret = self.handle_request(logger, params)
+            self._store_worker_log(store, logger, ret)
+            amqp_cb(trigger, ret)
+        except (KeyboardInterrupt, Exception) as e:
+            err = 'Unexpected exception occurred'
+            if isinstance(e, KeyboardInterrupt):
+                err = 'Worker terminated'
+            ret['error_message'] = err
+            ret['traceback'] = traceback.format_exc()
+            self._store_worker_log(store, logger, ret)
+            if trigger:
+                amqp_utils.progress_failed(trigger, ret)
+            log.exception(err)
+            if isinstance(e, KeyboardInterrupt):
+                raise  # re-raise so amqp_utils.process_queue can exit
+        finally:
+            msg.channel.basic_ack(msg.delivery_tag)

=== added file 'ci-utils/ci_utils/tests/test_amqp_worker.py'
--- ci-utils/ci_utils/tests/test_amqp_worker.py	1970-01-01 00:00:00 +0000
+++ ci-utils/ci_utils/tests/test_amqp_worker.py	2014-03-10 23:10:31 +0000
@@ -0,0 +1,171 @@
+# Ubuntu CI Engine
+# Copyright 2013 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+
+# This program is distributed in the hope that it will be useful, but
+# WITHOUT ANY WARRANTY; without even the implied warranties of
+# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+# PURPOSE.  See the GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import json
+import unittest
+
+import mock
+
+from ci_utils import amqp_worker
+
+
+class _worker(amqp_worker.AMQPWorker):
+    '''simple class to help mock the datastore api used in the worker'''
+    def __init__(self, amqp_utils, retval, exception=None, error_msg=None):
+        super(_worker, self).__init__()
+        self.store = mock.Mock()
+        if amqp_utils:
+            self.amqp_utils = amqp_utils
+        self.retval = retval
+        self.exception = exception
+        self.error_msg = error_msg
+
+    def _create_data_store(self, ticket):
+        return self.store
+
+    def handle_request(self, log, params):
+        log.info('called')
+        self.params = params
+        if self.exception:
+            raise self.exception(self.error_msg)
+        if self.error_msg:
+            return self.amqp_utils.progress_failed, self.retval
+        return self.amqp_utils.progress_completed, self.retval
+
+
+class TestAMQPWorker(unittest.TestCase):
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageSimple(self, amqp_utils):
+        '''Test simple on message logic'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+        })
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval)
+        worker._on_message(msg)
+        self.assertEqual(worker.params['param1'], 'foo')
+        self.assertEqual(worker.params['param2'], 42)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(0, amqp_utils.progress_failed.call_count)
+        self.assertEqual(1, worker.store.put_file.call_count)
+        amqp_utils.progress_completed.assert_called_once_with(
+            'queue-name', retval)
+        # assert_called_once_with works, but doesn't check the values
+        # added to retval
+        retval = amqp_utils.progress_completed.call_args[0][1]
+        self.assertEqual(1, len(retval['artifacts']))
+        self.assertEqual('queue-name.log', retval['artifacts'][0]['name'])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageFail(self, amqp_utils):
+        '''Test on message logic for a failure handled by the worker'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+        })
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval, None, True)
+        worker._on_message(msg)
+        self.assertEqual(worker.params['param1'], 'foo')
+        self.assertEqual(worker.params['param2'], 42)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        amqp_utils.progress_failed.assert_called_once_with(
+            'queue-name', retval)
+        self.assertEqual(0, amqp_utils.progress_completed.call_count)
+        # assert_called_once_with works, but doesn't check the values
+        # added to retval
+        retval = amqp_utils.progress_failed.call_args[0][1]
+        self.assertEqual(1, len(retval['artifacts']))
+        self.assertEqual('queue-name.log', retval['artifacts'][0]['name'])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageUnexpected(self, amqp_utils):
+        '''Test on message logic for an unexpected failure'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+        })
+        err = 'this is the worker error'
+        worker = _worker(None, None, RuntimeError, err)
+        worker._on_message(msg)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+        self.assertIn(
+            err, amqp_utils.progress_failed.call_args[0][1]['traceback'])
+        self.assertEqual(1, worker.store.put_file.call_count)
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testOnMessageKilled(self, amqp_utils):
+        '''Test on message logic for handling ctrl-c (upstart stopping)'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+        })
+        worker = _worker(None, None, KeyboardInterrupt)
+        with self.assertRaises(KeyboardInterrupt):
+            worker._on_message(msg)
+        self.assertEqual(1, amqp_utils.progress_update.call_count)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+        self.assertIn('KeyboardInterrupt',
+                      amqp_utils.progress_failed.call_args[0][1]['traceback'])
+        self.assertEqual(1, worker.store.put_file.call_count)
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testNoTicket(self, amqp_utils):
+        '''Ensure we can gracefully deal with a bad message in the queue'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'progress_trigger': 'queue-name',
+        })
+        worker = _worker(amqp_utils, 1)
+        worker._on_message(msg)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        self.assertEqual(1, amqp_utils.progress_failed.call_count)
+        self.assertEqual(
+            'queue-name', amqp_utils.progress_failed.call_args[0][0])
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    def testNoQueue(self, amqp_utils):
+        '''Ensure we can gracefully deal with a bad message in the queue'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+        })
+        worker = _worker(amqp_utils, 1)
+        worker._on_message(msg)
+        # all we can check is that the message got acked
+        msg.channel.basic_ack.assert_called_once_with('foo')