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             debug("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             debug(module.__name__ + ": reading configuration")
 259             try:
 260                 # get defaults from module config
 261                 defaults[key] = int(config.pop(key))
 262             except (KeyError, ValueError):
 263                 try:
 264                     # get defaults from module source code
 265                     defaults[key] = getattr(module, key)
 266                 except (KeyError, ValueError):
 267                     # if above failed, get defaults from global dict
 268                     defaults[key] = BASE_CONFIG[key]
 269
 270         # check if there are dict in config dict
 271         many_jobs = False
 272         for name in config:
 273             if type(config[name]) is dict:
 274                 many_jobs = True
 275                 break
 276
 277         # assign variables needed by supervisor to every job configuration
 278         if many_jobs:
 279             for name in config:
 280                 for key in defaults:
 281                     if key not in config[name]:
 282                         config[name][key] = defaults[key]
 283         # if only one job is needed, values doesn't have to be in dict (in YAML)
 284         else:
 285             config = {None: config.copy()}
 286             config[None].update(defaults)
 287
 288         # return dictionary of jobs where every job has BASE_CONFIG variables
 289         return config
 290
 291     @staticmethod
 292     def _create_jobs(modules):
 293         """
 294         Create jobs based on module.config dictionary and module.Service class definition.
 295         :param modules: list
 296         :return: list
 297         """
 298         jobs = []
 299         for module in modules:
 300             for name in module.config:
 301                 # register a new job
 302                 conf = module.config[name]
 303                 try:
 304                     job = module.Service(configuration=conf, name=name)
 305                 except Exception as e:
 306                     error(module.__name__ +
 307                           ("/" + str(name) if name is not None else "") +
 308                           ": cannot start job: '" +
 309                           str(e))
 310                     return None
 311                 else:
 312                     # set chart_name (needed to plot run time graphs)
 313                     job.chart_name = module.__name__
 314                     if name is not None:
 315                         job.chart_name += "_" + name
 316                 jobs.append(job)
 317                 debug(module.__name__ + ("/" + str(name) if name is not None else "") + ": job added")
 318
 319         return [j for j in jobs if j is not None]
 320
 321     def _stop(self, job, reason=None):
 322         """
 323         Stop specified job and remove it from self.jobs list
 324         Also notifies user about job failure if DEBUG_FLAG is set
 325         :param job: object
 326         :param reason: str
 327         """
 328         prefix = job.__module__
 329         if job.name is not None:
 330             prefix += "/" + job.name
 331         prefix += ": "
 332
 333         self.jobs.remove(job)
 334         if reason is None:
 335             return
 336         elif reason[:3] == "no ":
 337             error(prefix +
 338                   "does not seem to have " +
 339                   reason[3:] +
 340                   "() function. Disabling it.")
 341         elif reason[:7] == "failed ":
 342             error(prefix +
 343                   reason[7:] +
 344                   "() function reports failure.")
 345         elif reason[:13] == "configuration":
 346             error(prefix +
 347                   "configuration file '" +
 348                   self.configs +
 349                   job.__module__ +
 350                   ".conf' not found. Using defaults.")
 351         elif reason[:11] == "misbehaving":
 352             error(prefix + "is " + reason)
 353
 354     def check(self):
 355         """
 356         Tries to execute check() on every job.
 357         This cannot fail thus it is catching every exception
 358         If job.check() fails job is stopped
 359         """
 360         i = 0
 361         while i < len(self.jobs):
 362             job = self.jobs[i]
 363             try:
 364                 if not job.check():
 365                     self._stop(job, "failed check")
 366                 else:
 367                     debug(job.name, ": check succeeded")
 368                     i += 1
 369             except AttributeError:
 370                 self._stop(job, "no check")
 371             except (UnboundLocalError, Exception) as e:
 372                 self._stop(job, "misbehaving. Reason:", str(e))
 373
 374     def create(self):
 375         """
 376         Tries to execute create() on every job.
 377         This cannot fail thus it is catching every exception.
 378         If job.create() fails job is stopped.
 379         This is also creating job run time chart.
 380         """
 381         i = 0
 382         while i < len(self.jobs):
 383             job = self.jobs[i]
 384             try:
 385                 if not job.create():
 386                     self._stop(job, "failed create")
 387                 else:
 388                     chart = job.chart_name
 389                     sys.stdout.write(
 390                         "CHART netdata.plugin_pythond_" +
 391                         chart +
 392                         " '' 'Execution time for " +
 393                         chart +
 394                         " plugin' 'milliseconds / run' python.d netdata.plugin_python area 145000 " +
 395                         str(job.timetable['freq']) +
 396                         '\n')
 397                     sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")
 398                     debug("created charts for", job.chart_name)
 399                     # sys.stdout.flush()
 400                     i += 1
 401             except AttributeError:
 402                 self._stop(job, "no create")
 403             except (UnboundLocalError, Exception) as e:
 404                 self._stop(job, "misbehaving. Reason: " + str(e))
 405
 406     def _update_job(self, job):
 407         """
 408         Tries to execute update() on specified job.
 409         This cannot fail thus it is catching every exception.
 410         If job.update() returns False, number of retries_left is decremented.
 411         If there are no more retries, job is stopped.
 412         Job is also stopped if it throws an exception.
 413         This is also updating job run time chart.
 414         Return False if job is stopped
 415         :param job: object
 416         :return: boolean
 417         """
 418         t_start = time.time()
 419         # check if it is time to execute job update() function
 420         if job.timetable['next'] > t_start:
 421             debug("it is not a time to invoke update on", job.chart_name)
 422             return True
 423         try:
 424             if self.first_run:
 425                 since_last = 0
 426             else:
 427                 since_last = int((t_start - job.timetable['last']) * 1000000)
 428             if not job.update(since_last):
 429                 if job.retries_left <= 0:
 430                     self._stop(job, "update failed")
 431                     return False
 432                 job.retries_left -= 1
 433                 job.timetable['next'] += job.timetable['freq']
 434                 return True
 435         except AttributeError:
 436             self._stop(job, "no update")
 437             return False
 438         except (UnboundLocalError, Exception) as e:
 439             self._stop(job, "misbehaving. Reason: " + str(e))
 440             return False
 441         t_end = time.time()
 442         job.timetable['next'] = t_end - (t_end % job.timetable['freq']) + job.timetable['freq']
 443         # draw performance graph
 444         run_time = str(int((t_end - t_start) * 1000))
 445         debug(job.chart_name, "updated in", run_time)
 446         sys.stdout.write("BEGIN netdata.plugin_pythond_" + job.chart_name + " " + str(since_last) + '\n')
 447         sys.stdout.write("SET run_time = " + run_time + '\n')
 448         sys.stdout.write("END\n")
 449         # sys.stdout.flush()
 450         job.timetable['last'] = t_start
 451         job.retries_left = job.retries
 452         self.first_run = False
 453         return True
 454
 455     def update(self):
 456         """
 457         Tries to execute update() on every job by using _update_job()
 458         This will stay forever and ever and ever forever and ever it'll be the one...
 459         """
 460         self.first_run = True
 461         while True:
 462             next_runs = []
 463             i = 0
 464             while i < len(self.jobs):
 465                 job = self.jobs[i]
 466                 if self._update_job(job):
 467                     try:
 468                         next_runs.append(job.timetable['next'])
 469                     except KeyError:
 470                         pass
 471                     i += 1
 472             if len(next_runs) == 0:
 473                 fatal('no python.d modules loaded.')
 474             time.sleep(min(next_runs) - time.time())
 475
 476
 477 def read_config(path):
 478     """
 479     Read YAML configuration from specified file
 480     :param path: str
 481     :return: dict
 482     """
 483     try:
 484         with open(path, 'r') as stream:
 485             config = yaml.load(stream)
 486     except (OSError, IOError):
 487         error(str(path), "is not a valid configuration file")
 488         return None
 489     except yaml.YAMLError as e:
 490         error(str(path), "is malformed:", e)
 491         return None
 492     return config
 493
 494
 495 def parse_cmdline(directory, *commands):
 496     """
 497     Parse parameters from command line.
 498     :param directory: str
 499     :param commands: list of str
 500     :return: dict
 501     """
 502     global DEBUG_FLAG
 503     global OVERRIDE_UPDATE_EVERY
 504     global BASE_CONFIG
 505
 506     mods = []
 507     for cmd in commands[1:]:
 508         if cmd == "check":
 509             pass
 510         elif cmd == "debug" or cmd == "all":
 511             DEBUG_FLAG = True
 512             # redirect stderr to stdout?
 513         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 514             DEBUG_FLAG = True
 515             mods.append(cmd.replace(".chart.py", ""))
 516         else:
 517             try:
 518                 BASE_CONFIG['update_every'] = int(cmd)
 519                 OVERRIDE_UPDATE_EVERY = True
 520                 debug(PROGRAM, "overriding update interval to", str(int(cmd)))
 521             except ValueError:
 522                 pass
 523
 524     debug("started from", commands[0], "with options:", *commands[1:])
 525
 526     return mods
 527
 528
 529 # if __name__ == '__main__':
 530 def run():
 531     """
 532     Main program.
 533     """
 534     global DEBUG_FLAG, BASE_CONFIG
 535
 536     # read configuration file
 537     disabled = []
 538     configfile = CONFIG_DIR + "python.d.conf"
 539     debug(PROGRAM, "reading configuration file:", configfile)
 540
 541     conf = read_config(configfile)
 542     if conf is not None:
 543         try:
 544             # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'
 545             if conf['enabled'] is False:
 546                 fatal('disabled in configuration file.\n')
 547         except (KeyError, TypeError):
 548             pass
 549         try:
 550             for param in BASE_CONFIG:
 551                 BASE_CONFIG[param] = conf[param]
 552         except (KeyError, TypeError):
 553             pass  # use default update_every from NETDATA_UPDATE_EVERY
 554         try:
 555             DEBUG_FLAG = conf['debug']
 556         except (KeyError, TypeError):
 557             pass
 558         for k, v in conf.items():
 559             if k in ("update_every", "debug", "enabled"):
 560                 continue
 561             if v is False:
 562                 disabled.append(k)
 563
 564     # parse passed command line arguments
 565     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 566     info("MODULES_DIR='" + MODULES_DIR +
 567          "', CONFIG_DIR='" + CONFIG_DIR +
 568          "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +
 569          ", ONLY_MODULES=" + str(modules))
 570
 571     # run plugins
 572     charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 573     charts.check()
 574     charts.create()
 575     charts.update()
 576     fatal("finished")
 577
 578
 579 if __name__ == '__main__':
 580     run()