'''':;

if [[ "$OSTYPE" == "darwin"* ]]; then

export OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES

exec "$(command -v python || command -v python3 || command -v python2 ||

echo "ERROR python IS NOT AVAILABLE IN THIS SYSTEM")" "$0" "$@" # '''

import collections

import copy

import gc

import multiprocessing

import os

import re

import sys

import time

import threading

import types

PY_VERSION = sys.version_info[:2]

if PY_VERSION > (3, 1):

from importlib.machinery import SourceFileLoader

else:

from imp import load_source as SourceFileLoader

ENV_NETDATA_USER_CONFIG_DIR = 'NETDATA_USER_CONFIG_DIR'

ENV_NETDATA_STOCK_CONFIG_DIR = 'NETDATA_STOCK_CONFIG_DIR'

ENV_NETDATA_PLUGINS_DIR = 'NETDATA_PLUGINS_DIR'

ENV_NETDATA_UPDATE_EVERY = 'NETDATA_UPDATE_EVERY'

def dirs():

user_config = os.getenv(

ENV_NETDATA_USER_CONFIG_DIR,

'@configdir_POST@',

)

stock_config = os.getenv(

ENV_NETDATA_STOCK_CONFIG_DIR,

'@libconfigdir_POST@',

)

modules_user_config = os.path.join(user_config, 'python.d')

modules_stock_config = os.path.join(stock_config, 'python.d')

modules = os.path.abspath(

os.getenv(

ENV_NETDATA_PLUGINS_DIR,

os.path.dirname(__file__),

) + '/../python.d'

)

pythond_packages = os.path.join(modules, 'python_modules')

return collections.namedtuple(

'Dirs',

[

'user_config',

'stock_config',

'modules_user_config',

'modules_stock_config',

'modules',

'pythond_packages',

]

)(

user_config,

stock_config,

modules_user_config,

modules_stock_config,

modules,

pythond_packages,

)

DIRS = dirs()

sys.path.append(DIRS.pythond_packages)

from bases.collection import safe_print

from bases.loggers import PythonDLogger

from bases.loaders import load_config

try:

from collections import OrderedDict

except ImportError:

from third_party.ordereddict import OrderedDict

END_TASK_MARKER = None

IS_ATTY = sys.stdout.isatty()

PLUGIN_CONF_FILE = 'python.d.conf'

MODULE_SUFFIX = '.chart.py'

OBSOLETED_MODULES = (

'apache_cache', # replaced by web_log

'cpuidle', # rewritten in C

'cpufreq', # rewritten in C

'gunicorn_log', # replaced by web_log

'linux_power_supply', # rewritten in C

'nginx_log', # replaced by web_log

'mdstat', # rewritten in C

'sslcheck', # memory leak bug https://github.com/netdata/netdata/issues/5624

)

AVAILABLE_MODULES = [

m[:-len(MODULE_SUFFIX)] for m in sorted(os.listdir(DIRS.modules))

if m.endswith(MODULE_SUFFIX) and m[:-len(MODULE_SUFFIX)] not in OBSOLETED_MODULES

]

PLUGIN_BASE_CONF = {

'enabled': True,

'default_run': True,

'gc_run': True,

'gc_interval': 300,

}

JOB_BASE_CONF = {

'update_every': os.getenv(ENV_NETDATA_UPDATE_EVERY, 1),

'priority': 60000,

'autodetection_retry': 0,

'chart_cleanup': 10,

'penalty': True,

'name': str(),

}

def heartbeat():

if IS_ATTY:

return

safe_print('\n')

class HeartBeat(threading.Thread):

def __init__(self, every):

threading.Thread.__init__(self)

self.daemon = True

self.every = every

def run(self):

while True:

time.sleep(self.every)

heartbeat()

def load_module(name):

abs_path = os.path.join(DIRS.modules, '{0}{1}'.format(name, MODULE_SUFFIX))

module = SourceFileLoader(name, abs_path)

if isinstance(module, types.ModuleType):

return module

return module.load_module()

def multi_path_find(name, paths):

for path in paths:

abs_name = os.path.join(path, name)

if os.path.isfile(abs_name):

return abs_name

return ''

Task = collections.namedtuple(

'Task',

[

'module_name',

'explicitly_enabled',

],

)

Result = collections.namedtuple(

'Result',

[

'module_name',

'jobs_configs',

],

)

