host_session/main.py

import json
import os
import re
import sys
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
EXIT_RETRYABLE = 1
EXIT_PERMANENT = 2
REFUSED_STATUS_MIN = 400
REFUSED_STATUS_MAX = 499


class ConfigError(Exception):
    pass


class DbRefused(Exception):
    pass


def log(level, msg, **fields):
    print(json.dumps({"level": level, "msg": msg, **fields}, default=str), flush=True)


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():
    try:
        with open(SIGNALS_FILE) as file:
            names = re.findall(r"name:\s*'([^']+)'", file.read())
    except OSError as error:
        raise ConfigError(f"cannot read {SIGNALS_FILE}: {error}")
    hosts = sorted(name[: -len("-uptime")] for name in names if name.endswith("-uptime"))
    if not hosts:
        raise ConfigError(f"{SIGNALS_FILE} lists no <host>-uptime signal")
    return hosts


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 ConfigError("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 belongs_to_boot(time_ms, uptime_s, boot_ms):
    return abs(time_ms - uptime_s * 1000 - boot_ms) < MATCH_MS


def recent_hours(uptime, boot_ms):
    hours = last_samples(uptime, boot_ms - LOOKBACK_MS, boot_ms, "1hr")
    return [time_ms for time_ms, _ in reversed(hours[-2:])]


def last_sample_of_an_earlier_boot(uptime, hour_ms, boot_ms):
    window_end = min(hour_ms + 2 * HOUR_MS, boot_ms + MATCH_MS)
    found = last_samples(uptime, hour_ms, window_end, "1min")
    earlier = [(t, v) for t, v in found if not belongs_to_boot(t, v, boot_ms)]
    return earlier[-1] if earlier else None


def previous_session(host, boot_ms):
    uptime = f"{host}-uptime"
    for hour_ms in recent_hours(uptime, boot_ms):
        last = last_sample_of_an_earlier_boot(uptime, hour_ms, boot_ms)
        if last is not None:
            end_ms, uptime_s = last
            return int(end_ms - uptime_s * 1000), int(end_ms)
    return None


def is_distinct_session(session, boot_ms):
    start_ms, end_ms = session
    return end_ms > start_ms and abs(start_ms - boot_ms) >= MATCH_MS


def read_db_answer(body):
    status = body.get("statusCode", 200)
    if status == 200:
        return body.get("data")
    message = f"db-controller {status}: {body.get('errorMessage')}"
    if REFUSED_STATUS_MIN <= status <= REFUSED_STATUS_MAX:
        raise DbRefused(message)
    raise RuntimeError(message)


def db_call(method, path, payload):
    response = requests.request(method, DB_URL + path, json=payload, timeout=(5, 30))
    response.raise_for_status()
    return read_db_answer(response.json())


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 0, 0
    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 store_previous_session(host, rows, session):
    stored = find_row(rows, session[0])
    if stored is not None and stored["end_time"] != stored["start_time"]:
        log("INFO", "previous session already stored", id=stored["id"], start=session[0], end=session[1])
        return
    row = build_row(host, *session)
    if stored is None:
        created = db_call("POST", f"/api/db/{TABLE}/", row)
        log("INFO", "previous session stored", id=created["id"], start=session[0], end=session[1], coverage=row["coverage"])
        return
    db_call("PUT", f"/api/db/{TABLE}/{stored['id']}/", row)
    log("INFO", "previous session closed", id=stored["id"], start=session[0], end=session[1], coverage=row["coverage"])


def close_previous_session(host, rows, boot_ms):
    session = previous_session(host, boot_ms)
    if session is None:
        log("INFO", "no previous session in the uptime history")
        return
    if not is_distinct_session(session, boot_ms):
        log("WARN", "previous session ignored, it is the current boot", start=session[0], end=session[1], boot=boot_ms)
        return
    try:
        store_previous_session(host, rows, session)
    except DbRefused as error:
        log("ERROR", "previous session not stored, the database refused it", start=session[0], end=session[1], error=str(error))


def open_current_session(boot_ms):
    created = db_call("POST", f"/api/db/{TABLE}/", open_row(boot_ms))
    log("INFO", "current session opened", id=created["id"], start=boot_ms)


def main():
    boot_ms = current_boot_ms()
    rows = stored_rows()
    if find_row(rows, boot_ms) is not None:
        return
    close_previous_session(own_host(), rows, boot_ms)
    open_current_session(boot_ms)


def run():
    try:
        main()
    except (ConfigError, DbRefused) as error:
        log("ERROR", "this run can not succeed until the script or the schema changes", error=str(error))
        return EXIT_PERMANENT
    except Exception as error:
        log("ERROR", "run failed, it is tried again", error=repr(error))
        return EXIT_RETRYABLE
    return 0


if __name__ == "__main__":
    sys.exit(run())