plugins.d/python.d.plugin

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 # Description: netdata python modules supervisor
   5 # Author: Pawel Krupa (paulfantom)
   6
   7 import os
   8 import sys
   9 import time
  10 import threading
  11
  12 # -----------------------------------------------------------------------------
  13 # globals & environment setup
  14 # https://github.com/firehol/netdata/wiki/External-Plugins#environment-variables
  15 MODULE_EXTENSION = ".chart.py"
  16 BASE_CONFIG = {'update_every': os.getenv('NETDATA_UPDATE_EVERY', 1),
  17                'priority': 90000,
  18                'retries': 10}
  19
  20 MODULES_DIR = os.path.abspath(os.getenv('NETDATA_PLUGINS_DIR',
  21                                         os.path.dirname(__file__)) + "/../python.d") + "/"
  22 CONFIG_DIR = os.getenv('NETDATA_CONFIG_DIR', "/etc/netdata/")
  23 # directories should end with '/'
  24 if CONFIG_DIR[-1] != "/":
  25     CONFIG_DIR += "/"
  26 sys.path.append(MODULES_DIR + "python_modules")
  27
  28 PROGRAM = os.path.basename(__file__).replace(".plugin", "")
  29 DEBUG_FLAG = False
  30 OVERRIDE_UPDATE_EVERY = False
  31
  32 # -----------------------------------------------------------------------------
  33 # custom, third party and version specific python modules management
  34 import msg
  35
  36 try:
  37     assert sys.version_info >= (3, 1)
  38     import importlib.machinery
  39
  40     # change this hack below if we want PY_VERSION to be used in modules
  41     # import builtins
  42     # builtins.PY_VERSION = 3
  43     PY_VERSION = 3
  44     msg.info('Using python v3')
  45 except (AssertionError, ImportError):
  46     try:
  47         import imp
  48
  49         # change this hack below if we want PY_VERSION to be used in modules
  50         # import __builtin__
  51         # __builtin__.PY_VERSION = 2
  52         PY_VERSION = 2
  53         msg.info('Using python v2')
  54     except ImportError:
  55         msg.fatal('Cannot start. No importlib.machinery on python3 or lack of imp on python2')
  56 try:
  57     import yaml
  58 except ImportError:
  59     msg.fatal('Cannot find yaml library')
  60
  61
  62 class PythonCharts(object):
  63     """
  64     Main class used to control every python module.
  65     """
  66
  67     def __init__(self,
  68                  modules=None,
  69                  modules_path='../python.d/',
  70                  modules_configs='../conf.d/',
  71                  modules_disabled=None):
  72         """
  73         :param modules: list
  74         :param modules_path: str
  75         :param modules_configs: str
  76         :param modules_disabled: list
  77         """
  78
  79         if modules is None:
  80             modules = []
  81         if modules_disabled is None:
  82             modules_disabled = []
  83
  84         self.first_run = True
  85         # set configuration directory
  86         self.configs = modules_configs
  87
  88         # load modules
  89         loaded_modules = self._load_modules(modules_path, modules, modules_disabled)
  90
  91         # load configuration files
  92         configured_modules = self._load_configs(loaded_modules)
  93
  94         # good economy and prosperity:
  95         self.jobs = self._create_jobs(configured_modules)  # type: list
  96
  97         # enable timetable override like `python.d.plugin mysql debug 1`
  98         if DEBUG_FLAG and OVERRIDE_UPDATE_EVERY:
  99             for job in self.jobs:
 100                 job.create_timetable(BASE_CONFIG['update_every'])
 101
 102     @staticmethod
 103     def _import_module(path, name=None):
 104         """
 105         Try to import module using only its path.
 106         :param path: str
 107         :param name: str
 108         :return: object
 109         """
 110
 111         if name is None:
 112             name = path.split('/')[-1]
 113             if name[-len(MODULE_EXTENSION):] != MODULE_EXTENSION:
 114                 return None
 115             name = name[:-len(MODULE_EXTENSION)]
 116         try:
 117             if PY_VERSION == 3:
 118                 return importlib.machinery.SourceFileLoader(name, path).load_module()
 119             else:
 120                 return imp.load_source(name, path)
 121         except Exception as e:
 122             msg.error("Problem loading", name, str(e))
 123             return None
 124
 125     def _load_modules(self, path, modules, disabled):
 126         """
 127         Load modules from 'modules' list or dynamically every file from 'path' (only .chart.py files)
 128         :param path: str
 129         :param modules: list
 130         :param disabled: list
 131         :return: list
 132         """
 133
 134         # check if plugin directory exists
 135         if not os.path.isdir(path):
 136             msg.fatal("cannot find charts directory ", path)
 137
 138         # load modules
 139         loaded = []
 140         if len(modules) > 0:
 141             for m in modules:
 142                 if m in disabled:
 143                     continue
 144                 mod = self._import_module(path + m + MODULE_EXTENSION)
 145                 if mod is not None:
 146                     loaded.append(mod)
 147                 else:  # exit if plugin is not found
 148                     msg.fatal('no modules found.')
 149         else:
 150             # scan directory specified in path and load all modules from there
 151             names = os.listdir(path)
 152             for mod in names:
 153                 if mod.replace(MODULE_EXTENSION, "") in disabled:
 154                     msg.error(mod + ": disabled module ", mod.replace(MODULE_EXTENSION, ""))
 155                     continue
 156                 m = self._import_module(path + mod)
 157                 if m is not None:
 158                     msg.debug(mod + ": loading module '" + path + mod + "'")
 159                     loaded.append(m)
 160         return loaded
 161
 162     def _load_configs(self, modules):
 163         """
 164         Append configuration in list named `config` to every module.
 165         For multi-job modules `config` list is created in _parse_config,
 166         otherwise it is created here based on BASE_CONFIG prototype with None as identifier.
 167         :param modules: list
 168         :return: list
 169         """
 170         for mod in modules:
 171             configfile = self.configs + mod.__name__ + ".conf"
 172             if os.path.isfile(configfile):
 173                 msg.debug(mod.__name__ + ": loading module configuration: '" + configfile + "'")
 174                 try:
 175                     if not hasattr(mod, 'config'):
 176                         mod.config = {}
 177                     setattr(mod,
 178                             'config',
 179                             self._parse_config(mod, read_config(configfile)))
 180                 except Exception as e:
 181                     msg.error(mod.__name__ + ": cannot parse configuration file '" + configfile + "':", str(e))
 182             else:
 183                 msg.error(mod.__name__ + ": configuration file '" + configfile + "' not found. Using defaults.")
 184                 # set config if not found
 185                 if not hasattr(mod, 'config'):
 186                     msg.debug(mod.__name__ + ": setting configuration for only one job")
 187                     mod.config = {None: {}}
 188                     for var in BASE_CONFIG:
 189                         try:
 190                             mod.config[None][var] = getattr(mod, var)
 191                         except AttributeError:
 192                             mod.config[None][var] = BASE_CONFIG[var]
 193         return modules
 194
 195     @staticmethod
 196     def _parse_config(module, config):
 197         """
 198         Parse configuration file or extract configuration from module file.
 199         Example of returned dictionary:
 200             config = {'name': {
 201                             'update_every': 2,
 202                             'retries': 3,
 203                             'priority': 30000
 204                             'other_val': 123}}
 205         :param module: object
 206         :param config: dict
 207         :return: dict
 208         """
 209         # get default values
 210         defaults = {}
 211         msg.debug(module.__name__ + ": reading configuration")
 212         for key in BASE_CONFIG:
 213             try:
 214                 # get defaults from module config
 215                 defaults[key] = int(config.pop(key))
 216             except (KeyError, ValueError):
 217                 try:
 218                     # get defaults from module source code
 219                     defaults[key] = getattr(module, key)
 220                 except (KeyError, ValueError, AttributeError):
 221                     # if above failed, get defaults from global dict
 222                     defaults[key] = BASE_CONFIG[key]
 223
 224         # check if there are dict in config dict
 225         many_jobs = False
 226         for name in config:
 227             if type(config[name]) is dict:
 228                 many_jobs = True
 229                 break
 230
 231         # assign variables needed by supervisor to every job configuration
 232         if many_jobs:
 233             for name in config:
 234                 for key in defaults:
 235                     if key not in config[name]:
 236                         config[name][key] = defaults[key]
 237         # if only one job is needed, values doesn't have to be in dict (in YAML)
 238         else:
 239             config = {None: config.copy()}
 240             config[None].update(defaults)
 241
 242         # return dictionary of jobs where every job has BASE_CONFIG variables
 243         return config
 244
 245     @staticmethod
 246     def _create_jobs(modules):
 247         """
 248         Create jobs based on module.config dictionary and module.Service class definition.
 249         :param modules: list
 250         :return: list
 251         """
 252         jobs = []
 253         for module in modules:
 254             for name in module.config:
 255                 # register a new job
 256                 conf = module.config[name]
 257                 try:
 258                     job = module.Service(configuration=conf, name=name)
 259                 except Exception as e:
 260                     msg.error(module.__name__ +
 261                               ("/" + str(name) if name is not None else "") +
 262                               ": cannot start job: '" +
 263                               str(e))
 264                     return None
 265                 else:
 266                     # set chart_name (needed to plot run time graphs)
 267                     job.chart_name = module.__name__
 268                     if name is not None:
 269                         job.chart_name += "_" + name
 270                 jobs.append(job)
 271                 msg.debug(module.__name__ + ("/" + str(name) if name is not None else "") + ": job added")
 272
 273         return [j for j in jobs if j is not None]
 274
 275     def _stop(self, job, reason=None):
 276         """
 277         Stop specified job and remove it from self.jobs list
 278         Also notifies user about job failure if DEBUG_FLAG is set
 279         :param job: object
 280         :param reason: str
 281         """
 282         prefix = job.__module__
 283         if job.name is not None:
 284             prefix += "/" + job.name
 285         prefix += ": "
 286
 287         self.jobs.remove(job)
 288         if reason is None:
 289             return
 290         elif reason[:3] == "no ":
 291             msg.error(prefix +
 292                       "does not seem to have " +
 293                       reason[3:] +
 294                       "() function. Disabling it.")
 295         elif reason[:7] == "failed ":
 296             msg.error(prefix +
 297                       reason[7:] +
 298                       "() function reports failure.")
 299         elif reason[:13] == "configuration":
 300             msg.error(prefix +
 301                       "configuration file '" +
 302                       self.configs +
 303                       job.__module__ +
 304                       ".conf' not found. Using defaults.")
 305         elif reason[:11] == "misbehaving":
 306             msg.error(prefix + "is " + reason)
 307
 308     def check(self):
 309         """
 310         Tries to execute check() on every job.
 311         This cannot fail thus it is catching every exception
 312         If job.check() fails job is stopped
 313         """
 314         i = 0
 315         while i < len(self.jobs):
 316             job = self.jobs[i]
 317             try:
 318                 if not job.check():
 319                     self._stop(job, "failed check")
 320                 else:
 321                     msg.debug(job.chart_name, ": check succeeded")
 322                     i += 1
 323             except AttributeError:
 324                 self._stop(job, "no check")
 325             except (UnboundLocalError, Exception) as e:
 326                 self._stop(job, "misbehaving. Reason:" + str(e))
 327
 328     def create(self):
 329         """
 330         Tries to execute create() on every job.
 331         This cannot fail thus it is catching every exception.
 332         If job.create() fails job is stopped.
 333         This is also creating job run time chart.
 334         """
 335         i = 0
 336         while i < len(self.jobs):
 337             job = self.jobs[i]
 338             try:
 339                 if not job.create():
 340                     self._stop(job, "failed create")
 341                 else:
 342                     chart = job.chart_name
 343                     sys.stdout.write(
 344                         "CHART netdata.plugin_pythond_" +
 345                         chart +
 346                         " '' 'Execution time for " +
 347                         chart +
 348                         " plugin' 'milliseconds / run' python.d netdata.plugin_python area 145000 " +
 349                         str(job.timetable['freq']) +
 350                         '\n')
 351                     sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")
 352                     msg.debug("created charts for", job.chart_name)
 353                     # sys.stdout.flush()
 354                     i += 1
 355             except AttributeError:
 356                 self._stop(job, "no create")
 357             except (UnboundLocalError, Exception) as e:
 358                 self._stop(job, "misbehaving. Reason: " + str(e))
 359
 360     def update(self):
 361         """
 362         Creates and supervises every job thread.
 363         This will stay forever and ever and ever forever and ever it'll be the one...
 364         """
 365         for job in self.jobs:
 366             job.start()
 367
 368         while True:
 369             if threading.active_count() <= 1:
 370                 msg.fatal("no more jobs")
 371             time.sleep(1)
 372
 373
 374 def read_config(path):
 375     """
 376     Read YAML configuration from specified file
 377     :param path: str
 378     :return: dict
 379     """
 380     try:
 381         with open(path, 'r') as stream:
 382             config = yaml.load(stream)
 383     except (OSError, IOError):
 384         msg.error(str(path), "is not a valid configuration file")
 385         return None
 386     except yaml.YAMLError as e:
 387         msg.error(str(path), "is malformed:", e)
 388         return None
 389     return config
 390
 391
 392 def parse_cmdline(directory, *commands):
 393     """
 394     Parse parameters from command line.
 395     :param directory: str
 396     :param commands: list of str
 397     :return: dict
 398     """
 399     global DEBUG_FLAG
 400     global OVERRIDE_UPDATE_EVERY
 401     global BASE_CONFIG
 402
 403     changed_update = False
 404     mods = []
 405     for cmd in commands[1:]:
 406         if cmd == "check":
 407             pass
 408         elif cmd == "debug" or cmd == "all":
 409             DEBUG_FLAG = True
 410             # redirect stderr to stdout?
 411         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 412             DEBUG_FLAG = True
 413             mods.append(cmd.replace(".chart.py", ""))
 414         else:
 415             try:
 416                 BASE_CONFIG['update_every'] = int(cmd)
 417                 changed_update = True
 418             except ValueError:
 419                 pass
 420     if changed_update and DEBUG_FLAG:
 421         OVERRIDE_UPDATE_EVERY = True
 422         msg.debug(PROGRAM, "overriding update interval to", str(BASE_CONFIG['update_every']))
 423
 424     msg.debug("started from", commands[0], "with options:", *commands[1:])
 425
 426     return mods
 427
 428
 429 # if __name__ == '__main__':
 430 def run():
 431     """
 432     Main program.
 433     """
 434     global DEBUG_FLAG, BASE_CONFIG
 435
 436     # read configuration file
 437     disabled = []
 438     configfile = CONFIG_DIR + "python.d.conf"
 439     msg.PROGRAM = PROGRAM
 440     msg.info("reading configuration file:", configfile)
 441
 442     conf = read_config(configfile)
 443     if conf is not None:
 444         try:
 445             # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'
 446             if conf['enabled'] is False:
 447                 msg.fatal('disabled in configuration file.\n')
 448         except (KeyError, TypeError):
 449             pass
 450         try:
 451             for param in BASE_CONFIG:
 452                 BASE_CONFIG[param] = conf[param]
 453         except (KeyError, TypeError):
 454             pass  # use default update_every from NETDATA_UPDATE_EVERY
 455         try:
 456             DEBUG_FLAG = conf['debug']
 457         except (KeyError, TypeError):
 458             pass
 459         for k, v in conf.items():
 460             if k in ("update_every", "debug", "enabled"):
 461                 continue
 462             if v is False:
 463                 disabled.append(k)
 464
 465     # parse passed command line arguments
 466     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 467     msg.DEBUG_FLAG = DEBUG_FLAG
 468     msg.info("MODULES_DIR='" + MODULES_DIR +
 469              "', CONFIG_DIR='" + CONFIG_DIR +
 470              "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +
 471              ", ONLY_MODULES=" + str(modules))
 472
 473     # run plugins
 474     charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 475     charts.check()
 476     charts.create()
 477     charts.update()
 478     msg.fatal("finished")
 479
 480
 481 if __name__ == '__main__':
 482     run()