Replace uptime kuma with gatus

playbooks/filter_plugins/gatus.py

@@ -0,0 +1,86 @@
+class FilterModule(object):
+    def filters(self):
+        return {
+            'hosts_to_gatus': self.hosts_to_gatus,
+            'vault_hosts_backup_to_gatus': self.vault_hosts_backup_to_gatus,
+            'services_to_gatus': self.services_to_gatus,
+        }
+
+    default_alerts = [
+        {
+            "type": "ntfy",
+            "send-on-resolved": True,
+        },
+        {
+            "type": "email",
+            "send-on-resolved": True,
+        },
+    ]
+
+    def hosts_to_gatus(self, hosts):
+        result = []
+        for host in hosts.values():
+            result.append({
+                "name": host["hostname"],
+                "url": f"icmp://{host['fqdn']}",
+                "group": "1-hosts",
+                "conditions": [
+                    "[CONNECTED] == true",
+                ],
+                "alerts": self.default_alerts,
+            })
+        return result
+
+    def vault_hosts_backup_to_gatus(self, hosts):
+        result = []
+        for name, host_data in hosts.items():
+            result.append({
+                "name": f"backup@{name}",
+                "group": "8-backups",
+                "token": host_data["backup"]["gatus_token"],
+                "alerts": self.default_alerts,
+            })
+        return result
+
+    def services_to_gatus(self, services):
+        result = []
+
+        default_conditions = [
+            "[STATUS] == any(200, 204)",
+        ]
+
+        for service in services:
+            if mon := service.get("monitoring"):
+                if service.get("dns"):
+                    dns = service["dns"][0]
+                    url = "https://"
+                    if dns.get("target") != "@":
+                        url += f"{dns["target"]}."
+                    url += dns['domain']
+
+                if mon_url := mon.get("url"):
+                    if mon_url.startswith("/"):
+                        url += mon_url
+                    else:
+                        url = mon_url
+
+                if conditions := mon.get("conditions"):
+                    if conditions[0] == "DEFAULT":
+                        conditions.pop(0)
+                        conditions[:0] = default_conditions
+                    else:
+                        conditions = conditions
+                else:
+                    conditions = default_conditions
+
+                new_endpoint = {
+                    "name": service["name"],
+                    "group": mon.get("group"),
+                    "url": url,
+                    "conditions": conditions,
+                    "alerts": self.default_alerts,
+                }
+
+                result.append(new_endpoint)
+
+        return result