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