Job scheduler: Filter finished scheduled jobs on health endpoint query

The housekeeping of finished jobs is done once a minute. In case a job
finished, the health endpoint still reported the job as running until
the next housekeeping cycle. This change adds a filter to the health
endpoint query to exclude finished jobs on query.

Change-Id: I36fcde536727c95236df0f3c668e61622a973ec6

Author: LarsMichelsen

cmk/gui/job_scheduler/_background_jobs/_app.py

@@ -29,7 +29,7 @@
     StartResponse,
     TerminateRequest,
 )
-from cmk.gui.job_scheduler._scheduler import SchedulerState
+from cmk.gui.job_scheduler._scheduler import filter_running_jobs, SchedulerState
 
 
 def get_application(
@@ -105,7 +105,7 @@ async def check_health(request: Request) -> HealthResponse:
                 job_executions=executor.job_executions(),
             ),
             scheduled_jobs=ScheduledJobsHealth(
-                running_jobs=list(scheduler_state.running_jobs.keys()),
+                running_jobs=list(filter_running_jobs(scheduler_state.running_jobs).keys()),
                 job_executions=dict(scheduler_state.job_executions),
             ),
         )

cmk/gui/job_scheduler/_scheduler.py

@@ -188,6 +188,18 @@ def _collect_finished_threads(running_jobs: dict[str, threading.Thread]) -> None
             del running_jobs[job_name]
 
 
+def filter_running_jobs(
+    running_jobs: Mapping[str, threading.Thread],
+) -> dict[str, threading.Thread]:
+    """Provide an up-to-date list of running jobs.
+
+    collect_finished_threads might have not been executed since a job finished, which
+    causes some lag in the update of scheduler_state.running_jobs. This function
+    does some ad-hoc filtering to get the correct list of running jobs.
+    """
+    return {job_id: thread for job_id, thread in running_jobs.items() if thread.is_alive()}
+
+
 @dataclass
 class SchedulerState:
     running_jobs: dict[str, threading.Thread] = field(default_factory=dict)

tests/unit/cmk/gui/job_scheduler/background_jobs/test_app.py

@@ -81,6 +81,15 @@ def on_scheduler_start(cls, executor: JobExecutor) -> None:
         HelloJob.on_scheduler_start_called = True
 
 
+class DummyThread(threading.Thread):
+    def __init__(self, is_stopped: bool) -> None:
+        super().__init__()
+        self._is_stopped = is_stopped
+
+    def is_alive(self) -> bool:
+        return not self._is_stopped
+
+
 def _get_test_client(loaded_at: int) -> TestClient:
     return TestClient(
         get_application(
@@ -96,7 +105,10 @@ def _get_test_client(loaded_at: int) -> TestClient:
             registered_jobs={"hello_job": HelloJob},
             executor=DummyExecutor(logger),
             scheduler_state=SchedulerState(
-                running_jobs={"scheduled_1": threading.Thread()},
+                running_jobs={
+                    "scheduled_1_running": DummyThread(is_stopped=False),
+                    "scheduled_2_finished": DummyThread(is_stopped=True),
+                },
                 job_executions=Counter({"scheduled_1": 1, "scheduled_2": 2}),
             ),
         )
@@ -172,7 +184,7 @@ def test_health_check() -> None:
     )
     assert response.background_jobs.running_jobs == {"job_id": 42}
     assert response.background_jobs.job_executions == {"job_1": 1, "job_2": 2}
-    assert response.scheduled_jobs.running_jobs == ["scheduled_1"]
+    assert response.scheduled_jobs.running_jobs == ["scheduled_1_running"]
     assert response.scheduled_jobs.job_executions == {"scheduled_1": 1, "scheduled_2": 2}