tuxfight3r · April 11, 2025 12:30 · Apr 11, 2025
diff --git a/script.py b/script.py
@@ -0,0 +1,133 @@
+import os
+import time
+import re
+from datetime import datetime, time as dt_time
+from kubernetes import client, config
+from prometheus_client import start_http_server, Gauge
+import pytz
+
+uptime_gauge = Gauge(
+    'namespace_uptime_window_info',
+    'Uptime window info for namespace',
+    ['namespace', 'start', 'end', 'active', 'annotation_found', 'schedule_type']
+)
+
+# Patterns
+ABSOLUTE_PATTERN = re.compile(r'^\d{4}-\d{2}-\d{2}T.*-\d{4}-\d{2}-\d{2}T.*$')
+RECURRING_PATTERN = re.compile(r'^(Mon|Tue|Wed|Thu|Fri|Sat|Sun)(-(Mon|Tue|Wed|Thu|Fri|Sat|Sun))?\s+\d{2}:\d{2}-\d{2}:\d{2}\s+[\w/_+-]+$')
+
+WEEKDAYS = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
+
+def parse_absolute_window(window_str):
+    try:
+        start_str, end_str = window_str.split('-')
+        start = datetime.fromisoformat(start_str)
+        end = datetime.fromisoformat(end_str)
+        return start, end, 'absolute'
+    except Exception as e:
+        print(f"[ERROR] Invalid absolute window format: {window_str} — {e}")
+        return None, None, None
+
+def parse_recurring_window(window_str):
+    try:
+        parts = window_str.split()
+        days_part, time_part, timezone_str = parts[0], parts[1], parts[2]
+        tz = pytz.timezone(timezone_str)
+
+        # Handle day range: Mon-Fri
+        if '-' in days_part:
+            start_day, end_day = days_part.split('-')
+            valid_days = WEEKDAYS[WEEKDAYS.index(start_day):WEEKDAYS.index(end_day)+1]
+        else:
+            valid_days = [days_part]
+
+        # Handle time range
+        start_time_str, end_time_str = time_part.split('-')
+        start_time = datetime.strptime(start_time_str, "%H:%M").time()
+        end_time = datetime.strptime(end_time_str, "%H:%M").time()
+
+        now_utc = datetime.now(pytz.UTC)
+        now_local = now_utc.astimezone(tz)
+        current_day = now_local.strftime('%a')
+        current_time = now_local.time()
+
+        is_active = (
+            current_day in valid_days and
+            start_time <= current_time <= end_time
+        )
+
+        return now_local.replace(hour=start_time.hour, minute=start_time.minute).isoformat(), \
+               now_local.replace(hour=end_time.hour, minute=end_time.minute).isoformat(), \
+               'recurring', is_active
+
+    except Exception as e:
+        print(f"[ERROR] Invalid recurring window format: {window_str} — {e}")
+        return "", "", "", False
+
+def is_now_within(start, end):
+    now = datetime.now(pytz.UTC)
+    return start <= now <= end
+
+def scrape_namespace_annotations():
+    try:
+        config.load_incluster_config()
+    except:
+        config.load_kube_config()
+
+    v1 = client.CoreV1Api()
+    namespaces = v1.list_namespace().items
+
+    for ns in namespaces:
+        name = ns.metadata.name
+        annotations = ns.metadata.annotations or {}
+        window = annotations.get("downscaler/uptime", "")
+        annotation_found = "true" if window else "false"
+        active = "false"
+        start_str = ""
+        end_str = ""
+        schedule_type = ""
+
+        if annotation_found == "true":
+            if ABSOLUTE_PATTERN.match(window):
+                start, end, schedule_type = parse_absolute_window(window)
+                if start and end:
+                    active = str(is_now_within(start, end)).lower()
+                    start_str = start.isoformat()
+                    end_str = end.isoformat()
+            elif RECURRING_PATTERN.match(window):
+                start_str, end_str, schedule_type, is_active = parse_recurring_window(window)
+                active = str(is_active).lower()
+            else:
+                print(f"[WARN] Unrecognized format: {window}")
+                schedule_type = "unknown"
+        else:
+            schedule_type = "none"
+
+        # Emit metric
+        uptime_gauge.labels(
+            namespace=name,
+            start=start_str,
+            end=end_str,
+            active=active,
+            annotation_found=annotation_found,
+            schedule_type=schedule_type
+        ).set(1.0 if active == "true" else 0.0)
+
+if __name__ == "__main__":
+    print("[INFO] Starting Uptime Exporter on port 9116")
+    start_http_server(9116)
+
+    try:
+        interval = int(os.getenv("SCRAPE_INTERVAL_SECONDS", "60"))
+        if interval <= 0:
+            raise ValueError
+    except ValueError:
+        print("[WARN] Invalid SCRAPE_INTERVAL_SECONDS, defaulting to 60")
+        interval = 60
+
+    while True:
+        try:
+            scrape_namespace_annotations()
+        except Exception as e:
+            print(f"[ERROR] Exception while scraping: {e}")
+        time.sleep(interval)