feat: change workflow after clicking 'Apply changes' button (#82)

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Changed
 - add error handling for apply changes action
+- after clicking 'Apply changes' workflow is initially attempting to create new job immediately, if it is impossible, schedule it for the future
 
 ## [1.0.2]
 

backend/SC4SNMP_UI_backend/apply_changes/apply_changes.py

@@ -43,11 +43,13 @@ def __init__(self) -> None:
         mongo_config_collection.update_one(
             {
                 "previous_job_start_time": {"$exists": True},
-                "currently_scheduled": {"$exists": True}}
+                "currently_scheduled": {"$exists": True},
+                "task_id": {"$exists": True}}
             ,{
                 "$set":{
                     "previous_job_start_time": None,
-                    "currently_scheduled": False
+                    "currently_scheduled": False,
+                    "task_id": None
                 }
             },
             upsert=True

backend/SC4SNMP_UI_backend/apply_changes/handling_chain.py

@@ -2,7 +2,9 @@
 import ruamel.yaml
 from flask import current_app
 from SC4SNMP_UI_backend import mongo_client
-from SC4SNMP_UI_backend.apply_changes.tasks import run_job
+from SC4SNMP_UI_backend.apply_changes.tasks import run_job, get_job_config
+from SC4SNMP_UI_backend.apply_changes.kubernetes_job import create_job
+from kubernetes.client import ApiException
 import datetime
 import os
 
@@ -13,6 +15,7 @@
 VALUES_DIRECTORY = os.getenv("VALUES_DIRECTORY", "")
 VALUES_FILE = os.getenv("VALUES_FILE", "")
 KEEP_TEMP_FILES = os.getenv("KEEP_TEMP_FILES", "false")
+JOB_NAMESPACE = os.getenv("JOB_NAMESPACE", "sc4snmp")
 mongo_config_collection = mongo_client.sc4snmp.config_collection
 mongo_groups = mongo_client.sc4snmp.groups_ui
 mongo_inventory = mongo_client.sc4snmp.inventory_ui
@@ -138,31 +141,55 @@ def handle(self, request: dict = None):
         :return: pass dictionary with job_delay in seconds to the next handler
         """
         record = list(mongo_config_collection.find())[0]
-        last_update = record["previous_job_start_time"]
-        if last_update is None:
-            # If it's the first time that the job is run (record in mongo_config_collection has been created
-            # in ApplyChanges class and last_update attribute is None) then job delay should be equal to
-            # CHANGES_INTERVAL_SECONDS. Update the mongo record with job state accordingly.
-            job_delay = CHANGES_INTERVAL_SECONDS
+        schedule_new_job = True
+        # get_job_config return job configuration in "job" variable and BatchV1Api from kubernetes client
+        job, batch_v1 = get_job_config()
+        if job is None or batch_v1 is None:
+            raise ValueError("CheckJobHandler: Job configuration is empty")
+        try:
+            # Try creating a new kubernetes job immediately. If the previous job is still present in the namespace,
+            # ApiException will be thrown.
+            create_job(batch_v1, job, JOB_NAMESPACE)
+            task_id = record["task_id"]
+            if task_id is not None:
+                # revoke existing Celery task with the previously scheduled job
+                current_app.extensions["celery"].control.revoke(task_id,
+                   terminate=True, signal='SIGKILL')
             mongo_config_collection.update_one({"_id": record["_id"]},
-                                               {"$set": {"previous_job_start_time": datetime.datetime.utcnow()}})
-            # time from the last update
+                                               {"$set": {"previous_job_start_time": datetime.datetime.utcnow(),
+                                                         "currently_scheduled": False,
+                                                         "task_id": None}})
+            job_delay = 1
             time_difference = 0
-        else:
+            schedule_new_job = False
+        except ApiException:
             # Check how many seconds have elapsed since the last time that the job was run. If the time difference
-            # is greater than CHANGES_INTERVAL_SECONDS then job can be run immediately. Otherwise, calculate how
+            # is greater than CHANGES_INTERVAL_SECONDS then job can be scheduled within 1 second. Otherwise, calculate how
             # many seconds are left until minimum time difference between updates (CHANGES_INTERVAL_SECONDS).
-            current_time = datetime.datetime.utcnow()
-            delta = current_time - last_update
-            time_difference = delta.total_seconds()
-            if time_difference > CHANGES_INTERVAL_SECONDS:
-                job_delay = 1
+            last_update = record["previous_job_start_time"]
+            if last_update is None:
+                # If it's the first time that the job is run (record in mongo_config_collection has been created
+                # in ApplyChanges class and last_update attribute is None) but the previous job is still in the namespace
+                # then job delay should be equal to CHANGES_INTERVAL_SECONDS.
+                # Update the mongo record with job state accordingly.
+                job_delay = CHANGES_INTERVAL_SECONDS
+                mongo_config_collection.update_one({"_id": record["_id"]},
+                                                   {"$set": {"previous_job_start_time": datetime.datetime.utcnow()}})
+                # time from the last update
+                time_difference = 0
             else:
-                job_delay = int(CHANGES_INTERVAL_SECONDS - time_difference)
+                current_time = datetime.datetime.utcnow()
+                delta = current_time - last_update
+                time_difference = delta.total_seconds()
+                if time_difference > CHANGES_INTERVAL_SECONDS:
+                    job_delay = 1
+                else:
+                    job_delay = int(CHANGES_INTERVAL_SECONDS - time_difference)
 
         result = {
             "job_delay": job_delay,
-            "time_from_last_update": time_difference
+            "time_from_last_update": time_difference,
+            "schedule_new_job": schedule_new_job
         }
 
         current_app.logger.info(f"CheckJobHandler: {result}")
@@ -175,11 +202,11 @@ def handle(self, request: dict):
         ScheduleHandler schedules the kubernetes job with updated sc4snmp configuration
         """
         record = list(mongo_config_collection.find())[0]
-        if not record["currently_scheduled"]:
+        if not record["currently_scheduled"] and request["schedule_new_job"]:
             # If the task isn't currently scheduled, schedule it and update its state in mongo.
+            async_result = run_job.apply_async(countdown=request["job_delay"], queue='apply_changes')
             mongo_config_collection.update_one({"_id": record["_id"]},
-                                               {"$set": {"currently_scheduled": True}})
-            run_job.apply_async(countdown=request["job_delay"], queue='apply_changes')
+                                               {"$set": {"currently_scheduled": True, "task_id": async_result.id}})
             current_app.logger.info(
                 f"ScheduleHandler: scheduling new task with the delay of {request['job_delay']} seconds.")
         else:

