-
Notifications
You must be signed in to change notification settings - Fork 474
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Move job scheduler code to dedicated modules
Change-Id: I5858041e4891c32966379755fefbab8ef23f2749
- Loading branch information
1 parent
128f7ad
commit 31b5a3b
Showing
5 changed files
with
252 additions
and
230 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,153 @@ | ||
| #!/usr/bin/env python3 | ||
| # Copyright (C) 2024 Checkmk GmbH - License: GNU General Public License v2 | ||
| # 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. | ||
|
|
||
| import datetime | ||
| import threading | ||
| import time | ||
| from collections.abc import Callable, Mapping, Sequence | ||
| from pathlib import Path | ||
|
|
||
| from cmk.ccc import store | ||
|
|
||
| from cmk.utils import paths | ||
|
|
||
| from cmk.gui.cron import cron_job_registry, CronJob | ||
| from cmk.gui.log import logger | ||
| from cmk.gui.session import SuperUserContext | ||
| from cmk.gui.utils.script_helpers import gui_context | ||
|
|
||
| from cmk import trace | ||
|
|
||
| tracer = trace.get_tracer() | ||
|
|
||
|
|
||
| def run_scheduler(crash_report_callback: Callable[[Exception], str]) -> None: | ||
| job_threads: dict[str, threading.Thread] = {} | ||
| while True: | ||
| cycle_start = time.time() | ||
| _collect_finished_threads(job_threads) | ||
|
|
||
| try: | ||
| run_scheduled_jobs(list(cron_job_registry.values()), job_threads, crash_report_callback) | ||
| except Exception as exc: | ||
| crash_msg = crash_report_callback(exc) | ||
| logger.error("Exception in scheduler (Crash ID: %s)", crash_msg, exc_info=True) | ||
|
|
||
| if (sleep_time := 60 - (time.time() - cycle_start)) > 0: | ||
| time.sleep(sleep_time) | ||
| _wait_for_job_threads(job_threads) | ||
|
|
||
|
|
||
| def _load_last_job_runs() -> dict[str, datetime.datetime]: | ||
| return { | ||
| ident: datetime.datetime.fromtimestamp(ts, tz=datetime.UTC) | ||
| for ident, ts in store.load_object_from_file( | ||
| Path(paths.var_dir) / "last_job_runs.mk", default={} | ||
| ).items() | ||
| } | ||
|
|
||
|
|
||
| def _save_last_job_runs(runs: Mapping[str, datetime.datetime]) -> None: | ||
| store.save_object_to_file( | ||
| Path(paths.var_dir) / "last_job_runs.mk", | ||
| {ident: dt.timestamp() for ident, dt in runs.items()}, | ||
| ) | ||
|
|
||
|
|
||
| def _jobs_to_run(jobs: Sequence[CronJob], job_runs: dict[str, datetime.datetime]) -> list[CronJob]: | ||
| return [ | ||
| job | ||
| for job in jobs | ||
| if job.name not in job_runs | ||
| or datetime.datetime.now(tz=datetime.UTC) >= job_runs[job.name] + job.interval | ||
| ] | ||
|
|
||
|
|
||
| @tracer.instrument() | ||
| def run_scheduled_jobs( | ||
| jobs: Sequence[CronJob], | ||
| job_threads: dict[str, threading.Thread], | ||
| crash_report_callback: Callable[[Exception], str], | ||
| ) -> None: | ||
| logger.debug("Starting cron jobs") | ||
|
|
||
| for job in _jobs_to_run(jobs, job_runs := _load_last_job_runs()): | ||
| try: | ||
| if job.name in job_threads: | ||
| logger.debug("Skipping [%s] as it is already running", job.name) | ||
| continue | ||
|
|
||
| with tracer.span( | ||
| f"run_cron_job[{job.name}]", | ||
| attributes={ | ||
| "cmk.gui.job_name": job.name, | ||
| "cmk.gui.job_run_in_thread": str(job.run_in_thread), | ||
| "cmk.gui.job_interval": str(job.interval.total_seconds()), | ||
| }, | ||
| ) as span: | ||
| if job.run_in_thread: | ||
| logger.debug("Starting [%s] in thread", job.name) | ||
| job_threads[job.name] = thread = threading.Thread( | ||
| target=job_thread_main, | ||
| args=( | ||
| job, | ||
| trace.Link(span.get_span_context()), | ||
| crash_report_callback, | ||
| ), | ||
| ) | ||
| thread.start() | ||
| logger.debug("Started [%s]", job.name) | ||
| else: | ||
| logger.debug("Starting [%s] unthreaded", job.name) | ||
| with gui_context(), SuperUserContext(): | ||
| job.callable() | ||
| logger.debug("Finished [%s]", job.name) | ||
| except Exception as exc: | ||
| crash_msg = crash_report_callback(exc) | ||
| logger.error( | ||
| "Exception in cron job (Job: %s Crash ID: %s)", job.name, crash_msg, exc_info=True | ||
| ) | ||
| job_runs[job.name] = datetime.datetime.now() | ||
| _save_last_job_runs(job_runs) | ||
|
|
||
| logger.debug("Finished all cron jobs") | ||
|
|
||
|
|
||
| def job_thread_main( | ||
| job: CronJob, origin_span: trace.Link, crash_report_callback: Callable[[Exception], str] | ||
| ) -> None: | ||
| try: | ||
| with ( | ||
| tracer.span( | ||
| f"job_thread_main[{job.name}]", | ||
| attributes={"cmk.gui.job_name": job.name}, | ||
| links=[origin_span], | ||
| ), | ||
| gui_context(), | ||
| SuperUserContext(), | ||
| ): | ||
| job.callable() | ||
| except Exception as exc: | ||
| crash_msg = crash_report_callback(exc) | ||
| logger.error( | ||
| "Exception in cron job thread (Job: %s Crash ID: %s)", | ||
| job.name, | ||
| crash_msg, | ||
| exc_info=True, | ||
| ) | ||
|
|
||
|
|
||
| @tracer.instrument() | ||
| def _wait_for_job_threads(job_threads: dict[str, threading.Thread]) -> None: | ||
| logger.debug("Waiting for threads to terminate") | ||
| for thread in job_threads.values(): | ||
| thread.join() | ||
|
|
||
|
|
||
| def _collect_finished_threads(job_threads: dict[str, threading.Thread]) -> None: | ||
| for job_name, thread in list(job_threads.items()): | ||
| if not thread.is_alive(): | ||
| logger.debug("Removing finished thread [%s]", job_name) | ||
| del job_threads[job_name] |
Oops, something went wrong.