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