plugins.d/python.d.plugin

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