125 lines
4.3 KiB
Python
125 lines
4.3 KiB
Python
import os
|
|
import time
|
|
|
|
import requests
|
|
|
|
SIDECAR_URL = os.getenv("SIDECAR_URL", "http://sidecar:80")
|
|
DB_URL = os.getenv("DB_CONTROLLER_URL", "http://sidecar:8081")
|
|
TABLE = "host_session"
|
|
UPTIME, CPU, RAM, DISK = 63, 21, 22, 23
|
|
MINUTE_MS = 60000
|
|
HOUR_MS = 60 * MINUTE_MS
|
|
LOOKBACK_MS = 60 * 24 * HOUR_MS
|
|
MATCH_MS = 2 * MINUTE_MS
|
|
|
|
|
|
def read_history(signals, start_ms, end_ms, resolution):
|
|
names = ",".join(str(signal) for signal in signals)
|
|
path = f"/api/history/?signals={names}&start={int(start_ms)}&end={int(end_ms)}&resolution={resolution}"
|
|
response = requests.get(SIDECAR_URL + path, timeout=(5, 80))
|
|
response.raise_for_status()
|
|
return {item["signalId"]: item for item in response.json()["data"]["series"]}
|
|
|
|
|
|
def samples(series):
|
|
start, step = series["startMs"], series["stepMs"]
|
|
return [(start + i * step, value) for i, value in enumerate(series["values"]) if value is not None]
|
|
|
|
|
|
def last_samples(signal, start_ms, end_ms, resolution):
|
|
if end_ms <= start_ms:
|
|
return []
|
|
found = samples(read_history([signal], start_ms, end_ms, resolution)[signal])
|
|
return [(t, v) for t, v in found if t < end_ms]
|
|
|
|
|
|
def current_boot_ms():
|
|
with open("/proc/uptime") as file:
|
|
uptime_s = float(file.read().split()[0])
|
|
return int(time.time() * 1000 - uptime_s * 1000)
|
|
|
|
|
|
def previous_session(boot_ms):
|
|
hours = last_samples(UPTIME, boot_ms - LOOKBACK_MS, boot_ms, "1hr")
|
|
if not hours:
|
|
return None
|
|
minutes = last_samples(UPTIME, hours[-1][0], min(hours[-1][0] + HOUR_MS, boot_ms), "1min")
|
|
if not minutes:
|
|
return None
|
|
seconds = last_samples(UPTIME, minutes[-1][0], min(minutes[-1][0] + MINUTE_MS, boot_ms), "1s")
|
|
end_ms, uptime_s = (seconds or minutes)[-1]
|
|
return int(end_ms - uptime_s * 1000), int(end_ms)
|
|
|
|
|
|
def db_call(method, path, payload):
|
|
response = requests.request(method, DB_URL + path, json=payload, timeout=(5, 30))
|
|
response.raise_for_status()
|
|
body = response.json()
|
|
if body.get("statusCode", 200) != 200:
|
|
raise RuntimeError(f"db-controller {body.get('statusCode')}: {body.get('errorMessage')}")
|
|
return body.get("data")
|
|
|
|
|
|
def stored_rows():
|
|
return db_call("POST", f"/api/db-query/{TABLE}/", {"filters": [], "orderBy": ["start_time"], "limit": 1000}) or []
|
|
|
|
|
|
def find_row(rows, start_ms):
|
|
return next((row for row in rows if abs(row["start_time"] - start_ms) < MATCH_MS), None)
|
|
|
|
|
|
def open_row(start_ms):
|
|
row = {"start_time": start_ms, "end_time": start_ms, "duration_s": 0, "coverage": 0}
|
|
for prefix in ("cpu", "ram", "disk"):
|
|
row[f"{prefix}_mean"], row[f"{prefix}_max"] = 0, 0
|
|
return row
|
|
|
|
|
|
def mean_max(series, start_ms, end_ms):
|
|
values = [v for t, v in samples(series) if start_ms <= t < end_ms]
|
|
if not values:
|
|
return None, None
|
|
return sum(values) / len(values), max(values)
|
|
|
|
|
|
def coverage(series, start_ms, end_ms):
|
|
total = max(1, (end_ms - start_ms) // MINUTE_MS)
|
|
present = sum(1 for t, _ in samples(series) if start_ms <= t < end_ms)
|
|
return min(1.0, present / total)
|
|
|
|
|
|
def build_row(start_ms, end_ms):
|
|
history = read_history([CPU, RAM, DISK], start_ms, end_ms, "1min")
|
|
row = {"start_time": start_ms, "end_time": end_ms, "duration_s": (end_ms - start_ms) // 1000}
|
|
for prefix, signal in (("cpu", CPU), ("ram", RAM), ("disk", DISK)):
|
|
row[f"{prefix}_mean"], row[f"{prefix}_max"] = mean_max(history[signal], start_ms, end_ms)
|
|
row["coverage"] = coverage(history[CPU], start_ms, end_ms)
|
|
return row
|
|
|
|
|
|
def close_previous(rows, boot_ms):
|
|
session = previous_session(boot_ms)
|
|
if session is None:
|
|
print("no previous session in the uptime history", flush=True)
|
|
return
|
|
row = build_row(*session)
|
|
stored = find_row(rows, session[0])
|
|
if stored is None:
|
|
print("previous session stored", db_call("POST", f"/api/db/{TABLE}/", row), row, flush=True)
|
|
elif stored["end_time"] == stored["start_time"]:
|
|
db_call("PUT", f"/api/db/{TABLE}/{stored['id']}/", row)
|
|
print("previous session closed", stored["id"], row, flush=True)
|
|
|
|
|
|
def main():
|
|
boot_ms = current_boot_ms()
|
|
rows = stored_rows()
|
|
if find_row(rows, boot_ms) is not None:
|
|
return
|
|
close_previous(rows, boot_ms)
|
|
print("current session opened", db_call("POST", f"/api/db/{TABLE}/", open_row(boot_ms)), flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|