tracker.py

import json
import logging
import os
from datetime import date, datetime, timezone

import requests

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
log = logging.getLogger(__name__)


def fetch_vms(netbox_url: str, token: str) -> list[dict]:
    headers = {"Authorization": f"Token {token}"}
    url = f"{netbox_url}/api/virtualization/virtual-machines/?limit=100"
    vms: list[dict] = []
    while url:
        resp = requests.get(url, headers=headers, timeout=30)
        resp.raise_for_status()
        payload = resp.json()
        vms.extend(payload["results"])
        url = payload.get("next")
    return vms


def _name(obj: dict | None) -> str | None:
    return obj["name"] if obj else None


def build_entry(vm: dict) -> dict:
    cf = vm.get("custom_fields") or {}
    return {
        "name": vm["name"],
        "netbox_url": vm.get("display_url"),
        "status": (vm.get("status") or {}).get("value"),
        "tenant": _name(vm.get("tenant")),
        "cluster": _name(vm.get("cluster")),
        "site": _name(vm.get("site")),
        "tags": [t["slug"] for t in vm.get("tags") or []],
        "description": vm.get("description") or None,
        "environment": cf.get("environment"),
        "project": cf.get("project"),
        "expire_date": cf.get("expire_date"),
    }


def notify_mattermost(webhook_url: str, expiring_soon: list[dict], days: int) -> None:
    lines = [f"#### {len(expiring_soon)} VM(s) expiring within the next {days} days"]
    for vm in expiring_soon:
        lines.append(
            f"- [{vm['name']}]({vm['netbox_url']}) ({vm['tenant']}) — expires {vm['expire_date']}"
        )
    resp = requests.post(webhook_url, json={"text": "\n".join(lines)}, timeout=30)
    resp.raise_for_status()


def main() -> None:
    token = os.environ["NETBOX_TOKEN"]
    netbox_url = os.environ.get("NETBOX_URL", "https://netbox.ethquokkaops.io").rstrip("/")
    expiring_soon_days = int(os.environ.get("EXPIRING_SOON_DAYS", "7"))
    mm_webhook_url = os.environ.get("MM_WEBHOOK_URL")

    today = datetime.now(timezone.utc).date()
    try:
        vms = fetch_vms(netbox_url, token)
    except requests.exceptions.RequestException as e:
        log.error(f"NetBox API request failed: {e}")
        raise SystemExit(1)
    log.info(f"Fetched {len(vms)} VMs from {netbox_url}")

    expired, expiring_soon, no_expiry = [], [], []
    for vm in vms:
        entry = build_entry(vm)
        raw = entry["expire_date"]
        if not raw:
            no_expiry.append(entry)
            continue
        try:
            expire = date.fromisoformat(raw)
        except ValueError:
            log.warning(f"VM '{entry['name']}' has unparseable expire_date: {raw!r}")
            no_expiry.append(entry)
            continue
        delta = (expire - today).days
        entry["days_until_expiry"] = delta
        if delta < 0:
            expired.append(entry)
        elif delta < expiring_soon_days:
            expiring_soon.append(entry)

    expired.sort(key=lambda e: e["days_until_expiry"])
    expiring_soon.sort(key=lambda e: e["days_until_expiry"])
    no_expiry.sort(key=lambda e: e["name"])

    log.info(
        f"Expired: {len(expired)}, "
        f"expiring within {expiring_soon_days}d: {len(expiring_soon)}, "
        f"no expiry set: {len(no_expiry)}"
    )
    print(json.dumps({
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "netbox_url": netbox_url,
        "expiring_soon_days": expiring_soon_days,
        "expired": expired,
        "expiring_soon": expiring_soon,
        "no_expiry": no_expiry,
    }, indent=2))

    if mm_webhook_url and expiring_soon:
        try:
            notify_mattermost(mm_webhook_url, expiring_soon, expiring_soon_days)
        except requests.exceptions.RequestException as e:
            log.error(f"Mattermost webhook request failed: {e}")
            raise SystemExit(1)
        log.info(f"Notified Mattermost about {len(expiring_soon)} expiring VM(s)")


if __name__ == "__main__":
    main()