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