|
|
| """o7infra_agent - Oracle7-infrastructure: Oracle7 pointed at a system (Ben 2026-10-12: "it gets pointed at anything and it can
|
| become the sensors and run it intelligently"; "we want to be infrastructure"). One file, Python standard library only (3.10+).
|
|
|
| DISCOVER with no configuration: systemd services (long-running) and jobs (oneshot / timer-run), nginx sites and the TLS
|
| certificates they serve, listening ports, containers (if docker is present)
|
| SENSE host (load, cpu, memory, swap, processes), every real disk (space, inodes), network, every service (state,
|
| memory, cpu, RESTART RATE), every job (last run ok), journal errors per unit, certificate DAYS LEFT, site
|
| reachability and latency; plus any source in /etc/o7infra/sources.json (http_json / prometheus / command).
|
| Every value is a FACT (time, entity, signal, value, source) - the receipt says where it was read.
|
| STORE SQLite with a HARD SIZE CAP: raw 24 h, 10-minute means 30 days, pruned every cycle.
|
| READ readers generated for every signal: current, trend, time-to-threshold, anomaly vs its own history, what changed
|
| before an event. `ask` answers plain questions from them.
|
| THE GATE = THE FUTURE. Forecasts are logged when made and scored when their time arrives; a method is ADMITTED for a kind
|
| of signal only after >= 30 scored forecasts beat "stays the same" by >= 10%.
|
| ACT actions are CHECKED OPERATIONS: precondition, predicted effect, dry run. Ladder: recommend -> dry-run -> a human
|
| second key (`approve <id>`) -> autonomous ONLY for an action class whose observed effect matched its prediction
|
| 5 times, and only when O7INFRA_AUTONOMOUS=1. Default: recommend + dry-run, nothing changes the system.
|
|
|
| usage: o7infra_agent.py run | brief | status | forecast | gate | why <entity> | signals [f] | discover | plan
|
| | approve <id> | execute <id> | ask "<question>" """
|
| import glob
|
| import json
|
| import os
|
| import re
|
| import socket
|
| import sqlite3
|
| import ssl
|
| import subprocess
|
| import sys
|
| import time
|
| import urllib.request
|
|
|
| HOME = os.environ.get("O7INFRA_HOME", "/var/lib/o7infra")
|
| DB = os.path.join(HOME, "facts.sqlite3")
|
| SOURCES = os.environ.get("O7INFRA_SOURCES", "/etc/o7infra/sources.json")
|
| STEP = int(os.environ.get("O7INFRA_STEP", "60"))
|
| CAP_MB = float(os.environ.get("O7INFRA_CAP_MB", "60"))
|
| AUTONOMOUS = os.environ.get("O7INFRA_AUTONOMOUS") == "1"
|
| RAW_KEEP, AGG_KEEP = 24 * 3600, 30 * 86400
|
| FORECAST_EVERY, SLOW_EVERY, DISCOVER_EVERY = 600, 300, 3600
|
| HORIZONS = (3600, 6 * 3600)
|
| REAL_FS = {"ext2", "ext3", "ext4", "xfs", "btrfs", "zfs", "f2fs", "vfat", "ntfs"}
|
|
|
| THRESH = {"used_pct": (90.0, 97.0, "high"), "swap_used_pct": (80.0, 95.0, "high"), "mem_used_pct": (90.0, 97.0, "high"),
|
| "inode_pct": (85.0, 95.0, "high"), "cert_days_left": (14.0, 5.0, "low"), "restarts_per_h": (6.0, 30.0, "high"),
|
| "up": (0.5, 0.5, "low"), "last_run_ok": (0.5, 0.5, "low")}
|
|
|
|
|
|
|
| def db():
|
| os.makedirs(HOME, exist_ok=True)
|
| c = sqlite3.connect(DB, timeout=30)
|
| c.executescript("""
|
| PRAGMA journal_mode=WAL;
|
| CREATE TABLE IF NOT EXISTS fact (ts INTEGER, entity TEXT, signal TEXT, value REAL, source TEXT);
|
| CREATE INDEX IF NOT EXISTS fact_es ON fact(entity, signal, ts);
|
| CREATE INDEX IF NOT EXISTS fact_ts ON fact(ts);
|
| CREATE TABLE IF NOT EXISTS agg (ts INTEGER, entity TEXT, signal TEXT, value REAL);
|
| CREATE INDEX IF NOT EXISTS agg_es ON agg(entity, signal, ts);
|
| CREATE TABLE IF NOT EXISTS event (ts INTEGER, entity TEXT, kind TEXT, detail TEXT);
|
| CREATE TABLE IF NOT EXISTS forecast (made INTEGER, target INTEGER, entity TEXT, signal TEXT, method TEXT, value REAL,
|
| scored INTEGER DEFAULT 0, actual REAL);
|
| CREATE INDEX IF NOT EXISTS forecast_t ON forecast(scored, target);
|
| CREATE TABLE IF NOT EXISTS action (id INTEGER PRIMARY KEY, made INTEGER, entity TEXT, kind TEXT, command TEXT,
|
| precondition TEXT, predicted TEXT, dry_run TEXT, status TEXT, approved INTEGER,
|
| executed INTEGER, executed_at INTEGER, observed TEXT, matched INTEGER);
|
| CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT);
|
| """)
|
| return c
|
|
|
|
|
| def meta_get(c, k, d=None):
|
| r = c.execute("SELECT v FROM meta WHERE k=?", (k,)).fetchone()
|
| return json.loads(r[0]) if r else d
|
|
|
|
|
| def meta_set(c, k, v):
|
| c.execute("INSERT OR REPLACE INTO meta VALUES (?,?)", (k, json.dumps(v)))
|
|
|
|
|
| def _read(p):
|
| try:
|
| with open(p) as f:
|
| return f.read()
|
| except OSError:
|
| return ""
|
|
|
|
|
| def _run(cmd, timeout=20):
|
| try:
|
| return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout
|
| except (OSError, subprocess.TimeoutExpired):
|
| return ""
|
|
|
|
|
|
|
| def discover():
|
| """what is on this system, found with no configuration"""
|
| d = {"services": [], "jobs": [], "sites": [], "certs": {}, "ports": [], "containers": []}
|
| names = [l.split()[0] for l in _run(["systemctl", "list-units", "--type=service", "--all", "--no-legend", "--plain"]).splitlines()
|
| if l.split() and l.split()[0].endswith(".service")]
|
| timers = {l.split()[-2] if len(l.split()) >= 2 else "" for l in _run(["systemctl", "list-timers", "--all", "--no-legend", "--plain"]).splitlines()}
|
| if names:
|
| txt = _run(["systemctl", "show", "-p", "Id,Type,ActiveState,LoadState,TriggeredBy"] + names, timeout=40)
|
| for block in txt.strip().split("\n\n"):
|
| kv = dict(l.split("=", 1) for l in block.splitlines() if "=" in l)
|
| if kv.get("LoadState") != "loaded" or kv.get("ActiveState") not in ("active", "failed", "activating", "deactivating"):
|
| continue
|
| uid = kv.get("Id", "")
|
| if uid.startswith(("user@", "user-runtime-dir@")):
|
| continue
|
| job = kv.get("Type") == "oneshot" or ".timer" in kv.get("TriggeredBy", "") or uid in timers
|
| (d["jobs"] if job else d["services"]).append(uid[:-8])
|
| for conf in glob.glob("/etc/nginx/sites-enabled/*") + glob.glob("/etc/nginx/conf.d/*.conf"):
|
| txt = _read(conf)
|
| for block in re.split(r"\bserver\s*\{", txt)[1:]:
|
| names_ = re.findall(r"\bserver_name\s+([^;]+);", block)
|
| cert = re.findall(r"\bssl_certificate\s+([^;]+);", block)
|
| for n in (names_[0].split() if names_ else []):
|
| if n not in ("_", "localhost") and "." in n and not n.startswith("*") and not re.fullmatch(r"[\d.]+", n):
|
| if n not in d["sites"]:
|
| d["sites"].append(n)
|
| if cert:
|
| d["certs"].setdefault(cert[0].strip(), []).append(n)
|
| ss = _run(["ss", "-ltnH"])
|
| for line in ss.splitlines():
|
| p = line.split()
|
| if len(p) >= 4:
|
| port = p[3].rsplit(":", 1)[-1]
|
| if port.isdigit() and port not in d["ports"]:
|
| d["ports"].append(port)
|
| if os.path.exists("/usr/bin/docker"):
|
| d["containers"] = [l.strip() for l in _run(["docker", "ps", "--format", "{{.Names}}"]).splitlines() if l.strip()]
|
| return d
|
|
|
|
|
|
|
| def sense_host(prev):
|
| out = []
|
| la = _read("/proc/loadavg").split()
|
| if la:
|
| out += [("host", "load1", float(la[0]), "/proc/loadavg"), ("host", "load5", float(la[1]), "/proc/loadavg")]
|
| st = _read("/proc/stat").splitlines()
|
| if st:
|
| f = [int(x) for x in st[0].split()[1:]]
|
| idle, total = f[3] + (f[4] if len(f) > 4 else 0), sum(f)
|
| if "cpu" in prev and total > prev["cpu"][1]:
|
| out.append(("host", "cpu_pct", round(100.0 * (1 - (idle - prev["cpu"][0]) / (total - prev["cpu"][1])), 2), "/proc/stat"))
|
| prev["cpu"] = (idle, total)
|
| mi = {}
|
| for line in _read("/proc/meminfo").splitlines():
|
| k, _, v = line.partition(":")
|
| mi[k] = int(v.split()[0]) if v.split() else 0
|
| if mi.get("MemTotal"):
|
| avail = mi.get("MemAvailable", 0)
|
| out += [("host", "mem_used_pct", round(100.0 * (1 - avail / mi["MemTotal"]), 2), "/proc/meminfo"),
|
| ("host", "mem_avail_mb", round(avail / 1024, 1), "/proc/meminfo")]
|
| if mi.get("SwapTotal"):
|
| out.append(("host", "swap_used_pct", round(100.0 * (1 - mi.get("SwapFree", 0) / mi["SwapTotal"]), 2), "/proc/meminfo"))
|
| out.append(("host", "procs", float(len([d for d in os.listdir("/proc") if d.isdigit()])), "/proc"))
|
| return out
|
|
|
|
|
| def sense_disks():
|
| out, seen = [], set()
|
| for line in _read("/proc/mounts").splitlines():
|
| p = line.split()
|
| if len(p) < 3 or p[2] not in REAL_FS or p[0] in seen:
|
| continue
|
| seen.add(p[0])
|
| try:
|
| s = os.statvfs(p[1])
|
| except OSError:
|
| continue
|
| if not s.f_blocks:
|
| continue
|
| ent = "disk:" + p[1]
|
| out += [(ent, "used_pct", round(100.0 * (1 - s.f_bavail / s.f_blocks), 2), "statvfs " + p[1]),
|
| (ent, "free_gb", round(s.f_bavail * s.f_frsize / 1e9, 3), "statvfs " + p[1])]
|
| if s.f_files:
|
| out.append((ent, "inode_pct", round(100.0 * (1 - s.f_favail / s.f_files), 2), "statvfs " + p[1]))
|
| return out
|
|
|
|
|
| def sense_net(prev, dt):
|
| out = []
|
| for line in _read("/proc/net/dev").splitlines()[2:]:
|
| name, _, rest = line.partition(":")
|
| name = name.strip()
|
| if name == "lo" or not rest.split():
|
| continue
|
| f = rest.split()
|
| rx, tx = int(f[0]), int(f[8])
|
| key = "net:" + name
|
| if key in prev and dt > 0:
|
| out += [(key, "rx_kbps", round((rx - prev[key][0]) / dt / 1024, 2), "/proc/net/dev"),
|
| (key, "tx_kbps", round((tx - prev[key][1]) / dt / 1024, 2), "/proc/net/dev")]
|
| prev[key] = (rx, tx)
|
| return out
|
|
|
|
|
| def sense_units(prev, dt, disc):
|
| out, events = [], []
|
| jobs = set(disc.get("jobs", []))
|
| names = [n + ".service" for n in disc.get("services", []) + disc.get("jobs", [])]
|
| if not names:
|
| return out, events
|
| txt = _run(["systemctl", "show", "-p", "Id,ActiveState,NRestarts,MemoryCurrent,CPUUsageNSec,Result"] + names, timeout=40)
|
| now = time.time()
|
| for block in txt.strip().split("\n\n"):
|
| kv = dict(l.split("=", 1) for l in block.splitlines() if "=" in l)
|
| uid = kv.get("Id", "")[:-8]
|
| if not uid:
|
| continue
|
| ent = ("job:" if uid in jobs else "unit:") + uid
|
| state = kv.get("ActiveState", "")
|
| if uid in jobs:
|
| ok = 0.0 if (state == "failed" or kv.get("Result", "success") not in ("success", "")) else 1.0
|
| out.append((ent, "last_run_ok", ok, "systemctl show (Result=%s)" % kv.get("Result", "")))
|
| else:
|
| out.append((ent, "up", 1.0 if state == "active" else 0.0, "systemctl show (ActiveState=%s)" % state))
|
| last = prev.get("state:" + ent)
|
| if last and last != state:
|
| events.append((ent, "state", "%s -> %s" % (last, state)))
|
| prev["state:" + ent] = state
|
| try:
|
| r = int(kv.get("NRestarts", "0"))
|
| hist = prev.setdefault("rh:" + ent, [])
|
| hist.append((now, r))
|
| while hist and hist[0][0] < now - 3600:
|
| hist.pop(0)
|
| if len(hist) >= 2 and hist[-1][0] > hist[0][0]:
|
| rate = (hist[-1][1] - hist[0][1]) / (hist[-1][0] - hist[0][0]) * 3600
|
| out.append((ent, "restarts_per_h", round(rate, 2), "systemctl show NRestarts=%d" % r))
|
| if "restarts:" + ent in prev and r > prev["restarts:" + ent] and r - prev["restarts:" + ent] < 1000:
|
| events.append((ent, "restart", "NRestarts %d -> %d" % (prev["restarts:" + ent], r)))
|
| prev["restarts:" + ent] = r
|
| except ValueError:
|
| pass
|
| try:
|
| mem = int(kv.get("MemoryCurrent", ""))
|
| if mem < 1 << 60:
|
| out.append((ent, "mem_mb", round(mem / 1048576, 1), "systemctl show"))
|
| except ValueError:
|
| pass
|
| try:
|
| cpu = int(kv.get("CPUUsageNSec", ""))
|
| if cpu < 1 << 62 and ("cpu:" + ent) in prev and dt > 0:
|
| out.append((ent, "cpu_pct", round(100.0 * (cpu - prev["cpu:" + ent]) / 1e9 / dt, 2), "systemctl show"))
|
| if cpu < 1 << 62:
|
| prev["cpu:" + ent] = cpu
|
| except ValueError:
|
| pass
|
| return out, events
|
|
|
|
|
| def sense_journal(since):
|
| counts = {}
|
| for line in _run(["journalctl", "--since", "@%d" % since, "-p", "3", "-o", "json", "--no-pager", "-q"], timeout=30).splitlines():
|
| try:
|
| j = json.loads(line)
|
| except ValueError:
|
| continue
|
| u = j.get("_SYSTEMD_UNIT") or j.get("SYSLOG_IDENTIFIER") or "kernel"
|
| counts[u] = counts.get(u, 0) + 1
|
| out = [("unit:" + (u[:-8] if u.endswith(".service") else u), "errors", float(n), "journalctl -p 3") for u, n in counts.items()]
|
| out.append(("host", "errors", float(sum(counts.values())), "journalctl -p 3"))
|
| return out
|
|
|
|
|
| def sense_certs(disc):
|
| out = []
|
| now = time.time()
|
| for path, sites in disc.get("certs", {}).items():
|
| txt = _run(["openssl", "x509", "-enddate", "-noout", "-in", path])
|
| m = re.search(r"notAfter=(.+)", txt)
|
| if not m:
|
| continue
|
| try:
|
| end = time.mktime(time.strptime(m.group(1).strip().replace(" GMT", ""), "%b %d %H:%M:%S %Y")) - time.timezone
|
| except ValueError:
|
| continue
|
| out.append(("cert:" + sites[0], "cert_days_left", round((end - now) / 86400, 2), "openssl x509 %s" % path))
|
| return out
|
|
|
|
|
| def sense_sites(disc):
|
| out = []
|
| ctx = ssl.create_default_context()
|
| for name in disc.get("sites", [])[:60]:
|
| t0 = time.time()
|
| code = 0
|
| for method in ("HEAD", "GET"):
|
| try:
|
| req = urllib.request.Request("https://%s/" % name, method=method, headers={"User-Agent": "o7infra/2"})
|
| with urllib.request.urlopen(req, timeout=8, context=ctx) as r:
|
| code = r.status
|
| except urllib.error.HTTPError as e:
|
| code = e.code
|
| except Exception:
|
| code = 0
|
| if code not in (405, 501):
|
| break
|
| ms = (time.time() - t0) * 1000
|
| up = 1.0 if 0 < code < 500 else 0.0
|
| out += [("site:" + name, "up", up, "HTTPS HEAD status %d" % code), ("site:" + name, "latency_ms", round(ms, 1), "HTTPS HEAD")]
|
| return out
|
|
|
|
|
| def sense_sources():
|
| """any system: /etc/o7infra/sources.json = [{"type": "http_json", "url", "entity", "signals": {"name": "a.b.c"}},
|
| {"type": "prometheus", "url", "entity", "match": "regex"}, {"type": "command", "cmd": [...], "entity", "regex": "name=(value)"}]"""
|
| out = []
|
| try:
|
| srcs = json.load(open(SOURCES))
|
| except (OSError, ValueError):
|
| return out
|
| for s in srcs:
|
| ent = s.get("entity", "source")
|
| try:
|
| if s["type"] == "http_json":
|
| d = json.loads(urllib.request.urlopen(s["url"], timeout=8).read())
|
| for name, path in s.get("signals", {}).items():
|
| v = d
|
| for k in path.split("."):
|
| v = v[int(k)] if isinstance(v, list) else v[k]
|
| out.append((ent, name, float(v), "GET %s %s" % (s["url"], path)))
|
| elif s["type"] == "prometheus":
|
| rx = re.compile(s.get("match", ".*"))
|
| for line in urllib.request.urlopen(s["url"], timeout=8).read().decode().splitlines():
|
| if line.startswith("#") or not line.strip():
|
| continue
|
| name, _, val = line.rpartition(" ")
|
| if rx.search(name):
|
| out.append((ent, name[:120], float(val), "GET %s" % s["url"]))
|
| elif s["type"] == "command":
|
| txt = _run(s["cmd"], timeout=20)
|
| for name, val in re.findall(s.get("regex", r"(\w+)=([-\d.]+)"), txt):
|
| out.append((ent, name, float(val), "command %s" % " ".join(s["cmd"])[:80]))
|
| except Exception:
|
| out.append((ent, "source_ok", 0.0, "source %s" % s.get("url", s.get("cmd", "?"))))
|
| return out
|
|
|
|
|
|
|
| def series(c, entity, signal, since):
|
| rows = c.execute("SELECT ts, value FROM fact WHERE entity=? AND signal=? AND ts>=? ORDER BY ts", (entity, signal, since)).fetchall()
|
| if not rows:
|
| rows = c.execute("SELECT ts, value FROM agg WHERE entity=? AND signal=? AND ts>=? ORDER BY ts", (entity, signal, since)).fetchall()
|
| return rows
|
|
|
|
|
| def slope(rows):
|
| if len(rows) < 6:
|
| return None
|
| pts = rows[:: max(1, len(rows) // 40)]
|
| ss = sorted((b[1] - a[1]) / (b[0] - a[0]) for i, a in enumerate(pts) for b in pts[i + 1:] if b[0] > a[0])
|
| return ss[len(ss) // 2] if ss else None
|
|
|
|
|
| def thresholds(signal):
|
| for k in sorted(THRESH, key=len, reverse=True):
|
| if signal == k or signal.endswith(k):
|
| return THRESH[k]
|
| return None
|
|
|
|
|
| def level(signal, v):
|
| th = thresholds(signal)
|
| if not th:
|
| return None
|
| w, cr, d = th
|
| if d == "high":
|
| return "critical" if v >= cr else ("warning" if v >= w else None)
|
| return "critical" if v < cr or (signal in ("up", "last_run_ok") and v < 0.5) else ("warning" if v < w else None)
|
|
|
|
|
|
|
| def make_forecasts(c, now):
|
| for ent, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=? AND (signal LIKE '%used_pct' OR signal LIKE '%inode_pct' "
|
| "OR signal IN ('mem_mb','mem_avail_mb','free_gb','cert_days_left'))", (now - 3600,)).fetchall():
|
| rows = series(c, ent, s, now - 6 * 3600)
|
| if len(rows) < 6:
|
| continue
|
| last = rows[-1][1]
|
|
|
| sl = slope(rows) if rows[-1][0] - rows[0][0] >= 3 * 3600 else None
|
| for h in HORIZONS:
|
| c.execute("INSERT INTO forecast (made, target, entity, signal, method, value) VALUES (?,?,?,?,?,?)", (now, now + h, ent, s, "naive", last))
|
| if sl is not None:
|
| for m, k in METHODS.items():
|
| c.execute("INSERT INTO forecast (made, target, entity, signal, method, value) VALUES (?,?,?,?,?,?)",
|
| (now, now + h, ent, s, m, last + k * sl * h))
|
|
|
|
|
| def score_forecasts(c, now):
|
| for rid, target, ent, s in c.execute("SELECT rowid, target, entity, signal FROM forecast WHERE scored=0 AND target<=?", (now,)).fetchall():
|
| r = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts BETWEEN ? AND ? ORDER BY ABS(ts-?) LIMIT 1",
|
| (ent, s, target - STEP * 3, target + STEP * 3, target)).fetchone()
|
| if r:
|
| c.execute("UPDATE forecast SET scored=1, actual=? WHERE rowid=?", (r[0], rid))
|
| elif target < now - 3600:
|
| c.execute("UPDATE forecast SET scored=2 WHERE rowid=?", (rid,))
|
|
|
|
|
| METHODS = {"trend6h": 1.0, "damped": 0.5}
|
|
|
|
|
| def kind_of(signal):
|
| return signal
|
|
|
|
|
| def gate(c):
|
| acc = {}
|
| for s, h, m, e in c.execute("SELECT signal, target-made, method, ABS(value-actual) FROM forecast WHERE scored=1").fetchall():
|
| acc.setdefault((kind_of(s), int(h)), {}).setdefault(m, []).append(e)
|
| out = {}
|
| for k, ms in acc.items():
|
| nv = ms.get("naive", [])
|
| if not nv:
|
| continue
|
| mn = sum(nv) / len(nv)
|
| rec = {"n_naive": len(nv), "mae_naive": round(mn, 4), "methods": {}, "admitted": None}
|
| for m in METHODS:
|
| tr = ms.get(m, [])
|
| if tr:
|
| mt = sum(tr) / len(tr)
|
| ok = len(tr) >= 30 and mt <= 0.9 * mn
|
| rec["methods"][m] = {"n": len(tr), "mae": round(mt, 4), "admitted": ok}
|
| if ok and (rec["admitted"] is None or mt < rec["methods"][rec["admitted"]]["mae"]):
|
| rec["admitted"] = m
|
| out["%s@%dh" % (k[0], k[1] // 3600)] = rec
|
| return out
|
|
|
|
|
|
|
| SAFE_DISK = [
|
| ("journal_vacuum", ["journalctl", "--vacuum-size=200M"], ["journalctl", "--disk-usage"]),
|
| ("apt_clean", ["apt-get", "clean"], ["du", "-sm", "/var/cache/apt/archives"]),
|
| ]
|
|
|
|
|
| def plan_actions(c, now):
|
| """CHECKED OPERATIONS proposed from the current facts; dry-run results attached; nothing executes here"""
|
| t = c.execute("SELECT MAX(ts) FROM fact").fetchone()[0]
|
| if not t:
|
| return []
|
| _, rows = latest_values(c)
|
| rows = [(e, s, v) for e, s, v, _, _ in rows]
|
| loops = {e: v for e, s, v in rows if s == "restarts_per_h"}
|
| existing = {r[0] for r in c.execute("SELECT entity FROM action WHERE status IN ('proposed','approved')").fetchall()}
|
| proposals = []
|
| for e, s, v in rows:
|
| lv = level(s, v)
|
| if not lv or e in existing:
|
| continue
|
| if e.startswith("disk:") and s == "used_pct":
|
| for kind, cmd, dry in SAFE_DISK:
|
| out = _run(dry, timeout=30).strip().splitlines()
|
| proposals.append((e, kind, " ".join(cmd), "%s used %.1f%% (%s)" % (e, v, lv),
|
| "free_gb rises by the reclaimable amount; used_pct falls", out[-1][:160] if out else "no output"))
|
| elif e.startswith("unit:") and s == "restarts_per_h" and lv == "critical":
|
| u = e[5:]
|
| tail = _run(["journalctl", "-u", u, "-n", "3", "--no-pager", "-q", "-o", "cat"], timeout=15).strip().replace("\n", " | ")
|
| proposals.append((e, "investigate_crash_loop", "systemctl status %s; journalctl -u %s -n 50" % (u, u),
|
| "%s restarting %.0f times per hour" % (u, v), "read-only: shows the failing cause", tail[:200]))
|
| elif e.startswith("unit:") and s == "up" and v == 0 and loops.get(e, 0) < 6:
|
| u = e[5:]
|
| proposals.append((e, "restart_service", "systemctl restart %s" % u, "%s is not active" % u,
|
| "up becomes 1 within 60 s and restarts_per_h stays < 6", "dry run: unit file %s" %
|
| ("present" if _run(["systemctl", "cat", u + ".service"], timeout=10) else "missing")))
|
| elif e.startswith("cert:") and s == "cert_days_left":
|
| proposals.append((e, "renew_certificate", "certbot renew --cert-name %s" % e[5:], "%s expires in %.1f days" % (e[5:], v),
|
| "cert_days_left jumps to ~89", (_run(["certbot", "renew", "--dry-run", "--cert-name", e[5:]], timeout=120) or "certbot not runnable")[-200:]))
|
| elif e.startswith("job:") and s == "last_run_ok" and v == 0:
|
| u = e[4:]
|
| tail = _run(["journalctl", "-u", u, "-n", "3", "--no-pager", "-q", "-o", "cat"], timeout=15).strip().replace("\n", " | ")
|
| proposals.append((e, "investigate_job", "journalctl -u %s -n 50" % u, "job %s last run failed" % u, "read-only", tail[:200]))
|
| for e, kind, cmd, pre, pred, dry in proposals:
|
| c.execute("INSERT INTO action (made, entity, kind, command, precondition, predicted, dry_run, status, approved, executed) "
|
| "VALUES (?,?,?,?,?,?,?, 'proposed', 0, 0)", (now, e, kind, cmd, pre, pred, dry))
|
| return proposals
|
|
|
|
|
| EFFECT = {
|
| "restart_service": ("up", lambda b, a: a == 1.0),
|
| "journal_vacuum": ("free_gb", lambda b, a: b is not None and a > b),
|
| "apt_clean": ("free_gb", lambda b, a: b is not None and a >= b),
|
| "renew_certificate": ("cert_days_left", lambda b, a: a > 30),
|
| }
|
|
|
|
|
| def verify_actions(c, now):
|
| """an executed action's predicted effect is checked against the readings that follow it (>= 2 minutes later)"""
|
| for aid, ent, kind, exe, obs in c.execute("SELECT id, entity, kind, executed_at, observed FROM action "
|
| "WHERE executed=1 AND matched IS NULL").fetchall():
|
| if kind not in EFFECT or not exe or now - exe < 120:
|
| continue
|
| sig, test = EFFECT[kind]
|
| b = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts<? ORDER BY ts DESC LIMIT 1", (ent, sig, exe)).fetchone()
|
| a = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts>? ORDER BY ts DESC LIMIT 1", (ent, sig, exe + 60)).fetchone()
|
| if a is None:
|
| continue
|
| ok = bool(test(b[0] if b else None, a[0]))
|
| c.execute("UPDATE action SET observed=?, matched=? WHERE id=?",
|
| (json.dumps({"signal": sig, "before": b[0] if b else None, "after": a[0]}), 1 if ok else 0, aid))
|
|
|
|
|
| def admitted_actions(c):
|
| """an action CLASS is admitted for autonomy after 5 executions whose observed effect matched the prediction, none failed"""
|
| out = {}
|
| for kind, n, ok in c.execute("SELECT kind, COUNT(*), SUM(matched) FROM action WHERE executed=1 GROUP BY kind").fetchall():
|
| out[kind] = {"executed": n, "matched": ok or 0, "admitted": n >= 5 and (ok or 0) == n}
|
| return out
|
|
|
|
|
|
|
| def retain(c, now):
|
| cut = now - RAW_KEEP
|
| last_agg = meta_get(c, "last_agg", 0)
|
| for b in range(max(last_agg, cut - RAW_KEEP) // 600 * 600, cut // 600 * 600, 600):
|
| c.execute("INSERT INTO agg SELECT ?, entity, signal, AVG(value) FROM fact WHERE ts>=? AND ts<? GROUP BY entity, signal", (b, b, b + 600))
|
| meta_set(c, "last_agg", b + 600)
|
| c.execute("DELETE FROM fact WHERE ts<?", (cut,))
|
| c.execute("DELETE FROM agg WHERE ts<?", (now - AGG_KEEP,))
|
| c.execute("DELETE FROM forecast WHERE made<?", (now - AGG_KEEP,))
|
| c.execute("DELETE FROM event WHERE ts<?", (now - AGG_KEEP,))
|
| if os.path.exists(DB) and os.path.getsize(DB) / 1e6 > CAP_MB:
|
| c.execute("DELETE FROM agg WHERE ts < (SELECT MIN(ts) + (MAX(ts)-MIN(ts))/4 FROM agg)")
|
| c.execute("DELETE FROM fact WHERE ts < (SELECT MIN(ts) + (MAX(ts)-MIN(ts))/4 FROM fact)")
|
| c.commit()
|
| c.execute("VACUUM")
|
|
|
|
|
|
|
| def run():
|
| c = db()
|
| prev, last_t = {}, time.time()
|
| last_fc = last_slow = last_disc = 0
|
| disc = {}
|
| slow = []
|
| while True:
|
| t0 = time.time()
|
| now, dt = int(t0), t0 - last_t
|
| if now - last_disc >= DISCOVER_EVERY or not disc:
|
| disc = discover()
|
| meta_set(c, "discovery", disc)
|
| last_disc = now
|
| facts = sense_host(prev) + sense_disks() + sense_net(prev, dt)
|
| uf, ev = sense_units(prev, dt, disc)
|
| facts += uf + sense_journal(int(last_t))
|
| if now - last_slow >= SLOW_EVERY:
|
| slow = sense_certs(disc) + sense_sites(disc) + sense_sources()
|
| facts += slow
|
| last_slow = now
|
| c.executemany("INSERT INTO fact VALUES (?,?,?,?,?)", [(now, e, s, v, src) for e, s, v, src in facts])
|
| c.executemany("INSERT INTO event VALUES (?,?,?,?)", [(now, e, k, d) for e, k, d in ev])
|
| if now - last_fc >= FORECAST_EVERY:
|
| score_forecasts(c, now)
|
| make_forecasts(c, now)
|
| plan_actions(c, now)
|
| verify_actions(c, now)
|
| retain(c, now)
|
| last_fc = now
|
| c.commit()
|
| last_t = t0
|
| time.sleep(max(5, STEP - (time.time() - t0)))
|
|
|
|
|
|
|
| def latest_values(c):
|
| """the newest value of every (entity, signal) in the last 15 minutes (slow sensors run every 5)"""
|
| t = c.execute("SELECT MAX(ts) FROM fact").fetchone()[0]
|
| if not t:
|
| return None, []
|
| rows = c.execute("SELECT f.entity, f.signal, f.value, f.source, f.ts FROM fact f JOIN (SELECT entity, signal, MAX(ts) m FROM fact "
|
| "WHERE ts>=? GROUP BY entity, signal) x ON f.entity=x.entity AND f.signal=x.signal AND f.ts=x.m", (t - 900,)).fetchall()
|
| return t, rows
|
|
|
|
|
| def status(c):
|
| t, rows = latest_values(c)
|
| if not t:
|
| return {"facts": 0}
|
| crit, warn = [], []
|
| for e, s, v, src, ts in rows:
|
| lv = level(s, v)
|
| if lv:
|
| (crit if lv == "critical" else warn).append({"entity": e, "signal": s, "value": v, "receipt": src})
|
| ev = c.execute("SELECT ts, entity, kind, detail FROM event WHERE ts>=? ORDER BY ts DESC LIMIT 20", (t - 86400,)).fetchall()
|
| return {"at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t)), "critical": crit, "warning": warn,
|
| "events_24h": [{"at": time.strftime("%H:%M", time.gmtime(a)), "entity": b, "kind": k, "detail": d} for a, b, k, d in ev]}
|
|
|
|
|
| def forecast_view(c):
|
| now = int(time.time())
|
| g = gate(c)
|
| out = []
|
| for ent, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=? AND (signal LIKE '%used_pct' OR signal LIKE '%inode_pct')",
|
| (now - 3600,)).fetchall():
|
| rows = series(c, ent, s, now - 6 * 3600)
|
| sl = slope(rows)
|
| th = thresholds(s)
|
| if not rows or sl is None or sl <= 0 or not th or rows[-1][1] >= th[1]:
|
| continue
|
| rec = g.get("%s@6h" % kind_of(s), {})
|
| m = rec.get("admitted")
|
| k = METHODS.get(m, 1.0)
|
| out.append({"entity": ent, "signal": s, "now": rows[-1][1], "per_hour": round(k * sl * 3600, 4), "reaches": th[1],
|
| "in_hours": round((th[1] - rows[-1][1]) / (k * sl) / 3600, 1), "admitted": bool(m), "method": m or "trend6h",
|
| "record": rec or "not yet scored"})
|
| out.sort(key=lambda x: x["in_hours"])
|
| return out
|
|
|
|
|
| def why(c, entity):
|
| t, _ = latest_values(c)
|
| if not t:
|
| return []
|
| out = []
|
| for e, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=?", (t - 3600,)).fetchall():
|
| a = [v for _, v in series(c, e, s, t - 3600)]
|
| if len(a) < 20:
|
| continue
|
| x, y = a[:len(a) // 2], a[len(a) // 2:]
|
| mx, my = sum(x) / len(x), sum(y) / len(y)
|
| sd = (sum((v - mx) ** 2 for v in x) / len(x)) ** 0.5 or 1e-9
|
| z = (my - mx) / sd
|
| if abs(z) >= 3:
|
| out.append({"entity": e, "signal": s, "before": round(mx, 3), "after": round(my, 3), "shift_sd": round(z, 1)})
|
| out.sort(key=lambda d: (d["entity"] != entity, -abs(d["shift_sd"])))
|
| return out[:15]
|
|
|
|
|
| def ask(c, q):
|
| """plain questions answered by the generated readers; every answer cites the readers' facts"""
|
| ql = q.lower()
|
| if re.search(r"\b(fill|full|space|disk)\b", ql):
|
| return {"answer": forecast_view(c) or "no disk is trending toward its threshold", "reader": "forecast"}
|
| if re.search(r"\b(cert|certificate|tls|ssl|expire)", ql):
|
| _, rows = latest_values(c)
|
| return {"answer": sorted([{"site": e[5:], "days_left": v, "receipt": src} for e, s, v, src, _ in rows if s == "cert_days_left"],
|
| key=lambda x: x["days_left"]), "reader": "cert_days_left"}
|
| m = re.search(r"\bwhy\b.*?\b([a-z0-9][\w.:-]+)\??$", ql)
|
| if m:
|
| return {"answer": why(c, m.group(1)), "reader": "why"}
|
| if re.search(r"\b(wrong|broken|fail|down|problem|status|health)", ql):
|
| return {"answer": status(c), "reader": "status"}
|
| if re.search(r"\b(do|fix|action|should)\b", ql):
|
| return {"answer": [dict(zip(("id", "entity", "kind", "command", "precondition", "predicted", "dry_run", "status"), r))
|
| for r in c.execute("SELECT id, entity, kind, command, precondition, predicted, dry_run, status FROM action "
|
| "WHERE status='proposed' ORDER BY id DESC LIMIT 20").fetchall()], "reader": "plan"}
|
| return {"answer": "ask about disks, certificates, failures, why <service>, or what to do", "reader": None}
|
|
|
|
|
| def main():
|
| cmd = sys.argv[1] if len(sys.argv) > 1 else "brief"
|
| if cmd == "run":
|
| return run()
|
| c = db()
|
| if cmd == "status":
|
| print(json.dumps(status(c), indent=1))
|
| elif cmd == "forecast":
|
| print(json.dumps(forecast_view(c), indent=1))
|
| elif cmd == "gate":
|
| print(json.dumps({"forecasts": gate(c), "actions": admitted_actions(c)}, indent=1))
|
| elif cmd == "why":
|
| print(json.dumps(why(c, sys.argv[2] if len(sys.argv) > 2 else "host"), indent=1))
|
| elif cmd == "discover":
|
| d = discover()
|
| print(json.dumps({k: (v if k != "certs" else {p: s for p, s in v.items()}) for k, v in d.items()}, indent=1))
|
| elif cmd == "plan":
|
| plan_actions(c, int(time.time()))
|
| c.commit()
|
| for r in c.execute("SELECT id, entity, kind, command, precondition, predicted, dry_run FROM action WHERE status='proposed' ORDER BY id").fetchall():
|
| print("#%d %s %s\n do: %s\n because: %s\n expect: %s\n dry run: %s" % r)
|
| elif cmd == "approve":
|
| c.execute("UPDATE action SET status='approved', approved=? WHERE id=? AND status='proposed'", (int(time.time()), int(sys.argv[2])))
|
| c.commit()
|
| print("approved #%s (executes only with `execute %s`)" % (sys.argv[2], sys.argv[2]))
|
| elif cmd == "execute":
|
| r = c.execute("SELECT id, kind, command, status FROM action WHERE id=?", (int(sys.argv[2]),)).fetchone()
|
| if not r or r[3] != "approved":
|
| print("refused: action #%s is not approved" % sys.argv[2])
|
| return 1
|
| if r[1].startswith("investigate"):
|
| print(_run(["sh", "-c", r[2]], timeout=60))
|
| else:
|
| res = subprocess.run(["sh", "-c", r[2]], capture_output=True, text=True, timeout=600)
|
| print(res.stdout[-2000:], res.stderr[-1000:])
|
| c.execute("UPDATE action SET status='executed', executed=1, executed_at=? WHERE id=?", (int(time.time()), r[0]))
|
| c.commit()
|
| elif cmd == "ask":
|
| print(json.dumps(ask(c, " ".join(sys.argv[2:])), indent=1, default=str))
|
| elif cmd == "signals":
|
| f = sys.argv[2] if len(sys.argv) > 2 else ""
|
| _, rows = latest_values(c)
|
| print(json.dumps([{"entity": e, "signal": s, "value": v, "receipt": src} for e, s, v, src, _ in rows if f in e + s], indent=1))
|
| elif cmd == "brief":
|
| s = status(c)
|
| n = c.execute("SELECT COUNT(*) FROM fact").fetchone()[0]
|
| disc = meta_get(c, "discovery", {})
|
| print("%s at %s facts %d store %.1f MB | discovered: %d services, %d jobs, %d sites, %d certs, %d ports" % (
|
| socket.gethostname(), s.get("at"), n, os.path.getsize(DB) / 1e6, len(disc.get("services", [])), len(disc.get("jobs", [])),
|
| len(disc.get("sites", [])), len(disc.get("certs", {})), len(disc.get("ports", []))))
|
| for x in s.get("critical", []):
|
| print(" CRITICAL %-34s %-15s %-8s [%s]" % (x["entity"], x["signal"], x["value"], x["receipt"][:60]))
|
| for x in s.get("warning", [])[:10]:
|
| print(" warning %-34s %-15s %-8s [%s]" % (x["entity"], x["signal"], x["value"], x["receipt"][:60]))
|
| for x in forecast_view(c)[:3]:
|
| print(" forecast %-34s %-15s %s now, %+.3f/h -> %s in %sh [%s]" % (x["entity"], x["signal"], x["now"], x["per_hour"],
|
| x["reaches"], x["in_hours"], "ADMITTED" if x["admitted"] else "unproven"))
|
| g = gate(c)
|
| print(" gate: %d signal forecasts scored, %d with an admitted method; actions proposed %d, autonomy %s" % (
|
| len(g), sum(1 for v in g.values() if v.get("admitted")),
|
| c.execute("SELECT COUNT(*) FROM action WHERE status='proposed'").fetchone()[0], "ON" if AUTONOMOUS else "off"))
|
| else:
|
| print(__doc__)
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
