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

[Andy Doan]

run-worker: add cancelling logic

This provides "advisory" cancelling logic to the run-worker base. If
subclasses choose, they may periodically check the state of

   params.get('cancelled')

to see if they should clean up and exit

Part 2 of my 3 part series.

Adds logic for determining if a worker's corresponding jenkins job has been cancelled. The mechanism currently used is purely advisory for now (ie - the worker has the option to try and honor this or not).

We could enhance this to allow workers to specify they want a more forceful cancel option (ie sigterm ourselves or something). However, I think this is probably a better first step and keeps things easier to review.

I'm not a huge fan of tests that use time.sleep, but I really wanted to test this really worked without overly mocking it.

[PS Jenkins bot (ps-jenkins)]

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

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

review: Approve (continuous-integration)

[Vincent Ladeuil (vila)]

57 + # how often to check if the current task should be cancelled
58 + cancel_interval = 120

We'll probably want a config option there.

84 + t.start()
85 + return t

You don't check that the thread has really started (you would need an
additional even for that), we don't really care here.

158 + self.assertTrue(retval['cancelled'])

But you have a race here. That test should sleep at least as long the
cancel_interval or it could fail because the thread don't get enough time to
reach the code that set cancelled to True.

On the overall I'm not a big fan of that approach, I would rather see
SIGALARM be used to be able to raise an exception that could interrupt
handle_message.

Here, we're requiring collaboration from the request handler and delegating
the control to the url.

I approve nevertheless in case I don't get the use cases clearly and advisory is enough, happy to discuss more otherwise.

review: Approve

[Andy Doan (doanac)]

On 03/10/2014 01:14 PM, Vincent Ladeuil wrote:
> 158 + self.assertTrue(retval['cancelled'])
>
> But you have a race here. That test should sleep at least as long the
> cancel_interval or it could fail because the thread don't get enough time to
> reach the code that set cancelled to True.

Taking a more complete context:

153 + worker._on_message(msg)
154 + msg.channel.basic_ack.assert_called_once_with('foo')
155 + amqp_utils.progress_completed.assert_called_once_with(
156 + 'queue-name', retval)
157 + retval = amqp_utils.progress_completed.call_args[0][1]
158 + self.assertTrue(retval['cancelled'])

_on_message is actually a blocking call that will be dealing with the
threading. This test is just checking that we send a "canceled" value
back to the lander so that viewing the error logs would indicate this.

=== modified file 'ci-utils/ci_utils/amqp_worker.py'
--- ci-utils/ci_utils/amqp_worker.py	2014-03-09 06:27:58 +0000
+++ ci-utils/ci_utils/amqp_worker.py	2014-03-09 06:27:58 +0000
@@ -17,7 +17,9 @@
 import logging
 import os
 import StringIO
+import threading
 import traceback
+import urllib2
 import yaml
 
 from ci_utils import amqp_utils, data_store, dump_stack
@@ -26,6 +28,38 @@
 log = logging.getLogger(__name__)
 
 
+def _get(url):
+    try:
+        resp = urllib2.urlopen(url)
+        resp = resp.read()
+        return json.loads(resp)
+    except:
+        logging.exception('error checking url')
+        return {}
+
+
+class _timer(threading.Thread):
+    '''similar to threading.Timer, but repeats until cancel is called'''
+    def __init__(self, interval, cb):
+        super(_timer, self).__init__()
+        self.interval = interval
+        self.cb = cb
+        self.finished = threading.Event()
+        self.exit = False
+
+    def cancel(self):
+        """Stop the timer if it hasn't finished yet"""
+        self.finished.set()
+        self.exit = True
+
+    def run(self):
+        while not self.exit:
+            self.finished.wait(self.interval)
+            if not self.finished.is_set():
+                self.cb()
+        self.finished.set()
+
+
 class AMQPWorker(object):
     '''Base class that handles the more complex issues of a rabbit worker.
 
@@ -33,6 +67,9 @@
     method.
     '''
 
