Ci busy no more

ci/webhook-receiver.py

@@ -13,6 +13,8 @@
 import os
 import subprocess
 import logging
+import time
+import threading
 
 # Global vars
 EVENT_URL = "/github-webhook"
@@ -23,6 +25,8 @@
 secret_file_name = None
 private_key_file = None
 secret = None
+queue_lock = None
+ci_request_list = None
 
 app = Flask(__name__)
 
@@ -158,6 +162,20 @@ def filter_to_prs_and_pr_comments(json):
 
     return None
 
+# run the manager, and rest of CI process
+def run_manager(request_ctx):
+    log_access_granted(request_ctx, "Running CI")
+    os.system("./manager.sh config {} \"{}\" \"{}\"".format(request_ctx['id'], request_ctx['repo'], request_ctx['body']))
+
+def poll_request_list():
+    while True:
+        # continuously check for waiting requests
+        if len(ci_request_list) > 0:
+            run_manager(ci_request_list.pop(0))
+        else:
+            # no requests, sleep
+            time.sleep(1)
+
 @app.route(EVENT_URL, methods=['POST'])
 def init_ci_pipeline():
     request_ctx = filter_to_prs_and_pr_comments(request.json)
@@ -196,13 +214,28 @@ def init_ci_pipeline():
     out, err = proc2.communicate()
 
     if (out):
-        print("Can't run CI, another CI run in progress")
-        log_access_granted(request_ctx, "CI busy, posting busy msg")
-        os.system("./ci_busy.sh config {} \"{}\" \"{}\" \"Another CI run in progress, please try again in 15 minutes\""
-                  .format(request_ctx['id'], request_ctx['repo'], request_ctx['body']))
+        print("Can't run CI, another CI run in progress, placing in the queue")
+        log_access_granted(request_ctx, "CI busy, placing request in queue")
+        duplicate_req = False
+        busy_msg = "Duplicate request already waiting, ignoring message"
+        # make sure we don't bypass thread safety
+        with queue_lock:
+            for req in ci_request_list:
+                # make sure this is the same PR
+                if req['id'] == request_ctx['id'] and req['repo'] == request_ctx['repo']:
+                    duplicate_req = True
+                    break
+            if not duplicate_req:
+                # don't have this request yet, put in queue
+                ci_request_list.append(request_ctx)
+                busy_msg = "Another CI run in progress, adding request to the end of the list"
+        # ending frees the lock
+        os.system("./ci_busy.sh config {} \"{}\" \"{}\" \"{}\""
+                  .format(request_ctx['id'], request_ctx['repo'], request_ctx['body'], busy_msg))
     else:
-        log_access_granted(request_ctx, "Running CI")
-        os.system("./manager.sh config {} \"{}\" \"{}\"".format(request_ctx['id'], request_ctx['repo'], request_ctx['body']))
+        # no manager running yet, put in the list (presumably the top)
+        with queue_lock:
+            ci_request_list.append(request_ctx)
 
     return jsonify({"status": "ONLINE"})
 
@@ -251,8 +284,15 @@ def parse_config(cfg_name):
     cfg_name = sys.argv[4]
 
     parse_config(cfg_name)
-
     secret = decrypt_secret()
+    queue_lock = threading.Lock()
+    ci_request_list = []
+
+    # dedicate thread to check for and run new requests
+    poll_request = threading.Thread(target=poll_request_list, args=[])
+    # run as daemon to catch ^C termination
+    poll_request.daemon = True
+    poll_request.start()
 
     logging.info("Starting the CI service")
     app.run(host=host, port=port)