Auto merge pull request #57: Auto cancel outdated pipelines

badwolf/builder.py

@@ -11,7 +11,7 @@
 from flask import current_app, render_template, url_for
 from requests.exceptions import ReadTimeout
 from docker import DockerClient
-from docker.errors import APIError, DockerException, ImageNotFound
+from docker.errors import APIError, DockerException, ImageNotFound, NotFound
 from markupsafe import Markup
 
 from badwolf.utils import to_text, to_binary, sanitize_sensitive_data
@@ -81,7 +81,10 @@ def run(self):
                 self.context.repository,
                 exit_code
             )
-            self.update_build_status('FAILED', '1 of 1 test failed')
+            if exit_code == 137:
+                self.update_build_status('FAILED', 'build cancelled')
+            else:
+                self.update_build_status('FAILED', '1 of 1 test failed')
 
         context.update({
             'logs': Markup(deansi.deansi(output)),
@@ -177,12 +180,20 @@ def run_in_container(self, docker_image_name):
         })
         environment.setdefault('TERM', 'xterm-256color')
         branch = self.context.source['branch']
+        labels = {
+            'repo': self.context.repository,
+            'commit': self.commit_hash,
+            'task_id': self.context.task_id,
+        }
         if self.context.type == 'tag':
             environment['BADWOLF_TAG'] = branch['name']
+            labels['tag'] = branch['name']
         else:
             environment['BADWOLF_BRANCH'] = branch['name']
+            labels['branch'] = branch['name']
         if self.context.pr_id:
             environment['BADWOLF_PULL_REQUEST'] = str(self.context.pr_id)
+            labels['pull_request'] = str(self.context.pr_id)
 
         volumes = {
             self.context.clone_path: {
@@ -207,11 +218,7 @@ def run_in_container(self, docker_image_name):
             privileged=self.spec.privileged,
             stdin_open=False,
             tty=True,
-            labels={
-                'repo': self.context.repository,
-                'commit': self.commit_hash,
-                'task_id': self.context.task_id,
-            }
+            labels=labels,
         )
         container_id = container.id
         logger.info('Created container %s from image %s', container_id, docker_image_name)
@@ -229,6 +236,8 @@ def run_in_container(self, docker_image_name):
             try:
                 output.append(to_text(container.logs()))
                 container.remove(force=True)
+            except NotFound:
+                pass
             except (APIError, DockerException, ReadTimeout):
                 logger.exception('Error removing docker container')
 
@@ -254,6 +263,10 @@ def send_notifications(self, context):
         with open(log_file, 'wb') as f:
             f.write(to_binary(html))
 
+        if exit_code == 137:
+            logger.info('Build cancelled, will not sending notification')
+            return
+
         if exit_code == 0:
             subject = 'Test succeed for repository {}'.format(self.context.repository)
         else:

badwolf/webhook/views.py

@@ -2,6 +2,7 @@
 import json
 import logging
 
+from docker import DockerClient
 from flask import Blueprint, request, current_app, url_for, jsonify
 
 from badwolf.context import Context
@@ -14,6 +15,7 @@
 blueprint = Blueprint('webhook', __name__)
 
 _EVENT_HANDLERS = {}
+_RUNNING_PIPELINES = {}
 
 
 def register_event_handler(event_key):
@@ -23,6 +25,46 @@ def register(func):
     return register
 
 
+def _cancel_outdated_pipelines(context):
+    from docker.errors import NotFound
+    docker = DockerClient(
+        base_url=current_app.config['DOCKER_HOST'],
+        timeout=current_app.config['DOCKER_API_TIMEOUT'],
+        version='auto',
+    )
+    containers = docker.containers.list(filters=dict(
+        status='running',
+        label='repo={}'.format(context.repository),
+    ))
+    if not containers:
+        return
+
+    for container in containers:
+        labels = container.labels
+        if context.type == 'tag':
+            continue
+        if context.pr_id and labels.get('pull_request') != context.pr_id:
+            continue
+        if context.type == 'branch' and labels.get('branch') != context.source['branch']['name']:
+            continue
+
+        task_id = labels.get('task_id')
+        if not task_id:
+            continue
+
+        future = _RUNNING_PIPELINES.get(task_id)
+        if not future or future.cancelled():
+            continue
+
+        logger.info('Cancelling outdated pipeline for %s with task_id %s', context.repository, task_id)
+        # cancel the future and remove the container
+        try:
+            container.remove(force=True)
+        except NotFound:
+            pass
+        future.cancel()
+
+
 @blueprint.route('/register/<user>/<repo>', methods=['POST'])
 def register_webhook(user, repo):
     full_name = '{}/{}'.format(user, repo)
@@ -116,7 +158,10 @@ def handle_repo_push(payload):
             source,
             rebuild=rebuild,
         )
