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(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                     i += 1
 368             except AttributeError:
 369                 self._stop(job, "no check")
 370             except (UnboundLocalError, Exception) as e:
 371                 self._stop(job, "misbehaving. Reason: " + str(e))
 372
 373     def create(self):
 374         """
 375         Tries to execute create() on every job.
 376         This cannot fail thus it is catching every exception.
 377         If job.create() fails job is stopped.
 378         This is also creating job run time chart.
 379         """
 380         i = 0
 381         while i < len(self.jobs):
 382             job = self.jobs[i]
 383             try:
 384                 if not job.create():
 385                     self._stop(job, "failed create")
 386                 else:
 387                     debug("created charts for", job.chart_name)
 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                     # sys.stdout.flush()
 399                     i += 1
 400             except AttributeError:
 401                 self._stop(job, "no create")
 402             except (UnboundLocalError, Exception) as e:
 403                 self._stop(job, "misbehaving. Reason: " + str(e))
 404
 405     def _update_job(self, job):
 406         """
 407         Tries to execute update() on specified job.
 408         This cannot fail thus it is catching every exception.
 409         If job.update() returns False, number of retries_left is decremented.
 410         If there are no more retries, job is stopped.
 411         Job is also stopped if it throws an exception.
 412         This is also updating job run time chart.
 413         Return False if job is stopped
 414         :param job: object
 415         :return: boolean
 416         """
 417         t_start = time.time()
 418         # check if it is time to execute job update() function
 419         if job.timetable['next'] > t_start:
 420             return True
 421         try:
 422             if self.first_run:
 423                 since_last = 0
 424             else:
 425                 since_last = int((t_start - job.timetable['last']) * 1000000)
 426             if not job.update(since_last):
 427                 if job.retries_left <= 0:
 428                     self._stop(job, "update failed")
 429                     return False
 430                 job.retries_left -= 1
 431                 job.timetable['next'] += job.timetable['freq']
 432                 return True
 433         except AttributeError:
 434             self._stop(job, "no update")
 435             return False
 436         except (UnboundLocalError, Exception) as e:
 437             self._stop(job, "misbehaving. Reason: " + str(e))
 438             return False
 439         t_end = time.time()
 440         job.timetable['next'] = t_end - (t_end % job.timetable['freq']) + job.timetable['freq']
 441         # draw performance graph
 442         run_time = str(int((t_end - t_start) * 1000))
 443         debug(job.chart_name, "updated in", run_time)
 444         sys.stdout.write("BEGIN netdata.plugin_pythond_" + job.chart_name + " " + str(since_last) + '\n')
 445         sys.stdout.write("SET run_time = " + run_time + '\n')
 446         sys.stdout.write("END\n")
 447         # sys.stdout.flush()
 448         job.timetable['last'] = t_start
 449         job.retries_left = job.retries
 450         self.first_run = False
 451         return True
 452
 453     def update(self):
 454         """
 455         Tries to execute update() on every job by using _update_job()
 456         This will stay forever and ever and ever forever and ever it'll be the one...
 457         """
 458         self.first_run = True
 459         while True:
 460             next_runs = []
 461             i = 0
 462             while i < len(self.jobs):
 463                 job = self.jobs[i]
 464                 if self._update_job(job):
 465                     try:
 466                         next_runs.append(job.timetable['next'])
 467                     except KeyError:
 468                         pass
 469                     i += 1
 470             if len(next_runs) == 0:
 471                 fatal('no python.d modules loaded.')
 472             time.sleep(min(next_runs) - time.time())
 473
 474
 475 def read_config(path):
 476     """
 477     Read YAML configuration from specified file
 478     :param path: str
 479     :return: dict
 480     """
 481     try:
 482         with open(path, 'r') as stream:
 483             config = yaml.load(stream)
 484     except (OSError, IOError):
 485         error(str(path), "is not a valid configuration file")
 486         return None
 487     except yaml.YAMLError as e:
 488         error(str(path), "is malformed:", e)
 489         return None
 490     return config
 491
 492
 493 def parse_cmdline(directory, *commands):
 494     """
 495     Parse parameters from command line.
 496     :param directory: str
 497     :param commands: list of str
 498     :return: dict
 499     """
 500     global DEBUG_FLAG
 501     global OVERRIDE_UPDATE_EVERY
 502     global BASE_CONFIG
 503
 504     mods = []
 505     for cmd in commands[1:]:
 506         if cmd == "check":
 507             pass
 508         elif cmd == "debug" or cmd == "all":
 509             DEBUG_FLAG = True
 510             # redirect stderr to stdout?
 511         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 512             DEBUG_FLAG = True
 513             mods.append(cmd.replace(".chart.py", ""))
 514         else:
 515             try:
 516                 BASE_CONFIG['update_every'] = int(cmd)
 517                 OVERRIDE_UPDATE_EVERY = True
 518                 debug(PROGRAM, "overriding update interval to", str(int(cmd)))
 519             except ValueError:
 520                 pass
 521
 522     debug("started from", commands[0], "with options:", *commands[1:])
 523
 524     return mods
 525
 526
 527 # if __name__ == '__main__':
 528 def run():
 529     """
 530     Main program.
 531     """
 532     global DEBUG_FLAG, BASE_CONFIG
 533
 534     # read configuration file
 535     disabled = []
 536     configfile = CONFIG_DIR + "python.d.conf"
 537     debug(PROGRAM, "reading configuration file:", configfile)
 538
 539     conf = read_config(configfile)
 540     if conf is not None:
 541         try:
 542             # exit the whole plugin when 'enabled: no' is set in 'python.d.conf'
 543             if conf['enabled'] is False:
 544                 fatal('disabled in configuration file.\n')
 545         except (KeyError, TypeError):
 546             pass
 547         try:
 548             for param in BASE_CONFIG:
 549                 BASE_CONFIG[param] = conf[param]
 550         except (KeyError, TypeError):
 551             pass  # use default update_every from NETDATA_UPDATE_EVERY
 552         try:
 553             DEBUG_FLAG = conf['debug']
 554         except (KeyError, TypeError):
 555             pass
 556         for k, v in conf.items():
 557             if k in ("update_every", "debug", "enabled"):
 558                 continue
 559             if v is False:
 560                 disabled.append(k)
 561
 562     # parse passed command line arguments
 563     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 564     info("MODULES_DIR='" + MODULES_DIR +
 565          "', CONFIG_DIR='" + CONFIG_DIR +
 566          "', UPDATE_EVERY=" + str(BASE_CONFIG['update_every']) +
 567          ", ONLY_MODULES=" + str(modules))
 568
 569     # run plugins
 570     modules = ['nginx']
 571     charts = PythonCharts(modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 572     charts.check()
 573     charts.create()
 574     charts.update()
 575     fatal("finished")
 576
 577
 578 if __name__ == '__main__':
 579     run()