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 MODULE_EXTENSION = ".chart.py"
  13
  14 # -----------------------------------------------------------------------------
  15 # logging functions
  16
  17 PROGRAM = os.path.basename(__file__).replace(".plugin", "")
  18 DEBUG_FLAG = False
  19
  20
  21 def debug(*args):
  22     """
  23     Print message on stderr.
  24     """
  25     if not DEBUG_FLAG:
  26         return
  27     sys.stderr.write(PROGRAM + " DEBUG :")
  28     for i in args:
  29         sys.stderr.write(" " + str(i))
  30     sys.stderr.write("\n")
  31     sys.stderr.flush()
  32
  33
  34 def error(*args):
  35     """
  36     Print message on stderr.
  37     """
  38     sys.stderr.write(PROGRAM + " ERROR :")
  39     for i in args:
  40         sys.stderr.write(" " + str(i))
  41     sys.stderr.write("\n")
  42     sys.stderr.flush()
  43
  44
  45 def info(*args):
  46     """
  47     Print message on stderr.
  48     """
  49     sys.stderr.write(PROGRAM + " INFO :")
  50     for i in args:
  51         sys.stderr.write(" " + str(i))
  52     sys.stderr.write("\n")
  53     sys.stderr.flush()
  54
  55
  56 def fatal(*args):
  57     """
  58     Print message on stderr and exit.
  59     """
  60     sys.stderr.write(PROGRAM + " FATAL :")
  61     for i in args:
  62         sys.stderr.write(" " + str(i))
  63     sys.stderr.write("\n")
  64     sys.stderr.flush()
  65     sys.stdout.write('DISABLE\n')
  66     sys.exit(1)
  67
  68
  69 # -----------------------------------------------------------------------------
  70 # globals & python modules management
  71
  72 # setup environment
  73 # https://github.com/firehol/netdata/wiki/External-Plugins#environment-variables
  74 MODULES_DIR = os.path.abspath(os.getenv('NETDATA_PLUGINS_DIR',
  75                               os.path.dirname(__file__)) + "/../python.d") + "/"
  76 CONFIG_DIR = os.getenv('NETDATA_CONFIG_DIR', "/etc/netdata/")
  77 UPDATE_EVERY = os.getenv('NETDATA_UPDATE_EVERY', 1)
  78 # directories should end with '/'
  79 if CONFIG_DIR[-1] != "/":
  80     CONFIG_DIR += "/"
  81 sys.path.append(MODULES_DIR + "python_modules")
  82
  83 BASE_CONFIG = {'update_every': UPDATE_EVERY,
  84                'priority': 90000,
  85                'retries': 10}
  86
  87 try:
  88     assert sys.version_info >= (3, 1)
  89     import importlib.machinery
  90
  91     # change this hack below if we want PY_VERSION to be used in modules
  92     # import builtins
  93     # builtins.PY_VERSION = 3
  94     PY_VERSION = 3
  95     info('Using python v3')
  96 except (AssertionError, ImportError):
  97     try:
  98         import imp
  99
 100         # change this hack below if we want PY_VERSION to be used in modules
 101         # import __builtin__
 102         # __builtin__.PY_VERSION = 2
 103         PY_VERSION = 2
 104         info('Using python v2')
 105     except ImportError:
 106         fatal('Cannot start. No importlib.machinery on python3 or lack of imp on python2')
 107 try:
 108     import yaml
 109 except ImportError:
 110     fatal('Cannot find yaml library')
 111
 112
 113 class PythonCharts(object):
 114     """
 115     Main class used to control every python module.
 116     """
 117     def __init__(self,
 118                  update_every=None,  # TODO remove this
 119                  modules=None,
 120                  modules_path='../python.d/',
 121                  modules_configs='../conf.d/',
 122                  modules_disabled=None):
 123         """
 124         :param update_every: int
 125         :param modules: list
 126         :param modules_path: str
 127         :param modules_configs: str
 128         :param modules_disabled: list
 129         """
 130
 131         if modules is None:
 132             modules = []
 133         if modules_disabled is None:
 134             modules_disabled = []
 135
 136         self.first_run = True
 137         # set configuration directory
 138         self.configs = modules_configs
 139
 140         # load modules
 141         loaded_modules = self._load_modules(modules_path, modules, modules_disabled)
 142
 143         # load configuration files
 144         configured_modules = self._load_configs(loaded_modules)
 145
 146         # good economy and prosperity:
 147         self.jobs = self._create_jobs(configured_modules)  # type: list
 148 #        if DEBUG_FLAG and update_every is not None:
 149 #            for job in self.jobs:
 150 #                job.create_timetable(update_every)
 151
 152     @staticmethod
 153     def _import_module(path, name=None):
 154         """
 155         Try to import module using only its path.
 156         :param path: str
 157         :param name: str
 158         :return: object
 159         """
 160
 161         if name is None:
 162             name = path.split('/')[-1]
 163             if name[-len(MODULE_EXTENSION):] != MODULE_EXTENSION:
 164                 return None
 165             name = name[:-len(MODULE_EXTENSION)]
 166         try:
 167             if PY_VERSION == 3:
 168                 return importlib.machinery.SourceFileLoader(name, path).load_module()
 169             else:
 170                 return imp.load_source(name, path)
 171         except Exception as e:
 172             debug(str(e))
 173             return None
 174
 175     def _load_modules(self, path, modules, disabled):
 176         """
 177         Load modules from 'modules' list or dynamically every file from 'path' (only .chart.py files)
 178         :param path: str
 179         :param modules: list
 180         :param disabled: list
 181         :return: list
 182         """
 183
 184         # check if plugin directory exists
 185         if not os.path.isdir(path):
 186             fatal("cannot find charts directory ", path)
 187
 188         # load modules
 189         loaded = []
 190         if len(modules) > 0:
 191             for m in modules:
 192                 if m in disabled:
 193                     continue
 194                 mod = self._import_module(path + m + MODULE_EXTENSION)
 195                 if mod is not None:
 196                     loaded.append(mod)
 197                 else:  # exit if plugin is not found
 198                     fatal('no modules found.')
 199         else:
 200             # scan directory specified in path and load all modules from there
 201             names = os.listdir(path)
 202             for mod in names:
 203                 if mod.strip(MODULE_EXTENSION) in disabled:
 204                     error(mod + ": disabled module ", mod.strip(MODULE_EXTENSION))
 205                     continue
 206                 m = self._import_module(path + mod)
 207                 if m is not None:
 208                     debug(mod + ": loading module '" + path + mod + "'")
 209                     loaded.append(m)
 210         return loaded
 211
 212     def _load_configs(self, modules):
 213         """
 214         Append configuration in list named `config` to every module.
 215         For multi-job modules `config` list is created in _parse_config,
 216         otherwise it is created here based on BASE_CONFIG prototype with None as identifier.
 217         :param modules: list
 218         :return: list
 219         """
 220         for mod in modules:
 221             configfile = self.configs + mod.__name__ + ".conf"
 222             if os.path.isfile(configfile):
 223                 debug(mod.__name__ + ": loading module configuration: '" + configfile + "'")
 224                 try:
 225                     setattr(mod,
 226                             'config',
 227                             self._parse_config(mod, read_config(configfile)))
 228                 except Exception as e:
 229                     error(mod.__name__ + ": cannot parse configuration file '" + configfile + "':", str(e))
 230             else:
 231                 error(mod.__name__ + ": configuration file '" + configfile + "' not found. Using defaults.")
 232                 # set config if not found
 233                 if not hasattr(mod, 'config'):
 234                     mod.config = {None: {}}
 235                     for var in BASE_CONFIG:
 236                         try:
 237                             mod.config[None][var] = getattr(mod, var)
 238                         except AttributeError:
 239                             mod.config[None][var] = BASE_CONFIG[var]
 240         return modules
 241
 242     @staticmethod
 243     def _parse_config(module, config):
 244         """
 245         Parse configuration file or extract configuration from module file.
 246         Example of returned dictionary:
 247             config = {'name': {
 248                             'update_every': 2,
 249                             'retries': 3,
 250                             'priority': 30000
 251                             'other_val': 123}}
 252         :param module: object
 253         :param config: dict
 254         :return: dict
 255         """
 256         # get default values
 257         defaults = {}
 258         for key in BASE_CONFIG:
 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                     chart = job.chart_name
 388                     sys.stdout.write(
 389                         "CHART netdata.plugin_pythond_" +
 390                         chart +
 391                         " '' 'Execution time for " +
 392                         chart +
 393                         " plugin' 'milliseconds / run' python.d netdata.plugin_python area 145000 " +
 394                         str(job.timetable['freq']) +
 395                         '\n')
 396                     sys.stdout.write("DIMENSION run_time 'run time' absolute 1 1\n\n")
 397                     # sys.stdout.flush()
 398                     i += 1
 399             except AttributeError:
 400                 self._stop(job, "no create")
 401             except (UnboundLocalError, Exception) as e:
 402                 self._stop(job, "misbehaving. Reason: " + str(e))
 403
 404     def _update_job(self, job):
 405         """
 406         Tries to execute update() on specified job.
 407         This cannot fail thus it is catching every exception.
 408         If job.update() returns False, number of retries_left is decremented.
 409         If there are no more retries, job is stopped.
 410         Job is also stopped if it throws an exception.
 411         This is also updating job run time chart.
 412         Return False if job is stopped
 413         :param job: object
 414         :return: boolean
 415         """
 416         t_start = time.time()
 417         # check if it is time to execute job update() function
 418         if job.timetable['next'] > t_start:
 419             return True
 420         try:
 421             if self.first_run:
 422                 since_last = 0
 423             else:
 424                 since_last = int((t_start - job.timetable['last']) * 1000000)
 425             if not job.update(since_last):
 426                 if job.retries_left <= 0:
 427                     self._stop(job, "update failed")
 428                     return False
 429                 job.retries_left -= 1
 430                 job.timetable['next'] += job.timetable['freq']
 431                 return True
 432         except AttributeError:
 433             self._stop(job, "no update")
 434             return False
 435         except (UnboundLocalError, Exception) as e:
 436             self._stop(job, "misbehaving. Reason: " + str(e))
 437             return False
 438         t_end = time.time()
 439         job.timetable['next'] = t_end - (t_end % job.timetable['freq']) + job.timetable['freq']
 440         # draw performance graph
 441         sys.stdout.write("BEGIN netdata.plugin_pythond_" + job.chart_name + " " + str(since_last) + '\n')
 442         sys.stdout.write("SET run_time = " + str(int((t_end - t_start) * 1000)) + '\n')
 443         sys.stdout.write("END\n")
 444         # sys.stdout.flush()
 445         job.timetable['last'] = t_start
 446         job.retries_left = job.retries
 447         self.first_run = False
 448         return True
 449
 450     def update(self):
 451         """
 452         Tries to execute update() on every job by using _update_job()
 453         This will stay forever and ever and ever forever and ever it'll be the one...
 454         """
 455         self.first_run = True
 456         while True:
 457             next_runs = []
 458             i = 0
 459             while i < len(self.jobs):
 460                 job = self.jobs[i]
 461                 if self._update_job(job):
 462                     try:
 463                         next_runs.append(job.timetable['next'])
 464                     except KeyError:
 465                         pass
 466                     i += 1
 467             if len(next_runs) == 0:
 468                 fatal('no python.d modules loaded.')
 469             time.sleep(min(next_runs) - time.time())
 470
 471
 472 def read_config(path):
 473     """
 474     Read YAML configuration from specified file
 475     :param path: str
 476     :return: dict
 477     """
 478     try:
 479         with open(path, 'r') as stream:
 480             config = yaml.load(stream)
 481     except (OSError, IOError):
 482         error(str(path), "is not a valid configuration file")
 483         return None
 484     except yaml.YAMLError as e:
 485         error(str(path), "is malformed:", e)
 486         return None
 487     return config
 488
 489
 490 def parse_cmdline(directory, *commands):
 491     """
 492     Parse parameters from command line.
 493     :param directory: str
 494     :param commands: list of str
 495     :return: dict
 496     """
 497     global DEBUG_FLAG
 498     global UPDATE_EVERY
 499     update_every = UPDATE_EVERY
 500
 501     mods = []
 502     for cmd in commands[1:]:
 503         if cmd == "check":
 504             pass
 505         elif cmd == "debug" or cmd == "all":
 506             DEBUG_FLAG = True
 507             # redirect stderr to stdout?
 508         elif os.path.isfile(directory + cmd + ".chart.py") or os.path.isfile(directory + cmd):
 509             DEBUG_FLAG = True
 510             mods.append(cmd.replace(".chart.py", ""))
 511         else:
 512             try:
 513                 update_every = int(cmd)
 514             except ValueError:
 515                 update_every = UPDATE_EVERY
 516
 517     debug("started from", commands[0], "with options:", *commands[1:])
 518
 519     UPDATE_EVERY = update_every
 520     return {'modules': mods}
 521
 522
 523 # if __name__ == '__main__':
 524 def run():
 525     """
 526     Main program.
 527     """
 528     global PROGRAM, DEBUG_FLAG
 529     PROGRAM = sys.argv[0].split('/')[-1].split('.plugin')[0]
 530
 531     # read configuration file
 532     disabled = []
 533     configfile = CONFIG_DIR + "python.d.conf"
 534
 535     update_every = UPDATE_EVERY
 536     conf = read_config(configfile)
 537     if conf is not None:
 538         try:
 539             # exit the whole plugin when 'enable: no' is set in 'python.d.conf'
 540             if str(conf['enable']) is False:
 541                 fatal('disabled in configuration file.\n')
 542         except (KeyError, TypeError):
 543             pass
 544         try:
 545             update_every = conf['update_every']
 546         except (KeyError, TypeError):
 547             pass  # use default update_every from NETDATA_UPDATE_EVERY
 548         try:
 549             DEBUG_FLAG = conf['debug']
 550         except (KeyError, TypeError):
 551             pass
 552         for k, v in conf.items():
 553             if k in ("update_every", "debug", "enable"):
 554                 continue
 555             if v is False:
 556                 disabled.append(k)
 557
 558     # parse passed command line arguments
 559     modules = parse_cmdline(MODULES_DIR, *sys.argv)
 560     info("MODULES_DIR='" + MODULES_DIR +
 561          "', CONFIG_DIR='" + CONFIG_DIR +
 562          "', UPDATE_EVERY=" + str(UPDATE_EVERY) +
 563          ", ONLY_MODULES=" + str(modules))
 564
 565     # run plugins
 566     charts = PythonCharts(update_every, modules, MODULES_DIR, CONFIG_DIR + "python.d/", disabled)
 567     charts.check()
 568     charts.create()
 569     charts.update()
 570     fatal("finished")
 571
 572
 573 if __name__ == '__main__':
 574     run()