72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal SLO burn rate tracker for Prometheus."""
|
|
|
|
import argparse
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
|
|
import yaml
|
|
from prometheus_api_client import PrometheusConnect
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
console = Console()
|
|
|
|
|
|
def load_config(path: str) -> dict:
|
|
with open(path) as f:
|
|
return yaml.safe_load(f)
|
|
|
|
|
|
def evaluate_slo(prom: PrometheusConnect, slo: dict, window: str) -> dict:
|
|
result = prom.custom_query(slo["query"])
|
|
if not result:
|
|
return {"name": slo["name"], "value": None, "target": slo["target"], "ok": False}
|
|
value = float(result[0]["value"][1])
|
|
return {
|
|
"name": slo["name"],
|
|
"value": value,
|
|
"target": slo["target"],
|
|
"ok": value >= slo["target"],
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="SLO burn rate tracker")
|
|
parser.add_argument("--config", required=True)
|
|
parser.add_argument("--prometheus", default="http://localhost:9090")
|
|
parser.add_argument("--window", default="7d")
|
|
args = parser.parse_args()
|
|
|
|
cfg = load_config(args.config)
|
|
prom = PrometheusConnect(url=args.prometheus, disable_ssl=True)
|
|
|
|
table = Table(title=f"SLO Status ({args.window} window)")
|
|
table.add_column("SLO", style="cyan")
|
|
table.add_column("Target", justify="right")
|
|
table.add_column("Current", justify="right")
|
|
table.add_column("Status", justify="center")
|
|
|
|
all_ok = True
|
|
for slo in cfg.get("slos", []):
|
|
result = evaluate_slo(prom, slo, args.window)
|
|
status = "[green]OK[/green]" if result["ok"] else "[red]BURNING[/red]"
|
|
value_str = f"{result['value']:.4f}" if result["value"] is not None else "n/a"
|
|
table.add_row(result["name"], str(result["target"]), value_str, status)
|
|
if not result["ok"]:
|
|
all_ok = False
|
|
|
|
console.print(table)
|
|
sys.exit(0 if all_ok else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|
|
def burn_rate(error_rate: float, window_hours: float, target: float) -> float:
|
|
"""Calculate burn rate: how fast the error budget is being consumed."""
|
|
budget = 1.0 - target
|
|
if budget == 0:
|
|
return float("inf")
|
|
return error_rate / budget * (window_hours / (30 * 24))
|