from helpers import RepeatedTimer, create_instances, run_check import sys import os import importlib import inspect import json path = os.path.realpath(__file__) path = path[:path.rfind('/')] ##################### # Import extensions # ##################### # Import checkers # # relative imports suck - https://gideonbrimleaf.github.io/2021/01/26/relative-imports-python.html sys.path.insert(1, f'{path}/extensions/checkers') # importlib used to import stuff programmatically, rather than using the hardcoded import keyword, basically the same but it seems it can't import *just* a class checker_extension_imports = [] for ext_name in os.listdir(f'{path}/extensions/checkers'): if ext_name[0] != '.': checker_extension_imports.append(f'{ext_name}.{ext_name}') # checkers contains all the classes for the checker extensions # e.g. {'CheckerTemplate': checker_template.checker_template.CheckerTemplate} checkers = {} for ext in [importlib.import_module(ext) for ext in checker_extension_imports]: for name, obj in inspect.getmembers(ext): if inspect.isclass(obj): checkers[name] = obj # Import alerts # # same as above, just for alerts sys.path.insert(1, f'{path}/extensions/alerts') alerts_extension_imports = [] for ext_name in os.listdir(f'{path}/extensions/alerts'): if ext_name[0] != '.': alerts_extension_imports.append(f'{ext_name}.{ext_name}') # alerts contains all the classes for the checker extensions # e.g. {'AlertsTemplate': alerts_template.alerts_template.AlertsTemplate} alerts = {} for ext in [importlib.import_module(ext) for ext in alerts_extension_imports]: for name, obj in inspect.getmembers(ext): if inspect.isclass(obj): alerts[name] = obj # Import logging # # same as above, just for logging sys.path.insert(1, f'{path}/extensions/logging') logging_extension_imports = [] for ext_name in os.listdir(f'{path}/extensions/logging'): if ext_name[0] != '.': logging_extension_imports.append(f'{ext_name}.{ext_name}') # logging contains all the classes for the checker extensions # e.g. {'LoggingTemplate': logging_template.logging_template.LoggingTemplate} logging = {} for ext in [importlib.import_module(ext) for ext in logging_extension_imports]: for name, obj in inspect.getmembers(ext): if inspect.isclass(obj): logging[name] = obj # get config from services.json if 'services.json' in os.listdir(path): config_filename = f'{path}/services.json' elif 'services-example.json' in os.listdir(path): config_filename = f'{path}/services-example.json' with open(config_filename, 'rt') as config_file: config_text = ''.join(config_file.readlines()) # Replace static keywords in config keywords = {'%dir%': path} for key in keywords: config_text = config_text.replace(key, keywords[key]) config = json.loads(''.join(config_file.readlines())) instances = create_instances(config, checkers, alerts, logging) timers = {} try: for service in config.keys(): if service == 'global-args': continue timers[service] = RepeatedTimer(config[service]['rate'], run_check, service, instances[service]['checker'], instances[service]['alerts'], instances[service]['logging']) except: # Stops all RepeatedTimers since they're threaded # Doesn't actually stop the threads, just stops repeating it for service in timers.keys(): timers[service].stop()