change resolution

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 `<host>-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.
+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 `<host>-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.

main.py

@@ -1,5 +1,7 @@
+import json
 import os
 import re
+import sys
 import time
 
 import requests
@@ -12,6 +14,23 @@ MINUTE_MS = 60000
 HOUR_MS = 60 * MINUTE_MS
 LOOKBACK_MS = 60 * 24 * HOUR_MS
 MATCH_MS = 2 * MINUTE_MS
+MINUTE_RESOLUTION_LIMIT_MS = 7 * 24 * HOUR_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 +42,15 @@ def read_history(signals, start_ms, end_ms, resolution):
 
 
 def signal_hosts():
-    with open(SIGNALS_FILE) as file:
+    try:
-        names = re.findall(r"name:\s*'([^']+)'", file.read())
+        with open(SIGNALS_FILE) as file:
-    return sorted(name[: -len("-uptime")] for name in names if name.endswith("-uptime"))
+            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):
@@ -40,7 +65,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 +87,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")
+    for hour_ms in recent_hours(uptime, boot_ms):
-    if not hours:
+        last = last_sample_of_an_earlier_boot(uptime, hour_ms, boot_ms)
-        return None
+        if last is not None:
-    minutes = last_samples(uptime, hours[-1][0], min(hours[-1][0] + HOUR_MS, boot_ms), "1min")
+            end_ms, uptime_s = last
-    if not minutes:
+            return int(end_ms - uptime_s * 1000), int(end_ms)
-        return None
+    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 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()
+    return read_db_answer(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():
@@ -102,19 +152,23 @@ 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)
 
 
 def coverage(series, start_ms, end_ms):
-    total = max(1, (end_ms - start_ms) // MINUTE_MS)
+    total = max(1, (end_ms - start_ms) // series["stepMs"])
     present = sum(1 for t, _ in samples(series) if start_ms <= t < end_ms)
     return min(1.0, present / total)
 
 
+def pick_resolution(duration_ms):
+    return "1min" if duration_ms <= MINUTE_RESOLUTION_LIMIT_MS else "1hr"
+
+
 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")
+    history = read_history([cpu, ram, disk], start_ms, end_ms, pick_resolution(end_ms - start_ms))
     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)
@@ -122,18 +176,37 @@ def build_row(host, start_ms, end_ms):
     return row
 
 
-def close_previous(host, rows, boot_ms):
+def store_previous_session(host, rows, session):
-    session = previous_session(host, boot_ms)
+    stored = find_row(rows, session[0])
-    if session is None:
+    if stored is not None and stored["end_time"] != stored["start_time"]:
-        print("no previous session in the uptime history", flush=True)
+        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)
+        created = db_call("POST", f"/api/db/{TABLE}/", row)
-    elif stored["end_time"] == stored["start_time"]:
+        log("INFO", "previous session stored", id=created["id"], start=session[0], end=session[1], coverage=row["coverage"])
-        db_call("PUT", f"/api/db/{TABLE}/{stored['id']}/", row)
+        return
-        print("previous session closed", stored["id"], row, flush=True)
+    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 +214,21 @@ def main():
     rows = stored_rows()
     if find_row(rows, boot_ms) is not None:
         return
-    close_previous(own_host(), rows, boot_ms)
+    close_previous_session(own_host(), rows, boot_ms)
-    print("current session opened", db_call("POST", f"/api/db/{TABLE}/", open_row(boot_ms)), flush=True)
+    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()
+    sys.exit(run())

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
+autoremove: false
-logOption: 1