-        start_pipeline.delay(context)
+        _cancel_outdated_pipelines(context)
+        future = start_pipeline.delay(context)
+        future.add_done_callback(lambda fut: _RUNNING_PIPELINES.pop(context.task_id, None))
+        _RUNNING_PIPELINES[context.task_id] = future
         if push_type == 'branch':
             check_pr_mergeable.delay(context)
 
@@ -158,7 +203,10 @@ def handle_pull_request(payload):
         rebuild=rebuild,
         pr_id=pr['id']
     )
-    start_pipeline.delay(context)
+    _cancel_outdated_pipelines(context)
+    future = start_pipeline.delay(context)
+    future.add_done_callback(lambda fut: _RUNNING_PIPELINES.pop(context.task_id, None))
+    _RUNNING_PIPELINES[context.task_id] = future
 
 
 @register_event_handler('pullrequest:approved')
@@ -281,4 +329,7 @@ def handle_pull_request_comment(payload):
         pr_id=pr['id'],
         nocache=nocache
     )
-    start_pipeline.delay(context)
+    _cancel_outdated_pipelines(context)
+    future = start_pipeline.delay(context)
+    future.add_done_callback(lambda fut: _RUNNING_PIPELINES.pop(context.task_id, None))
+    _RUNNING_PIPELINES[context.task_id] = future

tests/test_bitbucket_webhook_comment.py

@@ -1,13 +1,14 @@
 # -*- coding: utf-8 -*-
 import json
 import unittest.mock as mock
+from concurrent.futures import Future
 
 from flask import url_for
 
 
 @mock.patch('badwolf.webhook.views.start_pipeline')
 def test_repo_commit_comment_created_ci_retry(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -37,7 +38,7 @@ def test_repo_commit_comment_created_ci_retry(mock_start_pipeline, test_client):
 
 @mock.patch('badwolf.webhook.views.start_pipeline')
 def test_repo_commit_comment_created_ci_rebuild(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -67,7 +68,7 @@ def test_repo_commit_comment_created_ci_rebuild(mock_start_pipeline, test_client
 
 @mock.patch('badwolf.webhook.views.start_pipeline')
 def test_repo_commit_comment_created_do_nothing(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -95,9 +96,10 @@ def test_repo_commit_comment_created_do_nothing(mock_start_pipeline, test_client
     mock_start_pipeline.delay.assert_not_called()
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_pr_commit_comment_created_do_nothing(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_pr_commit_comment_created_do_nothing(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -128,9 +130,10 @@ def test_pr_commit_comment_created_do_nothing(mock_start_pipeline, test_client):
     mock_start_pipeline.delay.assert_not_called()
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_pr_commit_comment_created_ci_retry(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_pr_commit_comment_created_ci_retry(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -161,9 +164,10 @@ def test_pr_commit_comment_created_ci_retry(mock_start_pipeline, test_client):
     assert mock_start_pipeline.delay.called
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_pr_commit_comment_created_ci_rebuild(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_pr_commit_comment_created_ci_rebuild(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -194,9 +198,10 @@ def test_pr_commit_comment_created_ci_rebuild(mock_start_pipeline, test_client):
     assert mock_start_pipeline.delay.called
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_pr_commit_comment_created_ci_retry_state_not_open(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_pr_commit_comment_created_ci_retry_state_not_open(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',

tests/test_bitbucket_webhook_pr_created_updated.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import json
 import unittest.mock as mock
+from concurrent.futures import Future
 
 from flask import url_for
 
@@ -70,9 +71,10 @@ def test_pr_updated_state_not_open(test_client):
     assert to_text(res.data) == ''
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_pr_created_trigger_start_pipeline(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_pr_created_trigger_start_pipeline(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',

tests/test_bitbucket_webhook_repo_push.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import json
 import unittest.mock as mock
+from concurrent.futures import Future
 
 from flask import url_for
 
@@ -176,9 +177,10 @@ def test_repo_push_ci_skip_found(test_client):
     assert to_text(res.data) == ''
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_repo_push_trigger_start_pipeline(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_repo_push_trigger_start_pipeline(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',
@@ -213,9 +215,10 @@ def test_repo_push_trigger_start_pipeline(mock_start_pipeline, test_client):
     assert mock_start_pipeline.delay.called
 
 
+@mock.patch('badwolf.webhook.views._cancel_outdated_pipelines')
 @mock.patch('badwolf.webhook.views.start_pipeline')
-def test_repo_push_tag_trigger_start_pipeline(mock_start_pipeline, test_client):
-    mock_start_pipeline.delay.return_value = None
+def test_repo_push_tag_trigger_start_pipeline(mock_start_pipeline, mock_cancel_pipelines, test_client):
+    mock_start_pipeline.delay.return_value = Future()
     payload = json.dumps({
         'repository': {
             'full_name': 'deepanalyzer/badwolf',