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