Source code for nemo_nowcast.scheduler

# Copyright 2016-2021 Doug Latornell, 43ravens

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at


# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

"""NEMO_Nowcast worker launch scheduler.
import logging
import logging.config
import os
import signal
import time

import schedule
import sentry_sdk
import zmq
import zmq.log.handlers

from nemo_nowcast import CommandLineInterface, Config, NextWorker

NAME = "scheduler"
logger = logging.getLogger(NAME)

context = zmq.Context()

[docs]def main(): """Set up and run the nowcast system worker launch scheduler. Set-up includes: * Building the command-line parser, and parsing the command-line used to launch the scheduler * Reading and parsing the configuration file given on the command-line * Configuring the logging system as specified in the configuration file * Logging the scheduler's PID, and the file path/name that was used to configure it. * Install signal handlers for hangup, interrupt, and kill signals. The set-up is repeated if the scheduler process receives a HUP signal so that the configuration can be re-loaded without having to stop and re-start the scheduler. After the set-up is complete, start the scheduled worker launching loop. See :command:`python -m nowcast.scheduler --help` for details of the command-line interface. """ cli = CommandLineInterface(NAME, package="nemo_nowcast", description=__doc__) cli.build_parser() parsed_args = cli.parser.parse_args() config = Config() config.load(parsed_args.config_file) msg = _configure_logging(config)"running in process {os.getpid()}")"read config from {config.file}") _install_signal_handlers() run(config)
def _configure_logging(config): """Configure the scheduler's logging system interface. :param config: Nowcast system configuration. :type config: :py:class:`nemo_nowcast.config.Config` """ # Initialize exception logging to Sentry with client DSN URL from SENTRY_DSN envvar; # does nothing if SENTRY_DSN does not exist, is empty, or is not recognized by Sentry sentry_sdk.init() if "publisher" in config["logging"]: # Publish log messages to distributed logging aggregator logging_config = config["logging"]["publisher"] logging_config["handlers"]["zmq_pub"]["context"] = context host = config["zmq"]["host"] port = config["zmq"]["ports"]["logging"][NAME] addr = f"tcp://*:{port}" logging_config["handlers"]["zmq_pub"]["interface_or_socket"] = addr logging.config.dictConfig(logging_config) for handler in logger.root.handlers: if isinstance(handler, zmq.log.handlers.PUBHandler): handler.root_topic = NAME handler.formatters = { logging.DEBUG: logging.Formatter("%(message)s\n"), logging.INFO: logging.Formatter("%(message)s\n"), logging.WARNING: logging.Formatter("%(message)s\n"), logging.ERROR: logging.Formatter("%(message)s\n"), logging.CRITICAL: logging.Formatter("%(message)s\n"), } # Not sure why, but we need a brief pause before we start logging # messages time.sleep(0.25) msg = f"publishing logging messages to {addr}" else: # Write log messages to local file system # # Replace logging RotatingFileHandlers with WatchedFileHandlers so # that we notice when log files are rotated and switch to writing # to the new ones logging_config = config["logging"] logging_handlers = logging_config["handlers"] rotating_handler = "logging.handlers.RotatingFileHandler" watched_handler = "logging.handlers.WatchedFileHandler" for handler in logging_handlers: if logging_handlers[handler]["class"] == rotating_handler: logging_handlers[handler]["class"] = watched_handler del logging_handlers[handler]["backupCount"] logging.config.dictConfig(logging_config) msg = "writing logging messages to local file system" return msg def run(config): """Run the nowcast system worker launch scheduler. * Prepare the schedule as specified in the configuration file. * Loop forever, periodically checking to see if it is time to launch the scheduled workers. :param config: Nowcast system configuration. :type config: :py:class:`nemo_nowcast.config.Config` """ sleep_seconds = _prep_schedule(config) while True: schedule.run_pending() time.sleep(sleep_seconds) def _prep_schedule(config): """Create the schedule to launch workers and set how often it is checked. :param config: Nowcast system configuration. :type config: :py:class:`nemo_nowcast.config.Config` """ sleep_seconds = 60 try: for sched_item in config["scheduled workers"]: worker_module = list(sched_item.keys())[0] _create_scheduled_job(worker_module, sched_item[worker_module], config) except (AttributeError, KeyError): # Do nothing if scheduled workers config section is missing or empty pass return sleep_seconds def _create_scheduled_job(worker_module, params, config): try: args = params["cmd line opts"].split() except KeyError: args = [] worker = NextWorker(worker_module, args) job = ( schedule.every() .__getattribute__(params["every"]) .at(params["at"]) .do(worker.launch, config, NAME) ) return job def _install_signal_handlers(): """Set up hangup, interrupt, and kill signal handlers. """ def sighup_handler(signal, frame):"hangup signal (SIGHUP) received; reloading configuration") main() signal.signal(signal.SIGHUP, sighup_handler) def sigint_handler(signal, frame):"interrupt signal (SIGINT or Ctrl-C) received; shutting down") raise SystemExit signal.signal(signal.SIGINT, sigint_handler) def sigterm_handler(signal, frame):"termination signal (SIGTERM) received; shutting down") raise SystemExit signal.signal(signal.SIGTERM, sigterm_handler) if __name__ == "__main__": main() # pragma: no cover