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         if config is None:
 210             config = {}
 211         # get default values
 212         defaults = {}
 213         msg.debug(module.__name__ + ": reading configuration")
 214         for key in BASE_CONFIG:
 215             try:
 216                 # get defaults from module config
 217                 defaults[key] = int(config.pop(key))
 218             except (KeyError, ValueError):
 219                 try:
 220                     # get defaults from module source code
 221                     defaults[key] = getattr(module, key)
 222                 except (KeyError, ValueError, AttributeError):
 223                     # if above failed, get defaults from global dict
 224                     defaults[key] = BASE_CONFIG[key]
 225
 226         # check if there are dict in config dict
 227         many_jobs = False
 228         for name in config:
 229             if type(config[name]) is dict:
 230                 many_jobs = True
 231                 break
 232
 233         # assign variables needed by supervisor to every job configuration
 234         if many_jobs:
 235             for name in config:
 236                 for key in defaults:
 237                     if key not in config[name]:
 238                         config[name][key] = defaults[key]
 239         # if only one job is needed, values doesn't have to be in dict (in YAML)
 240         else:
 241             config = {None: config.copy()}
 242             config[None].update(defaults)
 243
 244         # return dictionary of jobs where every job has BASE_CONFIG variables
 245         return config
 246
 247     @staticmethod
 248     def _create_jobs(modules):
 249         """
 250         Create jobs based on module.config dictionary and module.Service class definition.
 251         :param modules: list
 252         :return: list
 253         """
 254         jobs = []
 255         for module in modules:
 256             for name in module.config:
 257                 # register a new job
 258                 conf = module.config[name]
 259                 try:
 260                     job = module.Service(configuration=conf, name=name)
 261                 except Exception as e:
 262                     msg.error(module.__name__ +
 263                               ("/" + str(name) if name is not None else "") +
 264                               ": cannot start job: '" +
 265                               str(e))
 266                     return None
 267                 else:
 268                     # set chart_name (needed to plot run time graphs)
 269                     job.chart_name = module.__name__
 270                     if name is not None:
 271                         job.chart_name += "_" + name
 272                 jobs.append(job)
 273                 msg.debug(module.__name__ + ("/" + str(name) if name is not None else "") + ": job added")
 274
 275         return [j for j in jobs if j is not None]
 276
 277     def _stop(self, job, reason=None):
 278         """
 279         Stop specified job and remove it from self.jobs list
 280         Also notifies user about job failure if DEBUG_FLAG is set
 281         :param job: object
 282         :param reason: str
 283         """
 284         prefix = job.__module__
 285         if job.name is not None:
 286             prefix += "/" + job.name
 287         prefix += ": "
 288
 289         self.jobs.remove(job)
 290         if reason is None:
 291             return
 292         elif reason[:3] == "no ":
 293             msg.error(prefix +
 294                       "does not seem to have " +
 295                       reason[3:] +
 296                       "() function. Disabling it.")
 297         elif reason[:7] == "failed ":
 298             msg.error(prefix +
 299                       reason[7:] +
 300                       "() function reports failure.")
 301         elif reason[:13] == "configuration":
 302             msg.error(prefix +
 303                       "configuration file '" +
 304                       self.configs +
 305                       job.__module__ +
 306                       ".conf' not found. Using defaults.")
 307         elif reason[:11] == "misbehaving":
 308             msg.error(prefix + "is " + reason)
 309
 310     def check(self):
 311         """
 312         Tries to execute check() on every job.
 313         This cannot fail thus it is catching every exception
 314         If job.check() fails job is stopped
 315         """
 316         i = 0
 317         overridden = []
 318         while i < len(self.jobs):
 319             job = self.jobs[i]
 320             if job.name in overridden:
 321                 self._stop(job)
 322                 msg.error(job.name + " already exists")
 323             try:
 324                 if not job.check():
 325                     self._stop(job, "failed check")
 326                 else:
 327                     msg.debug(job.chart_name, ": check succeeded")
 328                     i += 1
 329                     try:
 330                         if job.override_name is not None:
 331                             job.name = job.override_name
 332                             msg.debug(job.chart_name + " changing chart name to: " + job.__module__ + job.name)
 333                             job.chart_name = job.__module__ + job.name
 334                             overridden.append(job.name)
 335                     except Exception:
 336                         pass
 337             except AttributeError:
 338                 self._stop(job, "no check")
 339             except (UnboundLocalError, Exception) as e:
 340                 self._stop(job, "misbehaving. Reason:" + str(e))
 341
 342     def create(self):
 343         """
 344         Tries to execute create() on every job.
 345         This cannot fail thus it is catching every exception.
 346         If job.create() fails job is stopped.
 347         This is also creating job run time chart.
 348         """
 349         i = 0
 350         while i < len(self.jobs):
 351             job = self.jobs[i]
 352             try:
 353                 if not job.create():
 354                     self._stop(job, "failed create")
 355                 else:
 356                     chart = job.chart_name
 357                     sys.stdout.write(
 358                         "CHART netdata.plugin_pythond_" +
 359                         chart +
 360                         " '' 'Execution time for " +
 361                         chart +
 362                         " plugin' 'milliseconds / run' python.d netdata.plugin_python area 145000 " +
 363                         str(job.timetable['freq']) +
 364                         '\n')
 365                     sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")
 366                     msg.debug("created charts for", job.chart_name)
 367                     # sys.stdout.flush()
 368                     i += 1
 369             except AttributeError:
 370                 self._stop(job, "no create")
 371             except (UnboundLocalError, Exception) as e:
 372                 self._stop(job, "misbehaving. Reason: " + str(e))
 373
 374     def update(self):
 375         """
 376         Creates and supervises every job thread.
 377         This will stay forever and ever and ever forever and ever it'll be the one...
 378         """
 379         for job in self.jobs:
 380             job.start()
 381
 382         while True:
 383             if threading.active_count() <= 1:
 384                 msg.fatal("no more jobs")
 385             time.sleep(1)
 386
 387
 388 def read_config(path):
 389     """
 390     Read YAML configuration from specified file
 391     :param path: str
 392     :return: dict
 393     """
 394     try:
 395         with open(path, 'r') as stream:
 396             config = yaml.load(stream)
 397     except (OSError, IOError):
 398         msg.error(str(path), "is not a valid configuration file")
 399         return None
 400     except yaml.YAMLError as e:
 401         msg.error(str(path), "is malformed:", e)
 402         return None
 403     return config
 404
 405
 406 def parse_cmdline(directory, *commands):
 407     """
 408     Parse parameters from command line.
 409     :param directory: str
 410     :param commands: list of str
 411     :return: dict
 412     """
 413     global DEBUG_FLAG
 414     global OVERRIDE_UPDATE_EVERY
 415     global BASE_CONFIG
 416
 417     changed_update = False
 418     mods = []
 419     for cmd in commands[1:]:
 420         if cmd == "check":
 421             pass
 422         elif cmd == "debug" or cmd == "all":
 423             DEBUG_FLAG = True
 424             # redirect stderr to stdout?
 425         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 426             DEBUG_FLAG = True
 427             mods.append(cmd.replace(".chart.py", ""))
 428         else:
 429             try:
 430                 BASE_CONFIG['update_every'] = int(cmd)
 431                 changed_update = True
 432             except ValueError:
 433                 pass
 434     if changed_update and DEBUG_FLAG:
 435         OVERRIDE_UPDATE_EVERY = True
 436         msg.debug(PROGRAM, "overriding update interval to", str(BASE_CONFIG['update_every']))
 437
 438     msg.debug("started from", commands[0], "with options:", *commands[1:])
 439
 440     return mods
 441
 442
 443 # if __name__ == '__main__':
 444 def run():
 445     """
 446     Main program.
 447     """
 448     global DEBUG_FLAG, BASE_CONFIG
 449
 450     # read configuration file
 451     disabled = []
 452     configfile = CONFIG_DIR + "python.d.conf"
 453     msg.PROGRAM = PROGRAM
 454     msg.info("reading configuration file:", configfile)
 455
 456     conf = read_config(configfile)
 457     if conf is not None:
 458         try:
 459             # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'
 460             if conf['enabled'] is False:
 461                 msg.fatal('disabled in configuration file.\n')
 462         except (KeyError, TypeError):
 463             pass
 464         try:
 465             for param in BASE_CONFIG:
 466                 BASE_CONFIG[param] = conf[param]
 467         except (KeyError, TypeError):
 468             pass  # use default update_every from NETDATA_UPDATE_EVERY
 469         try:
 470             DEBUG_FLAG = conf['debug']
 471         except (KeyError, TypeError):
 472             pass
 473         for k, v in conf.items():
 474             if k in ("update_every", "debug", "enabled"):
 475                 continue
 476             if v is False:
 477                 disabled.append(k)
 478
 479     # parse passed command line arguments
 480     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 481     msg.DEBUG_FLAG = DEBUG_FLAG
 482     msg.info("MODULES_DIR='" + MODULES_DIR +
 483              "', CONFIG_DIR='" + CONFIG_DIR +
 484              "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +
 485              ", ONLY_MODULES=" + str(modules))
 486
 487     # run plugins
 488     charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 489     charts.check()
 490     charts.create()
 491     charts.update()
 492     msg.fatal("finished")
 493
 494
 495 if __name__ == '__main__':
 496     run()