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