plugins.d/python.d.plugin

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