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: its start time, no end time, everything else 0. In the same run it closes the previous session: it finds that session in the history of the uptime signal (63) before the boot (last sample is the end, end minus that sample's uptime is the start), reads CPU, RAM and disk at one minute, and fills that session's open row with the end time, duration, means, maxima and coverage (share of minutes with a CPU sample). A previous session with no row, such as the one running when the job was first installed, gets a new full row. Every later run in the same boot sees the current row and stops 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.
 
-Runs every minute. The current boot comes from `/proc/uptime`. Signal ids are deb13-kde's in the home core. 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,4 +1,7 @@
+import json
 import os
+import re
+import sys
 import time
 
 import requests
@@ -6,11 +9,28 @@ 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
+SIGNALS_FILE = "/app/signals.yml"
 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):
@@ -18,7 +38,35 @@ def read_history(signals, start_ms, end_ms, resolution):
     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"]}
+    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):
@@ -39,25 +87,51 @@ def current_boot_ms():
     return int(time.time() * 1000 - uptime_s * 1000)
 
 
-def previous_session(boot_ms):
+def belongs_to_boot(time_ms, uptime_s, boot_ms):
-    hours = last_samples(UPTIME, boot_ms - LOOKBACK_MS, boot_ms, "1hr")
+    return abs(time_ms - uptime_s * 1000 - boot_ms) < MATCH_MS
-    if not hours:
+
-        return None
+
-    minutes = last_samples(UPTIME, hours[-1][0], min(hours[-1][0] + HOUR_MS, boot_ms), "1min")
+def recent_hours(uptime, boot_ms):
-    if not minutes:
+    hours = last_samples(uptime, boot_ms - LOOKBACK_MS, boot_ms, "1hr")
-        return None
+    return [time_ms for time_ms, _ in reversed(hours[-2:])]
-    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 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()
-    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():
@@ -69,7 +143,7 @@ def find_row(rows, start_ms):
 
 
 def open_row(start_ms):
-    row = {"start_time": start_ms, "end_time": None, "duration_s": 0, "coverage": 0}
+    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
@@ -78,37 +152,61 @@ 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 build_row(start_ms, end_ms):
+def pick_resolution(duration_ms):
-    history = read_history([CPU, RAM, DISK], start_ms, end_ms, "1min")
+    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, 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)):
+    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)
+    row["coverage"] = coverage(history[cpu], start_ms, end_ms)
     return row
 
 
-def close_previous(rows, boot_ms):
+def store_previous_session(host, rows, session):
-    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 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:
-        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.get("end_time") is None:
+        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():
@@ -116,9 +214,21 @@ def main():
     rows = stored_rows()
     if find_row(rows, boot_ms) is not None:
         return
-    close_previous(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

signals.yml

@@ -1,4 +1,28 @@
-- { id: 21, name: 'deb13kde-cpu',    read: 1, write: 0 }
+- { id: 21, name: 'deb13kde-cpu', read: 1, write: 0 }
-- { id: 22, name: 'deb13kde-ram',    read: 1, write: 0 }
+- { id: 22, name: 'deb13kde-ram', read: 1, write: 0 }
-- { id: 23, name: 'deb13kde-disk',   read: 1, write: 0 }
+- { id: 23, name: 'deb13kde-disk', read: 1, write: 0 }
 - { id: 63, name: 'deb13kde-uptime', read: 1, write: 0 }
+- { id: 1, name: 'rbx-01-cpu', read: 1, write: 0 }
+- { id: 2, name: 'rbx-01-ram', read: 1, write: 0 }
+- { id: 3, name: 'rbx-01-disk', read: 1, write: 0 }
+- { id: 64, name: 'rbx-01-uptime', read: 1, write: 0 }
+- { id: 6, name: 'fr-01-cpu', read: 1, write: 0 }
+- { id: 7, name: 'fr-01-ram', read: 1, write: 0 }
+- { id: 8, name: 'fr-01-disk', read: 1, write: 0 }
+- { id: 65, name: 'fr-01-uptime', read: 1, write: 0 }
+- { id: 11, name: 'sb-01-cpu', read: 1, write: 0 }
+- { id: 12, name: 'sb-01-ram', read: 1, write: 0 }
+- { id: 13, name: 'sb-01-disk', read: 1, write: 0 }
+- { id: 66, name: 'sb-01-uptime', read: 1, write: 0 }
+- { id: 16, name: 'grab-01-cpu', read: 1, write: 0 }
+- { id: 17, name: 'grab-01-ram', read: 1, write: 0 }
+- { id: 18, name: 'grab-01-disk', read: 1, write: 0 }
+- { id: 67, name: 'grab-01-uptime', read: 1, write: 0 }
+- { id: 26, name: 'pi102-cpu', read: 1, write: 0 }
+- { id: 27, name: 'pi102-ram', read: 1, write: 0 }
+- { id: 28, name: 'pi102-disk', read: 1, write: 0 }
+- { id: 68, name: 'pi102-uptime', read: 1, write: 0 }
+- { id: 32, name: 'pi103-cpu', read: 1, write: 0 }
+- { id: 33, name: 'pi103-ram', read: 1, write: 0 }
+- { id: 34, name: 'pi103-disk', read: 1, write: 0 }
+- { id: 69, name: 'pi103-uptime', read: 1, write: 0 }