+    # how often to check if the current task should be cancelled
+    cancel_interval = 120
+
     def handle_request(self, logger, params):
         raise NotImplementedError
 
@@ -84,16 +121,38 @@
         except:
             log.exception('unable to store worker log')
 
+    def _handle_cancel(self, params):
+        '''provide an advisory mechansim to inform the worker to stop
+
+        This checks a URL to determine if the job has been cancelled. If so,
+        it updates the "params" object with a params['cancelled'] = True. Thus
+        this is an advisory mechansim that the worker can choose how/if to
+        deal with.
+        '''
+        url = params.get('cancel_url')
+        if not url:
+            return None
+
+        def check_url():
+            data = _get(url)
+            if not data.get('building', True):
+                params['cancelled'] = True
+        t = _timer(self.cancel_interval, check_url)
+        t.start()
+        return t
+
     def _on_message(self, msg):
         log.info('on_message: %s', msg.body)
         ret = {}
-        store = logger = trigger = None
+        store = logger = trigger = cancel_thread = 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'])
 
+            cancel_thread = self._handle_cancel(params)
+
             logger = self._create_worker_logger(trigger)
             amqp_cb, ret = self.handle_request(logger, params)
             self._store_worker_log(store, logger, ret)
@@ -112,3 +171,5 @@
                 raise  # re-raise so amqp_utils.process_queue can exit
         finally:
             msg.channel.basic_ack(msg.delivery_tag)
+            if cancel_thread:
+                cancel_thread.cancel()

=== modified file 'ci-utils/ci_utils/tests/test_amqp_worker.py'
--- ci-utils/ci_utils/tests/test_amqp_worker.py	2014-03-09 06:27:58 +0000
+++ ci-utils/ci_utils/tests/test_amqp_worker.py	2014-03-09 06:27:58 +0000
@@ -14,6 +14,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import json
+import time
 import unittest
 
 import mock
@@ -42,6 +43,9 @@
             raise self.exception(self.error_msg)
         if self.error_msg:
             return self.amqp_utils.progress_failed, self.retval
+        if params.get('cancel_url'):
+            time.sleep(0.1)
+            self.retval['cancelled'] = params['cancelled']
         return self.amqp_utils.progress_completed, self.retval
 
 
@@ -169,3 +173,46 @@
         worker._on_message(msg)
         # all we can check is that the message got acked
         msg.channel.basic_ack.assert_called_once_with('foo')
+
+    @mock.patch('ci_utils.amqp_worker.amqp_utils')
+    @mock.patch('ci_utils.amqp_worker._get')
+    def testCancel(self, get, amqp_utils):
+        '''Ensure workers get the "cancel" message'''
+        msg = mock.Mock()
+        msg.delivery_tag = 'foo'
+        msg.body = json.dumps({
+            'ticket_id': 1,
+            'progress_trigger': 'queue-name',
+            'param1': 'foo',
+            'param2': 42,
+            'cancel_url': 'foo'
+        })
+        get.return_value = {'building': False}
+        retval = {'msg': 'this is the worker return value'}
+        worker = _worker(amqp_utils, retval)
+        worker.cancel_interval = 0.01
+        worker._on_message(msg)
+        msg.channel.basic_ack.assert_called_once_with('foo')
+        amqp_utils.progress_completed.assert_called_once_with(
+            'queue-name', retval)
+        retval = amqp_utils.progress_completed.call_args[0][1]
+        self.assertTrue(retval['cancelled'])
+
+
+class TestTimer(unittest.TestCase):
+    def testCanCancel(self):
+        def cb():
+            raise RuntimeError('cancel failed')
+        t = amqp_worker._timer(1, cb)
+        t.start()
+        t.cancel()
+
+    def testCBRuns(self):
+        def cb():
+            self.run = True
+        self.run = False
+        t = amqp_worker._timer(0.01, cb)
+        t.start()
+        time.sleep(.02)
+        t.cancel()
+        self.assertTrue(self.run)