import os import re 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" SIGNALS_FILE = "/app/signals.yml" 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["name"]: item for item in response.json()["data"]["series"]} def signal_hosts(): with open(SIGNALS_FILE) as file: names = re.findall(r"name:\s*'([^']+)'", file.read()) return sorted(name[: -len("-uptime")] for name in names if name.endswith("-uptime")) def owns(host, now_ms): path = f"/api/history/?signals={host}-uptime&start={int(now_ms - MINUTE_MS)}&end={int(now_ms)}&resolution=1min" status = requests.get(SIDECAR_URL + path, timeout=(5, 80)).status_code if status not in (200, 403): raise RuntimeError(f"history probe for {host} answered {status}") return status == 200 def own_host(): now_ms = time.time() * 1000 host = next((name for name in signal_hosts() if owns(name, now_ms)), None) if host is None: raise RuntimeError("no uptime signal of this gateway is listed in signals.yml") return host 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(host, boot_ms): uptime = f"{host}-uptime" 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(host, start_ms, end_ms): cpu, ram, disk = f"{host}-cpu", f"{host}-ram", f"{host}-disk" 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(host, rows, boot_ms): session = previous_session(host, boot_ms) if session is None: print("no previous session in the uptime history", flush=True) return row = build_row(host, *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(own_host(), rows, boot_ms) print("current session opened", db_call("POST", f"/api/db/{TABLE}/", open_row(boot_ms)), flush=True) if __name__ == "__main__": main()