plugins.d/python.d.plugin

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