python uptime exporter

script.py

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)