diff --git a/README.md b/README.md index d47800f..52dcf09 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,15 @@ One row per session of the host in the custom table `host_session` (schema 0.0.1). The hub adds the gateway as `spoke_id`. -The first run after a boot opens a row for the current session (start time equal to end time, everything else 0) and closes the previous one from the history of the host's uptime signal: CPU, RAM and disk at one minute, with coverage as the share of minutes with a CPU sample. Later runs in the same boot stop after one query. +The first run after a boot closes the previous session from the history of the host's uptime signal (CPU, RAM and disk at one minute, with coverage as the share of minutes with a CPU sample; an empty window is written as 0) and opens a row for the current session (start time equal to end time, everything else 0). Later runs in the same boot stop after one query and print nothing. -One script for every gateway. `signals.yml` lists the cpu, ram, disk and uptime signals of every host; the script asks the history of each `-uptime` until one answers 200 (a gateway answers 403 for signals it does not own) and uses that host's four signals. A new host needs its four signals added to `signals.yml` and a new tag. Runs every minute; needs only `requests` in the image. \ No newline at end of file +One script for every gateway. `signals.yml` lists the cpu, ram, disk and uptime signals of every host; the script asks the history of each `-uptime` until one answers 200 (a gateway answers 403 for signals it does not own) and uses that host's four signals. A new host needs its four signals added to `signals.yml` and a new tag. Runs every minute; needs only `requests` in the image. + +## Output and failures + +Every event is one JSON line with `level`, `msg` and its fields; the orchestrator logs them under `i7orchestrator/scheduler/job`. + +- A refusal of the previous session by the database is logged once as an ERROR and the current session is still opened, so it is not repeated. +- Exit code 2: the run cannot succeed until the script or the schema changes (a missing or empty `signals.yml`, a gateway whose uptime signal is not listed, the database refusing the current session). +- Exit code 1: anything else, such as the sidecar or the database being unavailable; the next run tries again. +- `maxConsecutiveFailures: 5` in `schedule.yaml`: the orchestrator stops the job after five failed runs in a row and shows why in the job report. A new tag starts it again. \ No newline at end of file diff --git a/main.py b/main.py index d82490a..32459ee 100644 --- a/main.py +++ b/main.py @@ -1,5 +1,7 @@ +import json import os import re +import sys import time import requests @@ -12,6 +14,22 @@ 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): @@ -23,9 +41,15 @@ def read_history(signals, start_ms, end_ms, resolution): 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")) + 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 -uptime signal") + return hosts def owns(host, now_ms): @@ -40,7 +64,7 @@ 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") + raise ConfigError("no uptime signal of this gateway is listed in signals.yml") return host @@ -62,26 +86,51 @@ def current_boot_ms(): 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" - 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) + 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() - body = response.json() - if body.get("statusCode", 200) != 200: - raise RuntimeError(f"db-controller {body.get('statusCode')}: {body.get('errorMessage')}") - return body.get("data") + return read_db_answer(response.json()) def stored_rows(): @@ -102,7 +151,7 @@ def open_row(start_ms): 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 0, 0 return sum(values) / len(values), max(values) @@ -122,18 +171,37 @@ def build_row(host, 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) +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) - 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) + 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(): @@ -141,9 +209,21 @@ def main(): 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) + 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__": - main() \ No newline at end of file + sys.exit(run()) \ No newline at end of file diff --git a/schedule.yaml b/schedule.yaml index 4b4f0a9..fd2d173 100644 --- a/schedule.yaml +++ b/schedule.yaml @@ -4,9 +4,9 @@ schedule: "* * * * *" action: run runOnStartup: true timeoutSeconds: 120 +maxConsecutiveFailures: 5 imageName: docker.izazpi.cloud/infra/infra-python:0.1.12 command: ["python", "-u", "/app/main.py"] maxCpuCores: 0.25 maxMemoryMb: 128 -autoremove: false -logOption: 1 \ No newline at end of file +autoremove: false \ No newline at end of file