services/nodemanager/arvnodeman/launcher.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import absolute_import, print_function
   4
   5 import argparse
   6 import logging
   7 import signal
   8 import sys
   9 import time
  10
  11 import daemon
  12 import pykka
  13
  14 from . import config as nmconfig
  15 from .computenode import \
  16     ComputeNodeSetupActor, ComputeNodeShutdownActor, ComputeNodeUpdateActor, \
  17     ShutdownTimer
  18 from .daemon import NodeManagerDaemonActor
  19 from .jobqueue import JobQueueMonitorActor, ServerCalculator
  20 from .nodelist import ArvadosNodeListMonitorActor, CloudNodeListMonitorActor
  21 from .timedcallback import TimedCallBackActor
  22
  23 node_daemon = None
  24
  25 def abort(msg, code=1):
  26     print("arvados-node-manager: " + msg)
  27     sys.exit(code)
  28
  29 def parse_cli(args):
  30     parser = argparse.ArgumentParser(
  31         prog='arvados-node-manager',
  32         description="Dynamically allocate Arvados cloud compute nodes")
  33     parser.add_argument(
  34         '--foreground', action='store_true', default=False,
  35         help="Run in the foreground.  Don't daemonize.")
  36     parser.add_argument(
  37         '--config', help="Path to configuration file")
  38     return parser.parse_args(args)
  39
  40 def load_config(path):
  41     if not path:
  42         abort("No --config file specified", 2)
  43     config = nmconfig.NodeManagerConfig()
  44     try:
  45         with open(path) as config_file:
  46             config.readfp(config_file)
  47     except (IOError, OSError) as error:
  48         abort("Error reading configuration file {}: {}".format(path, error))
  49     return config
  50
  51 def setup_logging(path, level, **sublevels):
  52     handler = logging.FileHandler(path)
  53     handler.setFormatter(logging.Formatter(
  54             '%(asctime)s %(name)s[%(process)d] %(levelname)s: %(message)s',
  55             '%Y-%m-%d %H:%M:%S'))
  56     root_logger = logging.getLogger()
  57     root_logger.addHandler(handler)
  58     root_logger.setLevel(level)
  59     for logger_name, sublevel in sublevels.iteritems():
  60         sublogger = logging.getLogger(logger_name)
  61         sublogger.setLevel(sublevel)
  62
  63 def launch_pollers(config):
  64     cloud_client = config.new_cloud_client()
  65     arvados_client = config.new_arvados_client()
  66     cloud_size_list = config.node_sizes(cloud_client.list_sizes())
  67     if not cloud_size_list:
  68         abort("No valid node sizes configured")
  69
  70     server_calculator = ServerCalculator(
  71         cloud_size_list, config.getint('Daemon', 'max_nodes'))
  72     poll_time = config.getint('Daemon', 'poll_time')
  73     max_poll_time = config.getint('Daemon', 'max_poll_time')
  74
  75     timer = TimedCallBackActor.start(poll_time / 10.0).proxy()
  76     cloud_node_poller = CloudNodeListMonitorActor.start(
  77         cloud_client, timer, poll_time, max_poll_time).proxy()
  78     arvados_node_poller = ArvadosNodeListMonitorActor.start(
  79         arvados_client, timer, poll_time, max_poll_time).proxy()
  80     job_queue_poller = JobQueueMonitorActor.start(
  81         config.new_arvados_client(), timer, server_calculator,
  82         poll_time, max_poll_time).proxy()
  83     return timer, cloud_node_poller, arvados_node_poller, job_queue_poller
  84
  85 _caught_signals = {}
  86 def shutdown_signal(signal_code, frame):
  87     current_count = _caught_signals.get(signal_code, 0)
  88     _caught_signals[signal_code] = current_count + 1
  89     if node_daemon is None:
  90         pykka.ActorRegistry.stop_all()
  91         sys.exit(-signal_code)
  92     elif current_count == 0:
  93         node_daemon.shutdown()
  94     elif current_count == 1:
  95         pykka.ActorRegistry.stop_all()
  96     else:
  97         sys.exit(-signal_code)
  98
  99 def main(args=None):
 100     global node_daemon
 101     args = parse_cli(args)
 102     config = load_config(args.config)
 103
 104     if not args.foreground:
 105         daemon.DaemonContext().open()
 106     for sigcode in [signal.SIGINT, signal.SIGQUIT, signal.SIGTERM]:
 107         signal.signal(sigcode, shutdown_signal)
 108
 109     setup_logging(config.get('Logging', 'file'), **config.log_levels())
 110     timer, cloud_node_poller, arvados_node_poller, job_queue_poller = \
 111         launch_pollers(config)
 112     cloud_node_updater = ComputeNodeUpdateActor.start(
 113         config.new_cloud_client).proxy()
 114     node_daemon = NodeManagerDaemonActor.start(
 115         job_queue_poller, arvados_node_poller, cloud_node_poller,
 116         cloud_node_updater, timer,
 117         config.new_arvados_client, config.new_cloud_client,
 118         config.shutdown_windows(), config.getint('Daemon', 'max_nodes'),
 119         config.getint('Daemon', 'poll_stale_after'),
 120         config.getint('Daemon', 'node_stale_after')).proxy()
 121
 122     signal.pause()
 123     daemon_stopped = node_daemon.actor_ref.actor_stopped.is_set
 124     while not daemon_stopped():
 125         time.sleep(1)
 126     pykka.ActorRegistry.stop_all()
 127
 128
 129 if __name__ == '__main__':
 130     main()