backend/SC4SNMP_UI_backend/apply_changes/tasks.py

@@ -15,8 +15,11 @@
 JOB_CONFIG_PATH = os.getenv("JOB_CONFIG_PATH", "/config/job_config.yaml")
 celery_logger = get_task_logger(__name__)
 
-@shared_task()
-def run_job():
+def get_job_config():
+    """
+    :return: job - configuration of the job
+             batch_v1 - BatchV1Api object from kubernetes client
+    """
     job = None
     batch_v1 = None
     with open(JOB_CONFIG_PATH, encoding="utf-8") as file:
@@ -26,6 +29,13 @@ def run_job():
         config.load_incluster_config()
         batch_v1 = client.BatchV1Api()
         job = create_job_object(config_file)
+    return job, batch_v1
+
+@shared_task()
+def run_job():
+    job, batch_v1 = get_job_config()
+    if job is None or batch_v1 is None:
+        raise ValueError("Scheduled kubernetes job: Job configuration is empty")
 
     with MongoClient(MONGO_URI) as connection:
         try_creating = True
@@ -39,8 +49,9 @@ def run_job():
                 try:
                     record = list(connection.sc4snmp.config_collection.find())[0]
                     connection.sc4snmp.config_collection.update_one({"_id": record["_id"]},
-                                                 {"$set": {"previous_job_start_time": datetime.datetime.utcnow(),
-                                                           "currently_scheduled": False}})
+                                                   {"$set": {"previous_job_start_time": datetime.datetime.utcnow(),
+                                                             "currently_scheduled": False,
+                                                             "task_id": None}})
                 except Exception as e:
                     celery_logger.info(f"Error occurred while updating job state after job creation: {str(e)}")
             except ApiException:
@@ -50,6 +61,6 @@ def run_job():
                     celery_logger.info(f"Kubernetes job was not created. Max retries ({JOB_CREATION_RETRIES}) exceeded.")
                     record = list(connection.sc4snmp.config_collection.find())[0]
                     connection.sc4snmp.config_collection.update_one({"_id": record["_id"]},
-                                                                    {"$set": {"currently_scheduled": False}})
+                                                                    {"$set": {"currently_scheduled": False, "task_id": None}})
                 else:
-                    time.sleep(10)
+                    time.sleep(10)