class ModuleChecker(multiprocessing.Process):

def __init__(

self,

task_queue,

result_queue,

):

multiprocessing.Process.__init__(self)

self.log = PythonDLogger()

self.log.job_name = 'checker'

self.task_queue = task_queue

self.result_queue = result_queue

def run(self):

self.log.info('starting...')

HeartBeat(1).start()

while self.run_once():

pass

self.log.info('terminating...')

def run_once(self):

task = self.task_queue.get()

if task is END_TASK_MARKER:

# TODO: find better solution, understand why heartbeat thread doesn't work

heartbeat()

self.task_queue.task_done()

self.result_queue.put(END_TASK_MARKER)

return False

result = self.do_task(task)

if result:

self.result_queue.put(result)

self.task_queue.task_done()

return True

def do_task(self, task):

self.log.info("{0} : checking".format(task.module_name))

# LOAD SOURCE

module = Module(task.module_name)

try:

module.load_source()

except Exception as error:

self.log.warning("{0} : error on loading source : {1}, skipping module".format(

task.module_name,

error,

))

return None

else:

self.log.info("{0} : source successfully loaded".format(task.module_name))

if module.is_disabled_by_default() and not task.explicitly_enabled:

self.log.info("{0} : disabled by default".format(task.module_name))

return None

# LOAD CONFIG

paths = [

DIRS.modules_user_config,

DIRS.modules_stock_config,

]

conf_abs_path = multi_path_find(

name='{0}.conf'.format(task.module_name),

paths=paths,

)

if conf_abs_path:

self.log.info("{0} : found config file '{1}'".format(task.module_name, conf_abs_path))

try:

module.load_config(conf_abs_path)

except Exception as error:

self.log.warning("{0} : error on loading config : {1}, skipping module".format(

task.module_name, error))

return None

else:

self.log.info("{0} : config was not found in '{1}', using default 1 job config".format(

task.module_name, paths))

# CHECK JOBS

jobs = module.create_jobs()

self.log.info("{0} : created {1} job(s) from the config".format(task.module_name, len(jobs)))

successful_jobs_configs = list()

for job in jobs:

if job.autodetection_retry() > 0:

successful_jobs_configs.append(job.config)

self.log.info("{0}[{1}]: autodetection job, will be checked in main".format(task.module_name, job.name))

continue

try:

job.init()

except Exception as error:

self.log.warning("{0}[{1}] : unhandled exception on init : {2}, skipping the job)".format(

task.module_name, job.name, error))

continue

try:

ok = job.check()

except Exception as error:

self.log.warning("{0}[{1}] : unhandled exception on check : {2}, skipping the job".format(

task.module_name, job.name, error))

continue

if not ok:

self.log.info("{0}[{1}] : check failed, skipping the job".format(task.module_name, job.name))

continue

self.log.info("{0}[{1}] : check successful".format(task.module_name, job.name))

job.config['autodetection_retry'] = job.config['update_every']

successful_jobs_configs.append(job.config)

if not successful_jobs_configs:

self.log.info("{0} : all jobs failed, skipping module".format(task.module_name))

return None

return Result(module.source.__name__, successful_jobs_configs)

class JobConf(OrderedDict):

def __init__(self, *args):

OrderedDict.__init__(self, *args)

def set_defaults_from_module(self, module):

for k in [k for k in JOB_BASE_CONF if hasattr(module, k)]:

self[k] = getattr(module, k)

def set_defaults_from_config(self, module_config):

for k in [k for k in JOB_BASE_CONF if k in module_config]:

self[k] = module_config[k]

def set_job_name(self, name):

self['job_name'] = re.sub(r'\s+', '_', name)

def set_override_name(self, name):

self['override_name'] = re.sub(r'\s+', '_', name)

def as_dict(self):

return copy.deepcopy(OrderedDict(self))

class Job:

def __init__(

self,

service,

module_name,

config,

):

self.service = service

self.config = config

self.module_name = module_name

self.name = config['job_name']

self.override_name = config['override_name']

self.wrapped = None

def init(self):

self.wrapped = self.service(configuration=self.config.as_dict())

def check(self):

return self.wrapped.check()

def post_check(self, min_update_every):

if self.wrapped.update_every < min_update_every:

self.wrapped.update_every = min_update_every

def create(self):

return self.wrapped.create()

def autodetection_retry(self):

