108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Publish Atlas request availability from deduplicated daily rollups."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Iterable
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
VM_URL = os.environ.get(
|
|
"VM_URL", "http://victoria-metrics-single-server:8428"
|
|
).rstrip("/")
|
|
SCOPE = "atlas"
|
|
DEFINITION = "request-v4"
|
|
REQUESTS_METRIC = "atlas:availability:requests_1d"
|
|
FAILURES_METRIC = "atlas:availability:failures_1d"
|
|
OUTPUT_METRIC = "atlas:availability:ratio_365d"
|
|
WINDOW_DAYS = 365
|
|
|
|
|
|
def parse_export(lines: Iterable[bytes]) -> dict[int, float]:
|
|
"""Return the last exported value for each timestamp."""
|
|
points: dict[int, float] = {}
|
|
for raw_line in lines:
|
|
if not raw_line.strip():
|
|
continue
|
|
series = json.loads(raw_line)
|
|
for timestamp, value in zip(series["timestamps"], series["values"], strict=True):
|
|
points[int(timestamp)] = float(value)
|
|
return points
|
|
|
|
|
|
def fetch_rollup(metric: str, start: datetime, end: datetime) -> dict[int, float]:
|
|
"""Stream one compact rollup series from VictoriaMetrics."""
|
|
matcher = (
|
|
f'{{__name__="{metric}",scope="{SCOPE}",definition="{DEFINITION}"}}'
|
|
)
|
|
query = urlencode(
|
|
{
|
|
"match[]": matcher,
|
|
"start": start.isoformat().replace("+00:00", "Z"),
|
|
"end": end.isoformat().replace("+00:00", "Z"),
|
|
}
|
|
)
|
|
with urlopen(f"{VM_URL}/api/v1/export?{query}", timeout=600) as response:
|
|
return parse_export(response)
|
|
|
|
|
|
def calculate_availability(requests: float, failures: float) -> float:
|
|
"""Calculate the bounded successful-request ratio."""
|
|
if requests <= 0:
|
|
raise ValueError("availability requires at least one observed request")
|
|
if failures < 0:
|
|
raise ValueError("failed request count cannot be negative")
|
|
return max(0.0, min(1.0, 1.0 - (failures / requests)))
|
|
|
|
|
|
def render_metric(value: float, timestamp_ms: int) -> str:
|
|
"""Render one VictoriaMetrics Prometheus-import sample."""
|
|
return (
|
|
f'{OUTPUT_METRIC}{{definition="{DEFINITION}",scope="{SCOPE}",'
|
|
f'rollup="yearly"}} {value:.12f} {timestamp_ms}\n'
|
|
)
|
|
|
|
|
|
def publish(value: float, timestamp_ms: int) -> None:
|
|
"""Write the calculated annual ratio to VictoriaMetrics."""
|
|
request = Request(
|
|
f"{VM_URL}/api/v1/import/prometheus",
|
|
data=render_metric(value, timestamp_ms).encode(),
|
|
headers={"Content-Type": "text/plain"},
|
|
method="POST",
|
|
)
|
|
with urlopen(request, timeout=30) as response:
|
|
if response.status not in {200, 204}:
|
|
raise RuntimeError(f"VictoriaMetrics import returned HTTP {response.status}")
|
|
|
|
|
|
def main() -> None:
|
|
"""Rebuild and publish the rolling request-availability sample."""
|
|
end = datetime.now(timezone.utc)
|
|
start = end - timedelta(days=WINDOW_DAYS)
|
|
requests = sum(fetch_rollup(REQUESTS_METRIC, start, end).values())
|
|
failures = sum(fetch_rollup(FAILURES_METRIC, start, end).values())
|
|
availability = calculate_availability(requests, failures)
|
|
timestamp_ms = time.time_ns() // 1_000_000
|
|
publish(availability, timestamp_ms)
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"requests": requests,
|
|
"failures": failures,
|
|
"availability_percent": availability * 100,
|
|
"timestamp_ms": timestamp_ms,
|
|
},
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|