Refactor filter_plugins

playbooks/roles/gatus/defaults/main.yml

@@ -1,8 +1,8 @@
 ---
-gatus_external_endpoints_backups: "{{ hostvars | hosts_backup_to_gatus() }}"
+gatus_external_endpoints_backups: "{{ hostvars | gatus_from_hosts_backup() }}"
 
-gatus_endpoints_hosts: "{{ opentofu.hosts | hosts_to_gatus() }}"
-gatus_endpoints_services: "{{ all_services | services_to_gatus() }}"
+gatus_endpoints_hosts: "{{ opentofu.hosts | gatus_from_hosts() }}"
+gatus_endpoints_services: "{{ all_services | gatus_from_services() }}"
 
 gatus_federation_tester: "https://federationtester.matrix.org/api/report?server_name=msrg.cc"
 

playbooks/roles/gatus/filter_plugins/gatus.py

@@ -0,0 +1,100 @@
+class FilterModule(object):
+    def filters(self):
+        return {
+            'gatus_from_hosts': self.from_hosts,
+            'gatus_from_hosts_backup': self.from_hosts_backup,
+            'gatus_from_services': self.from_services,
+        }
+
+    default_alerts = [
+        {
+            "type": "ntfy",
+            "send-on-resolved": True,
+        },
+        {
+            "type": "email",
+            "send-on-resolved": True,
+        },
+    ]
+
+    def from_hosts(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 from_hosts_backup(self, hostvars):
+        result = []
+        backup_alerts = []
+        for a in self.default_alerts:
+            backup_alerts.append(dict(a, **{
+                'failure-threshold': 1,
+                'success-threshold': 1
+            }))
+
+        for name, host_data in hostvars.items():
+            if not host_data.get("host_backup_gatus_token"):
+                continue
+            result.append({
+                "name": f"backup@{name}",
+                "group": "8-backups",
+                "token": host_data["host_backup_gatus_token"],
+                "alerts": backup_alerts,
+            })
+        return result
+
+    def from_services(self, services):
+        result = []
+
+        default_conditions = [
+            "[STATUS] == any(200, 204)",
+            "[CERTIFICATE_EXPIRATION] > 48h"
+        ]
+
+        for name, service in services.items():
+            if not bool(service.get("host")):
+                continue
+
+            if mon := service.get("monitoring"):
+
+                if service.get("dns"):
+                    url = f"https://{service["dns"][0]['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": name,
+                    "group": mon.get("group"),
+                    "url": url,
+                    "conditions": conditions,
+                    "interval": mon.get("interval"),
+                    "alerts": self.default_alerts,
+                    "ui": {
+                        "hide-url": True
+                    }
+                }
+
+                result.append(new_endpoint)
+
+        return result