return self.config['autodetection_retry']

def run(self):

self.wrapped.run()

class Module:

def __init__(self, name):

self.name = name

self.source = None

self.config = dict()

def is_disabled_by_default(self):

return bool(getattr(self.source, 'disabled_by_default', False))

def load_source(self):

self.source = load_module(self.name)

def load_config(self, abs_path):

self.config = load_config(abs_path) or dict()

def gather_jobs_configs(self):

job_names = [v for v in self.config if isinstance(self.config[v], dict)]

if len(job_names) == 0:

job_conf = JobConf(JOB_BASE_CONF)

job_conf.set_defaults_from_module(self.source)

job_conf.update(self.config)

job_conf.set_job_name(self.name)

job_conf.set_override_name(job_conf.pop('name'))

return [job_conf]

configs = list()

for job_name in job_names:

raw_job_conf = self.config[job_name]

job_conf = JobConf(JOB_BASE_CONF)

job_conf.set_defaults_from_module(self.source)

job_conf.set_defaults_from_config(self.config)

job_conf.update(raw_job_conf)

job_conf.set_job_name(job_name)

job_conf.set_override_name(job_conf.pop('name'))

configs.append(job_conf)

return configs

def create_jobs(self, jobs_conf=None):

return [Job(self.source.Service, self.name, conf) for conf in jobs_conf or self.gather_jobs_configs()]

class JobRunner(threading.Thread):

def __init__(self, job):

threading.Thread.__init__(self)

self.daemon = True

self.wrapped = job

def run(self):

self.wrapped.run()

class PluginConf(dict):

def __init__(self, *args):

dict.__init__(self, *args)

def is_module_enabled(self, module_name, explicit):

if module_name in self:

return self[module_name]

if explicit:

return False

return self['default_run']

class Plugin:

def __init__(

self,

min_update_every=1,

modules_to_run=tuple(AVAILABLE_MODULES),

):

self.log = PythonDLogger()

self.config = PluginConf(PLUGIN_BASE_CONF)

self.task_queue = multiprocessing.JoinableQueue()

self.result_queue = multiprocessing.JoinableQueue()

self.min_update_every = min_update_every

self.modules_to_run = modules_to_run

self.auto_detection_jobs = list()

self.tasks = list()

self.results = list()

self.checked_jobs = collections.defaultdict(list)

self.runs = 0

@staticmethod

def shutdown():

safe_print('DISABLE')

exit(0)

def run(self):

jobs = self.create_jobs()

if not jobs:

return

for job in self.prepare_jobs(jobs):

self.log.info('{0}[{1}] : started in thread'.format(job.module_name, job.name))

JobRunner(job).start()

self.serve()

def enqueue_tasks(self):

for task in self.tasks:

self.task_queue.put(task)

self.task_queue.put(END_TASK_MARKER)

def dequeue_results(self):

while True:

result = self.result_queue.get()

self.result_queue.task_done()

if result is END_TASK_MARKER:

break

self.results.append(result)

def load_config(self):

paths = [

DIRS.user_config,

DIRS.stock_config,

]

self.log.info("checking for config in {0}".format(paths))

abs_path = multi_path_find(name=PLUGIN_CONF_FILE, paths=paths)

if not abs_path:

self.log.warning('config was not found, using defaults')

return True

self.log.info("config found, loading config '{0}'".format(abs_path))

try:

config = load_config(abs_path) or dict()

except Exception as error:

self.log.error('error on loading config : {0}'.format(error))

return False

self.log.info('config successfully loaded')

self.config.update(config)

return True

def setup(self):

self.log.info('starting setup')

if not self.load_config():

return False

if not self.config['enabled']:

self.log.info('disabled in configuration file')

return False

for mod in self.modules_to_run:

if self.config.is_module_enabled(mod, False):

task = Task(mod, self.config.is_module_enabled(mod, True))

self.tasks.append(task)

else:

self.log.info("{0} : disabled in configuration file".format(mod))

if not self.tasks:

self.log.info('no modules to run')

return False

worker = ModuleChecker(self.task_queue, self.result_queue)

self.log.info('starting checker process ({0} module(s) to check)'.format(len(self.tasks)))

worker.start()

# TODO: timeouts?

self.enqueue_tasks()

self.task_queue.join()

self.dequeue_results()

self.result_queue.join()

self.task_queue.close()

