services/nodemanager/arvnodeman/config.py

   1 #!/usr/bin/env python
   2 # Copyright (C) The Arvados Authors. All rights reserved.
   3 #
   4 # SPDX-License-Identifier: AGPL-3.0
   5
   6 from __future__ import absolute_import, print_function
   7
   8 import ConfigParser
   9 import importlib
  10 import logging
  11 import sys
  12
  13 import arvados
  14 import httplib2
  15 import pykka
  16 from apiclient import errors as apierror
  17
  18 from .baseactor import BaseNodeManagerActor
  19
  20 from functools import partial
  21 from libcloud.common.types import LibcloudError
  22 from libcloud.common.exceptions import BaseHTTPError
  23
  24 # IOError is the base class for socket.error, ssl.SSLError, and friends.
  25 # It seems like it hits the sweet spot for operations we want to retry:
  26 # it's low-level, but unlikely to catch code bugs.
  27 NETWORK_ERRORS = (IOError,)
  28 ARVADOS_ERRORS = NETWORK_ERRORS + (apierror.Error,)
  29 CLOUD_ERRORS = NETWORK_ERRORS + (LibcloudError, BaseHTTPError)
  30
  31 actor_class = BaseNodeManagerActor
  32
  33 class NodeManagerConfig(ConfigParser.SafeConfigParser):
  34     """Node Manager Configuration class.
  35
  36     This a standard Python ConfigParser, with additional helper methods to
  37     create objects instantiated with configuration information.
  38     """
  39
  40     LOGGING_NONLEVELS = frozenset(['file'])
  41
  42     def __init__(self, *args, **kwargs):
  43         # Can't use super() because SafeConfigParser is an old-style class.
  44         ConfigParser.SafeConfigParser.__init__(self, *args, **kwargs)
  45         for sec_name, settings in {
  46             'Arvados': {'insecure': 'no',
  47                         'timeout': '15',
  48                         'jobs_queue': 'yes',
  49                         'slurm_queue': 'yes'
  50                     },
  51             'Daemon': {'min_nodes': '0',
  52                        'max_nodes': '1',
  53                        'poll_time': '60',
  54                        'max_poll_time': '300',
  55                        'poll_stale_after': '600',
  56                        'max_total_price': '0',
  57                        'boot_fail_after': str(sys.maxint),
  58                        'node_stale_after': str(60 * 60 * 2),
  59                        'watchdog': '600',
  60                        'node_mem_scaling': '0.95'},
  61             'Manage': {'address': '127.0.0.1',
  62                        'port': '-1',
  63                        'ManagementToken': ''},
  64             'Logging': {'file': '/dev/stderr',
  65                         'level': 'WARNING'}
  66         }.iteritems():
  67             if not self.has_section(sec_name):
  68                 self.add_section(sec_name)
  69             for opt_name, value in settings.iteritems():
  70                 if not self.has_option(sec_name, opt_name):
  71                     self.set(sec_name, opt_name, value)
  72
  73     def get_section(self, section, transformers={}, default_transformer=None):
  74         transformer_map = {
  75             int: self.getint,
  76             bool: self.getboolean,
  77             float: self.getfloat,
  78         }
  79         result = self._dict()
  80         for key, value in self.items(section):
  81             transformer = None
  82             if transformers.get(key) in transformer_map:
  83                 transformer = partial(transformer_map[transformers[key]], section)
  84             elif default_transformer in transformer_map:
  85                 transformer = partial(transformer_map[default_transformer], section)
  86             if transformer is not None:
  87                 try:
  88                     value = transformer(key)
  89                 except (TypeError, ValueError):
  90                     pass
  91             result[key] = value
  92         return result
  93
  94     def log_levels(self):
  95         return {key: getattr(logging, self.get('Logging', key).upper())
  96                 for key in self.options('Logging')
  97                 if key not in self.LOGGING_NONLEVELS}
  98
  99     def dispatch_classes(self):
 100         mod_name = 'arvnodeman.computenode.dispatch'
 101         if self.has_option('Daemon', 'dispatcher'):
 102             mod_name = '{}.{}'.format(mod_name,
 103                                       self.get('Daemon', 'dispatcher'))
 104         module = importlib.import_module(mod_name)
 105         return (module.ComputeNodeSetupActor,
 106                 module.ComputeNodeShutdownActor,
 107                 module.ComputeNodeUpdateActor,
 108                 module.ComputeNodeMonitorActor)
 109
 110     def new_arvados_client(self):
 111         if self.has_option('Daemon', 'certs_file'):
 112             certs_file = self.get('Daemon', 'certs_file')
 113         else:
 114             certs_file = None
 115         insecure = self.getboolean('Arvados', 'insecure')
 116         http = httplib2.Http(timeout=self.getint('Arvados', 'timeout'),
 117                              ca_certs=certs_file,
 118                              disable_ssl_certificate_validation=insecure)
 119         return arvados.api(version='v1',
 120                            host=self.get('Arvados', 'host'),
 121                            token=self.get('Arvados', 'token'),
 122                            insecure=insecure,
 123                            http=http)
 124
 125     def new_cloud_client(self):
 126         module = importlib.import_module('arvnodeman.computenode.driver.' +
 127                                          self.get('Cloud', 'provider'))
 128         driver_class = module.ComputeNodeDriver.DEFAULT_DRIVER
 129         if self.has_option('Cloud', 'driver_class'):
 130             d = self.get('Cloud', 'driver_class').split('.')
 131             mod = '.'.join(d[:-1])
 132             cls = d[-1]
 133             driver_class = importlib.import_module(mod).__dict__[cls]
 134         auth_kwargs = self.get_section('Cloud Credentials')
 135         if 'timeout' in auth_kwargs:
 136             auth_kwargs['timeout'] = int(auth_kwargs['timeout'])
 137         return module.ComputeNodeDriver(auth_kwargs,
 138                                         self.get_section('Cloud List'),
 139                                         self.get_section('Cloud Create'),
 140                                         driver_class=driver_class)
 141
 142     def node_sizes(self):
 143         """Finds all acceptable NodeSizes for our installation.
 144
 145         Returns a list of (NodeSize, kwargs) pairs for each NodeSize object
 146         returned by libcloud that matches a size listed in our config file.
 147         """
 148         all_sizes = self.new_cloud_client().list_sizes()
 149         size_kwargs = {}
 150         section_types = {
 151             'price': float,
 152             'preemptable': bool,
 153         }
 154         for sec_name in self.sections():
 155             sec_words = sec_name.split(None, 2)
 156             if sec_words[0] != 'Size':
 157                 continue
 158             size_spec = self.get_section(sec_name, section_types, int)
 159             if 'preemptable' not in size_spec:
 160                 size_spec['preemptable'] = False
 161             if 'instance_type' not in size_spec:
 162                 # Assume instance type is Size name if missing
 163                 size_spec['instance_type'] = sec_words[1]
 164             size_spec['id'] = sec_words[1]
 165             size_kwargs[sec_words[1]] = size_spec
 166         # EC2 node sizes are identified by id. GCE sizes are identified by name.
 167         matching_sizes = []
 168         for size in all_sizes:
 169             matching_sizes += [
 170                 (size, size_kwargs[s]) for s in size_kwargs
 171                 if size_kwargs[s]['instance_type'] == size.id
 172                 or size_kwargs[s]['instance_type'] == size.name
 173             ]
 174         return matching_sizes
 175
 176     def shutdown_windows(self):
 177         return [float(n)
 178                 for n in self.get('Cloud', 'shutdown_windows').split(',')]