Job scheduler: Add start time to scheduled job health info

Change-Id: I04dd4eb2e1214eed5cd997a1afa9a82d697e2454

Author: LarsMichelsen

cmk/gui/background_job/_models.py

@@ -3,7 +3,7 @@
 # This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
 # conditions defined in the file COPYING, which is part of this source code package.
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping
 
 from pydantic import BaseModel
 
@@ -56,7 +56,7 @@ class BackgroundJobsHealth(BaseModel, frozen=True):
 
 
 class ScheduledJobsHealth(BaseModel, frozen=True):
-    running_jobs: Sequence[str]
+    running_jobs: Mapping[str, int]
     job_executions: Mapping[str, int]
 
 

cmk/gui/job_scheduler/_background_jobs/_app.py

@@ -105,7 +105,10 @@ async def check_health(request: Request) -> HealthResponse:
                 job_executions=executor.job_executions(),
             ),
             scheduled_jobs=ScheduledJobsHealth(
-                running_jobs=list(filter_running_jobs(scheduler_state.running_jobs).keys()),
+                running_jobs={
+                    name: job.started_at
+                    for name, job in filter_running_jobs(scheduler_state.running_jobs).items()
+                },
                 job_executions=dict(scheduler_state.job_executions),
             ),
         )

cmk/gui/job_scheduler/_scheduler.py

@@ -122,14 +122,19 @@ def run_scheduled_jobs(
                 state.job_executions[job.name] += 1
                 if job.run_in_thread:
                     logger.debug("Starting [%s] in thread", job.name)
-                    state.running_jobs[job.name] = thread = threading.Thread(
-                        target=job_thread_main,
-                        args=(
-                            job,
-                            trace.Link(span.get_span_context()),
-                            crash_report_callback,
+                    state.running_jobs[job.name] = ScheduledJob(
+                        started_at=int(time.time()),
+                        thread=(
+                            thread := threading.Thread(
+                                target=job_thread_main,
+                                args=(
+                                    job,
+                                    trace.Link(span.get_span_context()),
+                                    crash_report_callback,
+                                ),
+                                name=f"scheduled-{job.name}",
+                            )
                         ),
-                        name=f"scheduled-{job.name}",
                     )
                     thread.start()
                     logger.debug("Started [%s]", job.name)
@@ -174,33 +179,37 @@ def job_thread_main(
 
 
 @tracer.start_as_current_span("wait_for_job_threads")
-def _wait_for_job_threads(running_jobs: dict[str, threading.Thread]) -> None:
+def _wait_for_job_threads(running_jobs: dict[str, ScheduledJob]) -> None:
     logger.debug("Waiting for threads to terminate")
-    for job_name, thread in list(running_jobs.items()):
-        thread.join()
+    for job_name, job in list(running_jobs.items()):
+        job.thread.join()
         del running_jobs[job_name]
 
 
-def _collect_finished_threads(running_jobs: dict[str, threading.Thread]) -> None:
-    for job_name, thread in list(running_jobs.items()):
-        if not thread.is_alive():
+def _collect_finished_threads(running_jobs: dict[str, ScheduledJob]) -> None:
+    for job_name, job in list(running_jobs.items()):
+        if not job.thread.is_alive():
             logger.debug("Removing finished thread [%s]", job_name)
             del running_jobs[job_name]
 
 
-def filter_running_jobs(
-    running_jobs: Mapping[str, threading.Thread],
-) -> dict[str, threading.Thread]:
+def filter_running_jobs(running_jobs: Mapping[str, ScheduledJob]) -> dict[str, ScheduledJob]:
     """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()}
+    return {job_id: job for job_id, job in running_jobs.items() if job.thread.is_alive()}
+
+
+@dataclass
+class ScheduledJob:
+    started_at: int
+    thread: threading.Thread
 
 
 @dataclass
 class SchedulerState:
-    running_jobs: dict[str, threading.Thread] = field(default_factory=dict)
+    running_jobs: dict[str, ScheduledJob] = field(default_factory=dict)
     job_executions: Counter[str] = field(default_factory=Counter)

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

@@ -28,7 +28,7 @@
     TerminateRequest,
 )
 from cmk.gui.job_scheduler._background_jobs._app import get_application
-from cmk.gui.job_scheduler._scheduler import SchedulerState
+from cmk.gui.job_scheduler._scheduler import ScheduledJob, SchedulerState
 
 logger = logging.getLogger(__name__)
 
@@ -106,8 +106,14 @@ def _get_test_client(loaded_at: int) -> TestClient:
             executor=DummyExecutor(logger),
             scheduler_state=SchedulerState(
                 running_jobs={
-                    "scheduled_1_running": DummyThread(is_stopped=False),
-                    "scheduled_2_finished": DummyThread(is_stopped=True),
+                    "scheduled_1_running": ScheduledJob(
+                        started_at=123,
+                        thread=DummyThread(is_stopped=False),
+                    ),
+                    "scheduled_2_finished": ScheduledJob(
+                        started_at=111,
+                        thread=DummyThread(is_stopped=True),
+                    ),
                 },
                 job_executions=Counter({"scheduled_1": 1, "scheduled_2": 2}),
             ),
@@ -184,7 +190,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_running"]
+    assert response.scheduled_jobs.running_jobs == {"scheduled_1_running": 123}
     assert response.scheduled_jobs.job_executions == {"scheduled_1": 1, "scheduled_2": 2}
 
 

tests/unit/cmk/gui/job_scheduler/test_job_scheduler.py

@@ -76,7 +76,7 @@ def test_run_scheduled_jobs_in_thread() -> None:
     run_scheduled_jobs(jobs, state, crash_report_callback=reraise_exception)
 
     assert "threaded_job" in state.running_jobs
-    state.running_jobs["threaded_job"].join()
+    state.running_jobs["threaded_job"].thread.join()
     assert called.is_set()
     assert state.job_executions == {"threaded_job": 1}
 
@@ -111,5 +111,5 @@ def test_run_scheduled_jobs_in_thread_does_not_start_twice(
     finally:
         shall_terminate.set()
         assert "threaded_job" in state.running_jobs
-        state.running_jobs["threaded_job"].join()
+        state.running_jobs["threaded_job"].thread.join()
         assert state.job_executions == {"threaded_job": 1}