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