plugins.d/python.d.plugin

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