beepi/playbooks/monitoring/topology/topology.py

import json
import os
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

PROMETHEUS_URL = os.environ.get("PROMETHEUS_URL", "http://prometheus:9090")
LISTEN_PORT = int(os.environ.get("LISTEN_PORT", "8088"))
SPEC_PATH = os.environ.get("SPEC_PATH", "/app/topology-spec.json")
CACHE_SECONDS = int(os.environ.get("CACHE_SECONDS", "10"))

STATUS_OK = "ok"
STATUS_FAIL = "fail"
STATUS_DISABLED = "disabled"
STATUS_UNKNOWN = "unknown"

ARCS = {
    STATUS_OK: "arc__ok",
    STATUS_FAIL: "arc__fail",
    STATUS_DISABLED: "arc__disabled",
    STATUS_UNKNOWN: "arc__unknown",
}

ARC_FIELDS = list(ARCS.values()) + ["arc__used", "arc__free"]

COLORS = {
    STATUS_OK: "green",
    STATUS_FAIL: "red",
    STATUS_DISABLED: "#6e7079",
    STATUS_UNKNOWN: "orange",
}

DASHBOARDS = {
    "apps": "/d/beepi-services",
    "platform": "/d/beepi-services",
    "backup": "/d/beepi-backups",
    "storage": "/d/beepi-backups",
    "edge": "/d/beepi-ingress",
    "host": "/d/beepi-ingress",
}

_lock = threading.Lock()
_cache = {"at": 0.0, "body": None}


def query(expr):
    url = PROMETHEUS_URL + "/api/v1/query?" + urllib.parse.urlencode({"query": expr})

    try:
        with urllib.request.urlopen(url, timeout=8) as response:
            payload = json.load(response)
    except (urllib.error.URLError, OSError, ValueError):
        return None

    if payload.get("status") != "success":
        return None

    result = payload.get("data", {}).get("result", [])

    if not result:
        return None

    try:
        return float(result[0]["value"][1])
    except (KeyError, IndexError, TypeError, ValueError):
        return None


def format_stat(value, unit):
    if value is None:
        return ""

    if unit == "s":
        return "%.0f ms" % (value * 1000) if value < 1 else "%.1f s" % value

    if unit == "h":
        return "%.1f h" % (value / 3600.0)

    if unit == "d":
        return "%.1f d" % (value / 86400.0)

    if unit == "bytes":
        size = float(value)
        for suffix in ("B", "KiB", "MiB", "GiB", "TiB"):
            if size < 1024 or suffix == "TiB":
                return "%.1f %s" % (size, suffix)
            size /= 1024.0

    if unit == "pct":
        return "%.0f%%" % value

    return "%.0f" % value


def evaluate(node):
    enabled = node.get("enabled")

    if enabled:
        value = query(enabled)

        if value is not None and value < 1:
            return STATUS_DISABLED

    up = node.get("up")

    if not up:
        return STATUS_OK

    value = query(up)

    if value is None:
        return STATUS_UNKNOWN

    return STATUS_OK if value >= 1 else STATUS_FAIL


def build():
    with open(SPEC_PATH) as handle:
        spec = json.load(handle)

    nodes = []
    statuses = {}

    for node in spec["nodes"]:
        status = evaluate(node)
        statuses[node["id"]] = status

        entry = {
            "id": node["id"],
            "title": node["title"],
            "subtitle": node.get("subtitle", ""),
            "mainstat": format_stat(query(node["stat"]), node.get("stat_unit", "")) if node.get("stat") else "",
            "detail__group": node.get("group", ""),
            "detail__status": status,
            "detail__used": "",
            "dashboard": node.get("dashboard") or DASHBOARDS.get(node.get("group", ""), "/d/beepi-map"),
        }

        for name in ARC_FIELDS:
            entry[name] = 0.0

        used = query(node["fill"]) if node.get("fill") and status == STATUS_OK else None

        if used is None:
            entry[ARCS[status]] = 1.0
        else:
            used = min(max(used, 0.0), 1.0)
            entry["arc__used"] = used
            entry["arc__free"] = 1.0 - used
            entry["detail__used"] = "%.0f%% used" % (used * 100)

        nodes.append(entry)

    edges = []

    for index, edge in enumerate(spec["edges"]):
        source = statuses.get(edge["source"], STATUS_UNKNOWN)
        target = statuses.get(edge["target"], STATUS_UNKNOWN)
        degraded = STATUS_FAIL in (source, target)
        idle = STATUS_DISABLED in (source, target)

        if degraded:
            color = COLORS[STATUS_FAIL]
        elif idle:
            color = COLORS[STATUS_DISABLED]
        else:
            color = COLORS[STATUS_OK]

        edges.append({
            "id": str(index),
            "source": edge["source"],
            "target": edge["target"],
            "mainstat": edge.get("label", ""),
            "color": color,
            "thickness": 3 if degraded else 1,
        })

    return {"nodes": nodes, "edges": edges}


def cached():
    with _lock:
        now = time.time()

        if _cache["body"] is None or now - _cache["at"] > CACHE_SECONDS:
            _cache["body"] = json.dumps(build()).encode()
            _cache["at"] = now

        return _cache["body"]


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.rstrip("/") not in ("", "/topology.json", "/topology"):
            self.send_error(404)
            return

        try:
            body = cached()
        except Exception:
            self.send_error(500)
            return

        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        pass


if __name__ == "__main__":
    ThreadingHTTPServer(("0.0.0.0", LISTEN_PORT), Handler).serve_forever()