mirror of
https://github.com/TrezOne/docker-mods-uptime-kuma-timeout-fix.git
synced 2026-07-22 02:50:29 -04:00
- Refactored whole code so that its not such a mess
- Updated documentation so that it explains better how the mod works, requirements, setting up labels and notifications - Reduced number of API calls when swag container restarts - Added better support for setting up notifications and fixed issue with default notifications not being applied - Fixed an issue with mod crashing when there were Manually added monitors in Uptime Kuma - Fixed an issue with labels that container numeric values - Added basic support for Monitor Groups - More log messages so that its easier to grasp what is happening - Fixed a crash when mod was executed while uptime kuma was not running (mod just gently stops now)
This commit is contained in:
labmonkey
@@ -0,0 +1,320 @@
|
||||
import requests
|
||||
from uptime_kuma_api.api import UptimeKumaApi, MonitorType
|
||||
from auto_uptime_kuma.log import Log
|
||||
from auto_uptime_kuma.config_service import ConfigService
|
||||
|
||||
|
||||
class UptimeKumaService:
|
||||
|
||||
api: UptimeKumaApi
|
||||
swag_tag_name = "swag"
|
||||
swag_tag_color = "#ff4f97"
|
||||
swag_tag = None
|
||||
monitors: list
|
||||
groups: list
|
||||
default_notifications: list
|
||||
|
||||
config_service: ConfigService
|
||||
|
||||
default_monitor_data = {
|
||||
"type": MonitorType.HTTP,
|
||||
"description": "Automatically generated by SWAG auto-uptime-kuma",
|
||||
}
|
||||
|
||||
def __init__(self, config_service):
|
||||
self.config_service = config_service
|
||||
|
||||
def connect(self, url, username, password):
|
||||
response = requests.get(url, allow_redirects=True, timeout=5)
|
||||
if response.status_code != 200:
|
||||
Log.info(
|
||||
f"Unable to connect to UptimeKuma at '{url}' (Status code: {response.status_code})."
|
||||
" Please check if the host is running."
|
||||
)
|
||||
return False
|
||||
|
||||
self.api = UptimeKumaApi(url)
|
||||
self.api.login(username, password)
|
||||
|
||||
return True
|
||||
|
||||
def disconnect(self):
|
||||
"""
|
||||
API has to be disconnected at the end as the connection is blocking
|
||||
"""
|
||||
self.api.disconnect()
|
||||
|
||||
def load_data(self):
|
||||
monitors = self.api.get_monitors()
|
||||
self.get_swag_tag()
|
||||
|
||||
self.default_notifications = self.get_default_notifications()
|
||||
|
||||
self.monitors = [
|
||||
monitor
|
||||
for monitor in monitors
|
||||
if monitor["type"] != MonitorType.GROUP
|
||||
and self.get_monitor_swag_tag_value(monitor) is not None
|
||||
]
|
||||
self.groups = [
|
||||
group
|
||||
for group in monitors
|
||||
if group["type"] == MonitorType.GROUP
|
||||
and self.get_monitor_swag_tag_value(group) is not None
|
||||
]
|
||||
|
||||
def build_monitor_data(self, container_name, configured_monitor_data):
|
||||
"""
|
||||
Some of the container label values might have to be converted before sending to API.
|
||||
Additionally merge default config with label config.
|
||||
"""
|
||||
result_data = dict(self.default_monitor_data)
|
||||
result_data.update(
|
||||
{
|
||||
"name": container_name.title(),
|
||||
"url": f"https://{container_name}.{self.config_service.domain_name}",
|
||||
}
|
||||
)
|
||||
result_data.update(configured_monitor_data)
|
||||
|
||||
# Convert strings that are lists in API
|
||||
for key in ["accepted_statuscodes", "notificationIDList"]:
|
||||
if key in result_data and isinstance(result_data[key], str):
|
||||
result_data[key] = result_data[key].split(",")
|
||||
|
||||
# If Monitor Groups are used then create them if needed and switch into ID as value
|
||||
if "parent" in result_data and not str(result_data["parent"]).isdigit():
|
||||
if self.group_exists(result_data["parent"]):
|
||||
group_data = self.get_group(result_data["parent"])
|
||||
else:
|
||||
group_data = self.create_group(result_data["parent"])
|
||||
result_data["parent"] = group_data["id"]
|
||||
|
||||
if self.default_notifications:
|
||||
default_notification_ids = [
|
||||
notification["id"] for notification in self.default_notifications
|
||||
]
|
||||
if "notificationIDList" in result_data:
|
||||
result_data["notificationIDList"] += default_notification_ids
|
||||
result_data["notificationIDList"] = list(
|
||||
set(result_data["notificationIDList"])
|
||||
)
|
||||
else:
|
||||
result_data["notificationIDList"] = default_notification_ids
|
||||
|
||||
# All numeric values sent to API have to be of type int
|
||||
for key, value in result_data.items():
|
||||
if str(value).isdigit():
|
||||
result_data[key] = int(value)
|
||||
|
||||
return result_data
|
||||
|
||||
def get_monitor(self, container_name):
|
||||
for monitor in self.monitors:
|
||||
swag_tag = self.get_monitor_swag_tag_value(monitor)
|
||||
if swag_tag is not None and swag_tag == container_name.lower():
|
||||
return monitor
|
||||
return None
|
||||
|
||||
def monitor_exists(self, container_name):
|
||||
return self.get_monitor(container_name) is not None
|
||||
|
||||
def create_monitor(self, container_name, monitor_data):
|
||||
monitor_data = self.build_monitor_data(container_name, monitor_data)
|
||||
if self.monitor_exists(container_name):
|
||||
Log.info(
|
||||
f"Uptime Kuma already contains Monitor '{monitor_data['name']}'"
|
||||
f" for container '{container_name}', skipping..."
|
||||
)
|
||||
return None
|
||||
|
||||
Log.info(
|
||||
f"Adding Monitor '{monitor_data['name']}' for container '{container_name}'"
|
||||
)
|
||||
|
||||
monitor = self.api.add_monitor(**monitor_data)
|
||||
|
||||
self.api.add_monitor_tag(
|
||||
tag_id=self.get_swag_tag()["id"],
|
||||
monitor_id=monitor["monitorID"],
|
||||
value=container_name.lower(),
|
||||
)
|
||||
|
||||
self.config_service.create_config(container_name, monitor_data)
|
||||
|
||||
monitor = self.api.get_monitor(monitor["monitorID"])
|
||||
self.monitors.append(monitor)
|
||||
|
||||
return monitor
|
||||
|
||||
def edit_monitor(self, container_name, monitor_data):
|
||||
"""
|
||||
Please note that due to API limitations the "update" action
|
||||
is actually "delete" followed by "add"
|
||||
so that in the end the monitors are actually recreated
|
||||
"""
|
||||
new_monitor_data = self.build_monitor_data(container_name, monitor_data)
|
||||
existing_monitor_data = self.get_monitor(container_name)
|
||||
old_content = self.config_service.read_config_content(container_name)
|
||||
new_content = self.config_service.build_config_content(new_monitor_data)
|
||||
|
||||
if not old_content == new_content:
|
||||
Log.info(
|
||||
"Updating (Delete and Add) Monitor"
|
||||
f" {existing_monitor_data['id']}:{existing_monitor_data['name']}"
|
||||
)
|
||||
self.delete_monitor(container_name)
|
||||
self.create_monitor(container_name, new_monitor_data)
|
||||
else:
|
||||
Log.info(
|
||||
f"Monitor {existing_monitor_data['id']}:{existing_monitor_data['name']}"
|
||||
" has no changes, skipping..."
|
||||
)
|
||||
|
||||
def delete_monitor(self, container_name: str):
|
||||
monitor_data = self.get_monitor(container_name)
|
||||
if monitor_data is not None:
|
||||
Log.info(f"Deleting Monitor {monitor_data['id']}:{monitor_data['name']}")
|
||||
self.api.delete_monitor(monitor_data["id"])
|
||||
|
||||
for i, monitor in enumerate(self.monitors):
|
||||
if monitor["id"] == monitor_data["id"]:
|
||||
del self.monitors[i]
|
||||
break
|
||||
|
||||
def delete_monitors(self, container_names: list[str]):
|
||||
if container_names:
|
||||
Log.info(
|
||||
f"Deleting Monitors for the following containers: {container_names}"
|
||||
)
|
||||
for container_name in container_names:
|
||||
self.delete_monitor(container_name)
|
||||
else:
|
||||
Log.info("No Monitors to delete")
|
||||
|
||||
def get_group(self, group_name):
|
||||
for group in self.groups:
|
||||
swag_tag = self.get_monitor_swag_tag_value(group)
|
||||
if swag_tag is not None and swag_tag == group_name.lower():
|
||||
return group
|
||||
return None
|
||||
|
||||
def group_exists(self, container_name):
|
||||
return self.get_group(container_name) is not None
|
||||
|
||||
def create_group(self, group_name):
|
||||
monitor_data = {
|
||||
"type": MonitorType.GROUP,
|
||||
"name": group_name,
|
||||
"description": "Automatically generated by SWAG auto-uptime-kuma",
|
||||
}
|
||||
if self.group_exists(group_name):
|
||||
Log.info(
|
||||
f"Uptime Kuma already contains Group '{monitor_data['name']}', skipping..."
|
||||
)
|
||||
return
|
||||
|
||||
Log.info(f"Adding Group '{monitor_data['name']}'")
|
||||
|
||||
group = self.api.add_monitor(**monitor_data)
|
||||
|
||||
self.api.add_monitor_tag(
|
||||
tag_id=self.get_swag_tag()["id"],
|
||||
monitor_id=group["monitorID"],
|
||||
value=group_name.lower(),
|
||||
)
|
||||
|
||||
group = self.api.get_monitor(group["monitorID"])
|
||||
self.groups.append(group)
|
||||
|
||||
return group
|
||||
|
||||
## This intentionally does not exist. Groups management by this mod is made
|
||||
## simple so that they can be only added or deleted.
|
||||
# def edit_group(self, group_name):
|
||||
|
||||
def delete_group(self, group_name):
|
||||
group_data = self.get_group(group_name)
|
||||
if group_data is not None:
|
||||
Log.info(f"Deleting Group {group_data['id']}:{group_data['name']}")
|
||||
self.api.delete_monitor(group_data["id"])
|
||||
|
||||
for i, group in enumerate(self.groups):
|
||||
if group["id"] == group_data["id"]:
|
||||
del self.groups[i]
|
||||
break
|
||||
|
||||
def delete_groups(self, group_names: list[str]):
|
||||
if group_names:
|
||||
Log.info(f"Deleting Groups with following names: {group_names}")
|
||||
for group_name in group_names:
|
||||
self.delete_group(group_name)
|
||||
else:
|
||||
Log.info("No Groups to delete")
|
||||
|
||||
def get_swag_tag(self):
|
||||
"""
|
||||
The "swag" tag is used to detect in API which monitors were created using this script.
|
||||
"""
|
||||
# If the tag was not fetched yet
|
||||
if self.swag_tag is None:
|
||||
for tag in self.api.get_tags():
|
||||
if tag["name"] == self.swag_tag_name:
|
||||
self.swag_tag = tag
|
||||
break
|
||||
|
||||
# If the tag was not in API then it has to be created
|
||||
if self.swag_tag is None:
|
||||
self.swag_tag = self.create_swag_tag()
|
||||
|
||||
return self.swag_tag
|
||||
|
||||
def swag_tag_exists(self):
|
||||
return self.get_swag_tag() is not None
|
||||
|
||||
def create_swag_tag(self):
|
||||
self.swag_tag = self.api.add_tag(
|
||||
name=self.swag_tag_name, color=self.swag_tag_color
|
||||
)
|
||||
return self.swag_tag
|
||||
|
||||
def delete_swag_tag(self):
|
||||
swag_tag = self.get_swag_tag()
|
||||
if swag_tag is not None:
|
||||
self.api.delete_tag(self.swag_tag["id"])
|
||||
swag_tag = None
|
||||
|
||||
def get_monitor_swag_tag_value(self, monitor_data):
|
||||
"""
|
||||
This value is container name itself. Used to link containers with monitors.
|
||||
"""
|
||||
for tag in monitor_data.get("tags"):
|
||||
if "name" in tag and tag["name"] == self.swag_tag_name:
|
||||
return tag["value"]
|
||||
return None
|
||||
|
||||
def get_default_notifications(self):
|
||||
default_notifications = [
|
||||
notification
|
||||
for notification in self.api.get_notifications()
|
||||
if notification["isDefault"] is True
|
||||
]
|
||||
|
||||
return default_notifications
|
||||
|
||||
def purge_data(self):
|
||||
"""
|
||||
Deletes all of the monitors created with this script
|
||||
"""
|
||||
Log.info("Purging all Uptime Kuma Monitors added by this mod")
|
||||
|
||||
monitor_names = [
|
||||
self.get_monitor_swag_tag_value(monitor) for monitor in self.monitors
|
||||
]
|
||||
self.delete_monitors(monitor_names)
|
||||
Log.info("Monitor purging finished")
|
||||
group_names = [self.get_monitor_swag_tag_value(group) for group in self.groups]
|
||||
self.delete_groups(group_names)
|
||||
Log.info("Group purging finished")
|
||||
Log.info("Deleting the Swag tag")
|
||||
self.delete_swag_tag()
|
||||
Reference in New Issue
Block a user