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