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