self.result_queue.close()

self.log.info('stopping checker process')

worker.join()

if not self.results:

self.log.info('no modules to run')

return False

self.log.info("setup complete, {0} active module(s) : '{1}'".format(

len(self.results),

[v.module_name for v in self.results])

)

return True

def create_jobs(self):

jobs = list()

for result in self.results:

module = Module(result.module_name)

try:

module.load_source()

except Exception as error:

self.log.warning("{0} : error on loading module source : {1}, skipping module".format(

result.module_name, error))

continue

module_jobs = module.create_jobs(result.jobs_configs)

self.log.info("{0} : created {1} job(s)".format(module.name, len(module_jobs)))

jobs.extend(module_jobs)

return jobs

def prepare_jobs(self, jobs):

prepared = list()

for job in jobs:

check_name = job.override_name or job.name

if check_name in self.checked_jobs[job.module_name]:

self.log.info('{0}[{1}] : already served by another job, skipping the job'.format(

job.module_name, job.name))

continue

try:

job.init()

except Exception as error:

self.log.warning("{0}[{1}] : unhandled exception on init : {2}, skipping the job".format(

job.module_name, job.name, error))

continue

self.log.info("{0}[{1}] : init successful".format(job.module_name, job.name))

try:

ok = job.check()

except Exception as error:

self.log.warning("{0}[{1}] : unhandled exception on check : {2}, skipping the job".format(

job.module_name, job.name, error))

continue

if not ok:

self.log.info('{0}[{1}] : check failed'.format(job.module_name, job.name))

if job.autodetection_retry() > 0:

self.log.info('{0}[{1}] : will recheck every {2} second(s)'.format(

job.module_name, job.name, job.autodetection_retry()))

self.auto_detection_jobs.append(job)

continue

self.log.info('{0}[{1}] : check successful'.format(job.module_name, job.name))

job.post_check(int(self.min_update_every))

if not job.create():

self.log.info('{0}[{1}] : create failed'.format(job.module_name, job.name))

self.checked_jobs[job.module_name].append(check_name)

prepared.append(job)

return prepared

def serve(self):

gc_run = self.config['gc_run']

gc_interval = self.config['gc_interval']

while True:

self.runs += 1

# threads: main + heartbeat

if threading.active_count() <= 2 and not self.auto_detection_jobs:

return

time.sleep(1)

if gc_run and self.runs % gc_interval == 0:

v = gc.collect()

self.log.debug('GC collection run result: {0}'.format(v))

self.auto_detection_jobs = [job for job in self.auto_detection_jobs if not self.retry_job(job)]

def retry_job(self, job):

stop_retrying = True

retry_later = False

if self.runs % job.autodetection_retry() != 0:

return retry_later

check_name = job.override_name or job.name

if check_name in self.checked_jobs[job.module_name]:

self.log.info("{0}[{1}]: already served by another job, give up on retrying".format(

job.module_name, job.name))

return stop_retrying

try:

ok = job.check()

except Exception as error:

self.log.warning("{0}[{1}] : unhandled exception on recheck : {2}, give up on retrying".format(

job.module_name, job.name, error))

return stop_retrying

if not ok:

self.log.info('{0}[{1}] : recheck failed, will retry in {2} second(s)'.format(

job.module_name, job.name, job.autodetection_retry()))

return retry_later

self.log.info('{0}[{1}] : recheck successful'.format(job.module_name, job.name))

if not job.create():

return stop_retrying

job.post_check(int(self.min_update_every))

self.checked_jobs[job.module_name].append(check_name)

JobRunner(job).start()

return stop_retrying

def parse_cmd():

opts = sys.argv[:][1:]

debug = False

trace = False

update_every = 1

modules_to_run = list()

v = next((opt for opt in opts if opt.isdigit() and int(opt) >= 1), None)

unknown_modules = set(cmd.modules_to_run) - set(AVAILABLE_MODULES)

if unknown_modules:

logger.error('unknown modules : {0}'.format(sorted(list(unknown_modules))))

safe_print('DISABLE')

return

plugin = Plugin(

cmd.update_every,

cmd.modules_to_run or AVAILABLE_MODULES,

)

HeartBeat(1).start()

if not plugin.setup():

safe_print('DISABLE')

return

plugin.run()

logger.info('exiting from main...')

plugin.shutdown()

if __name__ == '__main__':

main()