updog/updog.py

96 lines
3.4 KiB
Python
Raw Normal View History

from helpers import RepeatedTimer, create_instances, run_check
import sys
import os
import importlib
import inspect
2024-01-30 13:35:56 -06:00
import json
path = os.path.realpath(__file__)
path = path[:path.rfind('/')]
2024-01-30 13:35:56 -06:00
#####################
# 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
2024-01-30 13:35:56 -06:00
checker_extension_imports = []
for ext_name in os.listdir(f'{path}/extensions/checkers'):
if ext_name[0] != '.':
2024-01-30 13:35:56 -06:00
checker_extension_imports.append(f'{ext_name}.{ext_name}')
2024-01-30 13:35:56 -06:00
# checkers contains all the classes for the checker extensions
# e.g. {'CheckerTemplate': checker_template.checker_template.CheckerTemplate}
2024-01-30 13:35:56 -06:00
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
2024-01-30 13:35:56 -06:00
# 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'
2024-01-30 13:35:56 -06:00
with open(config_filename, 'rt') as config_file:
2024-02-20 12:04:52 -06:00
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])
2024-01-30 13:35:56 -06:00
config = json.loads(''.join(config_file.readlines()))
2024-02-20 12:04:52 -06:00
instances = create_instances(config, checkers, alerts, logging)
timers = {}
try:
for service in config.keys():
2024-02-06 12:37:19 -06:00
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()