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         overridden = []
 316         while i < len(self.jobs):
 317             job = self.jobs[i]
 318             if job.name in overridden:
 319                 self._stop(job)
 320                 msg.error(job.name + " already exists")
 321             try:
 322                 if not job.check():
 323                     self._stop(job, "failed check")
 324                 else:
 325                     msg.debug(job.chart_name, ": check succeeded")
 326                     i += 1
 327                     try:
 328                         if job.override_name is not None:
 329                             job.name = job.override_name
 330                             msg.debug(job.chart_name + " changing chart name to: " + job.__module__ + job.name)
 331                             job.chart_name = job.__module__ + job.name
 332                             overridden.append(job.name)
 333                     except Exception:
 334                         pass
 335             except AttributeError:
 336                 self._stop(job, "no check")
 337             except (UnboundLocalError, Exception) as e:
 338                 self._stop(job, "misbehaving. Reason:" + str(e))
 339
 340     def create(self):
 341         """
 342         Tries to execute create() on every job.
 343         This cannot fail thus it is catching every exception.
 344         If job.create() fails job is stopped.
 345         This is also creating job run time chart.
 346         """
 347         i = 0
 348         while i < len(self.jobs):
 349             job = self.jobs[i]
 350             try:
 351                 if not job.create():
 352                     self._stop(job, "failed create")
 353                 else:
 354                     chart = job.chart_name
 355                     sys.stdout.write(
 356                         "CHART netdata.plugin_pythond_" +
 357                         chart +
 358                         " '' 'Execution time for " +
 359                         chart +
 360                         " plugin' 'milliseconds / run' python.d netdata.plugin_python area 145000 " +
 361                         str(job.timetable['freq']) +
 362                         '\n')
 363                     sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")
 364                     msg.debug("created charts for", job.chart_name)
 365                     # sys.stdout.flush()
 366                     i += 1
 367             except AttributeError:
 368                 self._stop(job, "no create")
 369             except (UnboundLocalError, Exception) as e:
 370                 self._stop(job, "misbehaving. Reason: " + str(e))
 371
 372     def update(self):
 373         """
 374         Creates and supervises every job thread.
 375         This will stay forever and ever and ever forever and ever it'll be the one...
 376         """
 377         for job in self.jobs:
 378             job.start()
 379
 380         while True:
 381             if threading.active_count() <= 1:
 382                 msg.fatal("no more jobs")
 383             time.sleep(1)
 384
 385
 386 def read_config(path):
 387     """
 388     Read YAML configuration from specified file
 389     :param path: str
 390     :return: dict
 391     """
 392     try:
 393         with open(path, 'r') as stream:
 394             config = yaml.load(stream)
 395     except (OSError, IOError):
 396         msg.error(str(path), "is not a valid configuration file")
 397         return None
 398     except yaml.YAMLError as e:
 399         msg.error(str(path), "is malformed:", e)
 400         return None
 401     return config
 402
 403
 404 def parse_cmdline(directory, *commands):
 405     """
 406     Parse parameters from command line.
 407     :param directory: str
 408     :param commands: list of str
 409     :return: dict
 410     """
 411     global DEBUG_FLAG
 412     global OVERRIDE_UPDATE_EVERY
 413     global BASE_CONFIG
 414
 415     changed_update = False
 416     mods = []
 417     for cmd in commands[1:]:
 418         if cmd == "check":
 419             pass
 420         elif cmd == "debug" or cmd == "all":
 421             DEBUG_FLAG = True
 422             # redirect stderr to stdout?
 423         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 424             DEBUG_FLAG = True
 425             mods.append(cmd.replace(".chart.py", ""))
 426         else:
 427             try:
 428                 BASE_CONFIG['update_every'] = int(cmd)
 429                 changed_update = True
 430             except ValueError:
 431                 pass
 432     if changed_update and DEBUG_FLAG:
 433         OVERRIDE_UPDATE_EVERY = True
 434         msg.debug(PROGRAM, "overriding update interval to", str(BASE_CONFIG['update_every']))
 435
 436     msg.debug("started from", commands[0], "with options:", *commands[1:])
 437
 438     return mods
 439
 440
 441 # if __name__ == '__main__':
 442 def run():
 443     """
 444     Main program.
 445     """
 446     global DEBUG_FLAG, BASE_CONFIG
 447
 448     # read configuration file
 449     disabled = []
 450     configfile = CONFIG_DIR + "python.d.conf"
 451     msg.PROGRAM = PROGRAM
 452     msg.info("reading configuration file:", configfile)
 453
 454     conf = read_config(configfile)
 455     if conf is not None:
 456         try:
 457             # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'
 458             if conf['enabled'] is False:
 459                 msg.fatal('disabled in configuration file.\n')
 460         except (KeyError, TypeError):
 461             pass
 462         try:
 463             for param in BASE_CONFIG:
 464                 BASE_CONFIG[param] = conf[param]
 465         except (KeyError, TypeError):
 466             pass  # use default update_every from NETDATA_UPDATE_EVERY
 467         try:
 468             DEBUG_FLAG = conf['debug']
 469         except (KeyError, TypeError):
 470             pass
 471         for k, v in conf.items():
 472             if k in ("update_every", "debug", "enabled"):
 473                 continue
 474             if v is False:
 475                 disabled.append(k)
 476
 477     # parse passed command line arguments
 478     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 479     msg.DEBUG_FLAG = DEBUG_FLAG
 480     msg.info("MODULES_DIR='" + MODULES_DIR +
 481              "', CONFIG_DIR='" + CONFIG_DIR +
 482              "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +
 483              ", ONLY_MODULES=" + str(modules))
 484
 485     # run plugins
 486     charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 487     charts.check()
 488     charts.create()
 489     charts.update()
 490     msg.fatal("finished")
 491
 492
 493 if __name__ == '__main__':
 494     run()