change resolution

README.md

@@ -1,7 +1,16 @@
-# host-session
+# host_session
 
-Stores one row per finished session of the host in the custom table `host_session` (schema 0.0.1): start and end time, duration, mean and max of CPU, RAM and disk percentage, and coverage (share of the session's minutes with a CPU sample). The hub adds the gateway as `spoke_id`.
+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`.
 
-Runs every minute. The current boot comes from `/proc/uptime`. The previous session is found in the history of the uptime signal (63) before that boot: its last sample is the end, end minus that sample's uptime is the start. A session already stored (start within two minutes) is skipped, so repeated runs write nothing twice. The first row appears after the first reboot that follows the uptime signal's creation (2026-11-06 18:38).
+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.
 
-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,65 +87,148 @@ 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 already_stored(start_ms):
+def stored_rows():
-    rows = db_call("POST", f"/api/db-query/{TABLE}/", {"filters": [], "orderBy": ["start_time"], "limit": 1000}) or []
+    return db_call("POST", f"/api/db-query/{TABLE}/", {"filters": [], "orderBy": ["start_time"], "limit": 1000}) or []
-    return any(abs(row["start_time"] - start_ms) < MATCH_MS for row in rows)
+
+
+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 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 main():
+def store_previous_session(host, rows, session):
-    session = previous_session(current_boot_ms())
+    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:
-        print("no previous session in the uptime history", flush=True)
+        log("INFO", "no previous session in the uptime history")
         return
-    if already_stored(session[0]):
+    if not is_distinct_session(session, boot_ms):
-        print("previous session already stored", session[0], flush=True)
+        log("WARN", "previous session ignored, it is the current boot", start=session[0], end=session[1], boot=boot_ms)
         return
-    row = build_row(*session)
+    try:
-    print("stored", db_call("POST", f"/api/db/{TABLE}/", row), row, flush=True)
+        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__":
-    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 }