feat: Add exit job functionality and enhance job error handling with JobKillSignalError

Author: lordgreg

examples/jobs/desktop_example.yaml

@@ -6,6 +6,13 @@ jobs:
     params:
       message: "Rclone isnt running"
 
+  - id: exit_job
+    name: Exit Job
+    enabled: false
+    plugin: exit
+    params:
+      code: 1
+
   - id: check_rclone_health
     name: Check Rclone Health
     plugin: binary
@@ -19,3 +26,5 @@ jobs:
           "
     on_failure:
       - notification_on_failure
+      # you can send kill switch to exit taskcrafter
+      # - exit_job

taskcrafter/exceptions/job.py

@@ -12,3 +12,7 @@ class JobFailedError(JobError):
 
 class JobValidationError(JobError):
     pass
+
+
+class JobKillSignalError(JobError):
+    pass

taskcrafter/job_loader.py

@@ -3,9 +3,13 @@
 from datetime import datetime
 from taskcrafter.exceptions.job import (
     JobFailedError,
+    JobKillSignalError,
     JobNotFoundError,
 )
-from taskcrafter.exceptions.plugin import PluginExecutionTimeoutError
+from taskcrafter.exceptions.plugin import (
+    PluginExecutionError,
+    PluginExecutionTimeoutError,
+)
 from taskcrafter.plugin_loader import plugin_execute
 from taskcrafter.logger import app_logger
 from taskcrafter.container import run_job_in_docker
@@ -162,7 +166,11 @@ def run_job(self, job: Job, execution_stack: list[str] = [], force: bool = False
                         queue_result = queue.get()
 
                         if isinstance(queue_result, Exception) or job.plugin == "exit":
-                            raise queue_result
+                            # add job to executed jobs since its kill switch afterwards.
+                            job.result.stop()
+                            job.result.set_status(JobStatus.ERROR)
+                            self.executed_jobs.append(deepcopy(job))
+                            raise JobKillSignalError(queue_result)
 
                     process.terminate()
 
@@ -185,7 +193,7 @@ def run_job(self, job: Job, execution_stack: list[str] = [], force: bool = False
                 job.result.set_status(JobStatus.ERROR)
                 break
 
-            except Exception as e:
+            except PluginExecutionError as e:
                 app_logger.error(
                     f"Job {job.id} executed with exception ({type(e)}): {e}"
                 )

taskcrafter/preview.py

@@ -79,7 +79,7 @@ def result_table(jobs: list[Job]):
             case JobStatus.SUCCESS:
                 status = Text(job.result.get_status().value, "green")
             case _:
-                continue
+                status = Text("n/a", "yellow")
 
         table.add_row(
             str(job.id),

taskcrafter/scheduler.py

@@ -11,6 +11,7 @@
     JobEvent,
 )
 from taskcrafter.exceptions.hook import HookNotFound
+from taskcrafter.exceptions.job import JobKillSignalError
 from taskcrafter.logger import app_logger
 from taskcrafter.job_loader import JobManager
 from taskcrafter.hook_loader import HookManager
@@ -36,7 +37,7 @@ def start_scheduler(self):
         self.scheduler.add_listener(self.event_listener_job, EVENT_ALL)
         self.scheduler.start()
 
-        app_logger.info("Scheduler started.")
+        app_logger.debug("Scheduler started.")
 
         # check and execute BEFORE_ALL hook
         self.schedule_hook_jobs(HookType.BEFORE_ALL)
@@ -47,7 +48,7 @@ def start_scheduler(self):
                 time.sleep(1)
         except (KeyboardInterrupt, SystemExit):
             self.stop_scheduler()
-            app_logger.info("Scheduler stopped.")
+            app_logger.debug("Scheduler stopped.")
 
     def event_listener_job(self, event):
         if isinstance(event, JobEvent):
@@ -57,6 +58,13 @@ def event_listener_job(self, event):
             self.schedule_hook_jobs(HookType.BEFORE_JOB, event)
         elif isinstance(event, JobExecutionEvent):
             if event.exception:
+                if isinstance(event.exception, JobKillSignalError):
+                    app_logger.warning(
+                        f"Job {job_id} is exit job, scheduler will be stopped."
+                    )
+                    self._event.set()
+                    return
+
                 self.schedule_hook_jobs(HookType.ON_ERROR, event)
                 app_logger.error(
                     f"scheduler: {event.job_id} failed with exception: {event.exception}"
@@ -73,13 +81,6 @@ def event_listener_job(self, event):
 
             self.schedule_hook_jobs(HookType.AFTER_JOB, event)
 
-            if self.job_manager.job_get_by_id(job_id).plugin == "exit":
-                app_logger.warning(
-                    f"Job {job_id} is exit job, scheduler will be stopped."
-                )
-                self._event.set()
-                return
-
             if self.job_manager.get_in_progress() == 0:
                 hook_executed = self.schedule_hook_jobs(HookType.AFTER_ALL, event)