vh
4e778aeddc
The fallback recorded `lkraven@nh3-dev` when ALTHING_HANDLE was unset, which reads like an answer while being the absence of one. It now records `unattributed(lkraven@nh3-dev)` -- handles never contain @ or parentheses, so the fallback cannot be mistaken for an identity. Found by the commit hook's first independent test, which was not one I ran: infra-hermes committed0fe4da6and the hook caught it unprompted -- but logged him as lkraven@nh3-dev, not infra-hermes. His seat exports ALTHING_HANDLE from althing-infra-hermes-seat-run.sh, while neither hermes-gateway.service nor his pump unit carries it, so anything he does outside that one script logs with no handle. So the hook is real but partially blind: it distinguishes not-infra-ops from infra-ops, which is enough to have told mee43e262was not mine, and not enough to name who made it. The env-var gap is his to close. Making the blind spot announce itself in the record is mine -- an instrument that cannot answer should say so rather than return something answer-shaped.
651 lines
27 KiB
Python
Executable File
651 lines
27 KiB
Python
Executable File
#!/usr/bin/python3
|
|
# Pinned to /usr/bin/python3 rather than `env python3` so an active user
|
|
# virtualenv can't shadow it — same reasoning as scripts/elway.
|
|
"""
|
|
ops-log — who changed what, when, on the fleet; plus a lightweight claim
|
|
so two agents don't deploy the same stack at the same time.
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
`infra-ops` and `infra-hermes` are two different agents that act as the SAME
|
|
OS identity (`ssh infra-ops@<host>`), and dockerd does not log `exec` per
|
|
caller. Host-side changes are therefore fingerprint-less: when a host differs
|
|
from expectation, neither agent can tell whether the other did it, a prior
|
|
session did, or something broke on its own. Every commit in this repo is
|
|
attributed to Vuong Hoang by convention, so git does not close the gap either.
|
|
|
|
Not hypothetical: on 2026-09-18 a second session edited the searxng stack
|
|
while another was deploying it, crash-looping fleet search for ~4 minutes,
|
|
and the author was unidentifiable afterwards.
|
|
|
|
DESIGN DECISIONS (settled 2026-09-19 — see docs/pfi/ops-log.md for the
|
|
rationale in full)
|
|
* The log is CENTRAL on nh3-dev, not per-host and not on the post office.
|
|
Both agents run as the same unix user on nh3-dev, so one file is shared
|
|
instantly with zero provisioning, zero permissions story, and no
|
|
dependency on a service that could be down during the incident you are
|
|
trying to reconstruct.
|
|
* The claim is ADVISORY-BUT-ENFORCED-IN-TOOLING: `deploy-stack.sh` refuses
|
|
a stack another agent has claimed. Nothing stops a raw `ssh`; the point
|
|
is to make the tooling path safe, not to build a cage.
|
|
* Writers are AUTOMATIC. `deploy-stack.sh` and `elway` record themselves.
|
|
A log that depends on remembering is the same class of instrument as a
|
|
health check that passes in both states — so `audit` exists to catch the
|
|
manual changes that were never recorded, rather than trusting discipline.
|
|
|
|
STORAGE
|
|
<repo>/.ops-log/ops.jsonl append-only, one JSON object per line
|
|
<repo>/.ops-log/claims/<key>/ one directory per live claim (mkdir = atomic)
|
|
Both are gitignored: this is runtime record, not committed intent. Override
|
|
the location with OPS_LOG_DIR.
|
|
|
|
USAGE
|
|
ops-log record --host <h> --action <a> [--target <t>] [--outcome <o>]
|
|
[--detail <text>]
|
|
ops-log claim <host> <target> [--ttl 30m] [--why '<what you are doing>']
|
|
ops-log release <host> <target>
|
|
ops-log check <host> <target> # exit 0 free or mine, 3 held by another
|
|
ops-log claims # list live claims
|
|
ops-log tail [--host H] [--target T] [--since 24h] [-n 40]
|
|
ops-log audit [host ...] # on-host changes with no log line
|
|
ops-log baseline [host ...] # lay the epoch (run once, at adoption)
|
|
|
|
EXIT CODES
|
|
0 success / claim acquired / nothing to report
|
|
2 usage error
|
|
3 claim refused (held by another agent)
|
|
10 claim REFRESHED — you already held it. The caller did NOT acquire it and
|
|
must NOT release it on the way out (see deploy-stack.sh).
|
|
4 audit found unlogged changes
|
|
5 audit could not reach every host (incomplete — NOT the same as clean)
|
|
"""
|
|
|
|
import argparse
|
|
import datetime
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
OPS_DIR = Path(os.environ.get("OPS_LOG_DIR") or (REPO_ROOT / ".ops-log"))
|
|
LOG_PATH = OPS_DIR / "ops.jsonl"
|
|
CLAIMS_DIR = OPS_DIR / "claims"
|
|
|
|
DEFAULT_TTL_S = 30 * 60
|
|
OUTCOMES = ("ok", "changed", "failed", "refused", "skipped")
|
|
|
|
|
|
# ─── identity ──────────────────────────────────────────────────────────────
|
|
|
|
def agent_id() -> str:
|
|
"""Who is acting. ALTHING_HANDLE is the fleet's agent identity.
|
|
|
|
When it is unset the record says so IN THE NAME — `unattributed(user@box)`
|
|
— rather than quietly recording a unix login as though it were an agent
|
|
handle. Handles never contain `@` or parentheses, so the fallback can never
|
|
be misread as one.
|
|
|
|
This is not hypothetical: infra-hermes's seat exports ALTHING_HANDLE from
|
|
`althing-infra-hermes-seat-run.sh`, but neither `hermes-gateway.service`
|
|
nor his pump unit carries it, so work he does outside that one script logs
|
|
with no handle. An entry that merely said `lkraven@nh3-dev` would look like
|
|
an answer while being the absence of one."""
|
|
handle = os.environ.get("ALTHING_HANDLE", "").strip()
|
|
if handle:
|
|
return handle
|
|
user = os.environ.get("USER") or os.environ.get("LOGNAME") or "unknown"
|
|
return f"unattributed({user}@{os.uname().nodename})"
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def parse_duration(text: str) -> int:
|
|
"""'30m' / '2h' / '90s' / '45' (bare = seconds) → seconds."""
|
|
m = re.fullmatch(r"\s*(\d+)\s*([smhd]?)\s*", text or "")
|
|
if not m:
|
|
raise SystemExit(f"ops-log: bad duration '{text}' (want e.g. 30m, 2h, 90s)")
|
|
n = int(m.group(1))
|
|
return n * {"": 1, "s": 1, "m": 60, "h": 3600, "d": 86400}[m.group(2)]
|
|
|
|
|
|
def claim_key(host: str, target: str) -> str:
|
|
"""Filesystem-safe join of host and target. Both are sanitized rather than
|
|
quoted because this becomes a directory name we mkdir/rmdir by hand."""
|
|
safe = lambda s: re.sub(r"[^A-Za-z0-9._-]", "_", s.strip()) or "_"
|
|
return f"{safe(host)}__{safe(target)}"
|
|
|
|
|
|
# ─── the log ───────────────────────────────────────────────────────────────
|
|
|
|
def append(record: dict) -> None:
|
|
"""Append one JSON line under an exclusive flock.
|
|
|
|
Appends under PIPE_BUF are already atomic on Linux, but the flock costs
|
|
nothing and makes the guarantee explicit rather than inherited from a
|
|
size assumption that a long --detail could quietly break."""
|
|
OPS_DIR.mkdir(parents=True, exist_ok=True)
|
|
line = json.dumps(record, separators=(",", ":"), sort_keys=True) + "\n"
|
|
with open(LOG_PATH, "a", encoding="utf-8") as fh:
|
|
fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
fh.write(line)
|
|
fh.flush()
|
|
finally:
|
|
fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
|
|
|
|
|
|
def record(host: str, action: str, target: str = "", outcome: str = "changed",
|
|
detail: str = "", extra: dict | None = None) -> dict:
|
|
rec = {
|
|
"ts": now_iso(),
|
|
"agent": agent_id(),
|
|
"host": host,
|
|
"action": action,
|
|
"target": target,
|
|
"outcome": outcome,
|
|
"detail": detail,
|
|
"origin": f"{os.environ.get('USER', '?')}@{os.uname().nodename}",
|
|
"pid": os.getpid(),
|
|
}
|
|
if extra:
|
|
rec.update(extra)
|
|
append(rec)
|
|
return rec
|
|
|
|
|
|
def read_records(limit: int | None = None, since_s: int | None = None,
|
|
host: str | None = None, target: str | None = None) -> list[dict]:
|
|
if not LOG_PATH.exists():
|
|
return []
|
|
cutoff = None
|
|
if since_s is not None:
|
|
cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds=since_s)
|
|
out = []
|
|
with open(LOG_PATH, encoding="utf-8") as fh:
|
|
for line in fh:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
rec = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
# A torn or hand-edited line must not blind the reader to the
|
|
# rest of the file — surface it, keep going.
|
|
out.append({"ts": "?", "agent": "?", "host": "?", "action": "UNPARSEABLE",
|
|
"target": "", "outcome": "failed", "detail": line[:200]})
|
|
continue
|
|
if host and rec.get("host") != host:
|
|
continue
|
|
if target and rec.get("target") != target:
|
|
continue
|
|
if cutoff is not None:
|
|
try:
|
|
ts = datetime.datetime.strptime(rec.get("ts", ""), "%Y-%m-%dT%H:%M:%SZ")
|
|
ts = ts.replace(tzinfo=datetime.timezone.utc)
|
|
except ValueError:
|
|
ts = None
|
|
if ts is not None and ts < cutoff:
|
|
continue
|
|
out.append(rec)
|
|
if limit:
|
|
out = out[-limit:]
|
|
return out
|
|
|
|
|
|
# ─── claims ────────────────────────────────────────────────────────────────
|
|
|
|
def claim_state(host: str, target: str) -> tuple[str, dict | None]:
|
|
"""→ ('free'|'mine'|'theirs'|'stale', holder-or-None)."""
|
|
d = CLAIMS_DIR / claim_key(host, target)
|
|
holder_path = d / "holder.json"
|
|
if not holder_path.exists():
|
|
# A claim dir with no holder file is a half-written acquire; treat it
|
|
# as stale so it can be reclaimed rather than wedging the stack.
|
|
return ("stale", None) if d.exists() else ("free", None)
|
|
try:
|
|
holder = json.loads(holder_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return ("stale", None)
|
|
try:
|
|
since = datetime.datetime.strptime(holder["since"], "%Y-%m-%dT%H:%M:%SZ")
|
|
since = since.replace(tzinfo=datetime.timezone.utc)
|
|
age = (datetime.datetime.now(datetime.timezone.utc) - since).total_seconds()
|
|
except (KeyError, ValueError):
|
|
return ("stale", holder)
|
|
if age > holder.get("ttl_s", DEFAULT_TTL_S):
|
|
return ("stale", holder)
|
|
return ("mine" if holder.get("agent") == agent_id() else "theirs", holder)
|
|
|
|
|
|
def _drop_claim(host: str, target: str) -> None:
|
|
"""Remove a claim. Deliberately NOT a recursive delete on a computed path:
|
|
one named file, then rmdir, which refuses if anything else is in there."""
|
|
d = CLAIMS_DIR / claim_key(host, target)
|
|
holder_path = d / "holder.json"
|
|
try:
|
|
os.remove(holder_path)
|
|
except FileNotFoundError:
|
|
pass
|
|
try:
|
|
os.rmdir(d)
|
|
except (FileNotFoundError, OSError):
|
|
pass
|
|
|
|
|
|
def cmd_claim(args) -> int:
|
|
state, holder = claim_state(args.host, args.target)
|
|
if state == "theirs" and not args.steal:
|
|
age = holder_age(holder)
|
|
sys.stderr.write(
|
|
f"ops-log: REFUSED — {args.host}/{args.target} is claimed by "
|
|
f"{holder.get('agent')} since {holder.get('since')} ({age}).\n"
|
|
f" why: {holder.get('why') or '(not stated)'}\n"
|
|
f" If that claim is dead, break it with --steal (it is recorded).\n")
|
|
record(args.host, "claim-refused", args.target, outcome="refused",
|
|
detail=f"held by {holder.get('agent')}: {holder.get('why', '')}")
|
|
return 3
|
|
if state in ("stale", "theirs"):
|
|
detail = (f"broke {holder.get('agent')}'s claim ({holder.get('why', '')})"
|
|
if holder else "broke an empty claim dir")
|
|
_drop_claim(args.host, args.target)
|
|
record(args.host, "claim-broken", args.target,
|
|
outcome="changed" if state == "theirs" else "ok", detail=detail)
|
|
if state == "mine":
|
|
# Already ours: leave the holder file ALONE. A sub-tool refreshing a
|
|
# claim would otherwise overwrite the reason and TTL the original
|
|
# claimant chose — so a 45-minute "rollout in progress" becomes
|
|
# "deploy-stack.sh <host> <stack>", and whoever gets refused reads the
|
|
# wrong story about why. Report 10 and change nothing.
|
|
if not args.quiet:
|
|
print(f"already claimed by you: {args.host}/{args.target} "
|
|
f"({(holder or {}).get('why') or 'no reason given'}, "
|
|
f"{holder_age(holder)})")
|
|
return 10
|
|
d = CLAIMS_DIR / claim_key(args.host, args.target)
|
|
try:
|
|
d.mkdir(parents=True, exist_ok=False) # atomic: this IS the acquire
|
|
except FileExistsError:
|
|
if state != "mine":
|
|
sys.stderr.write(f"ops-log: REFUSED — {args.host}/{args.target} was "
|
|
f"claimed by someone else between the check and the "
|
|
f"acquire. Re-run.\n")
|
|
return 3
|
|
holder = {
|
|
"agent": agent_id(),
|
|
"host": args.host,
|
|
"target": args.target,
|
|
"since": now_iso(),
|
|
"ttl_s": parse_duration(args.ttl),
|
|
"why": args.why or "",
|
|
"pid": os.getpid(),
|
|
"origin": f"{os.environ.get('USER', '?')}@{os.uname().nodename}",
|
|
}
|
|
(d / "holder.json").write_text(json.dumps(holder, indent=2) + "\n", encoding="utf-8")
|
|
if not args.quiet:
|
|
print(f"claimed {args.host}/{args.target} for {args.ttl} "
|
|
f"({holder['why'] or 'no reason given'})")
|
|
# Exit 10 when this only REFRESHED a claim the caller already held. A tool
|
|
# that claims-then-releases around its own work must not drop a longer
|
|
# claim wrapping a multi-step operation. Measured 2026-09-19: a 45-minute
|
|
# operation claim on nh3-docker/althing-post-office was silently released
|
|
# by deploy-stack.sh's exit trap partway through the rollout it was
|
|
# protecting.
|
|
return 0
|
|
|
|
|
|
def holder_age(holder: dict | None) -> str:
|
|
if not holder:
|
|
return "unknown age"
|
|
try:
|
|
since = datetime.datetime.strptime(holder["since"], "%Y-%m-%dT%H:%M:%SZ")
|
|
since = since.replace(tzinfo=datetime.timezone.utc)
|
|
except (KeyError, ValueError):
|
|
return "unknown age"
|
|
secs = int((datetime.datetime.now(datetime.timezone.utc) - since).total_seconds())
|
|
if secs < 90:
|
|
return f"{secs}s ago"
|
|
if secs < 5400:
|
|
return f"{secs // 60}m ago"
|
|
return f"{secs // 3600}h ago"
|
|
|
|
|
|
def cmd_release(args) -> int:
|
|
state, holder = claim_state(args.host, args.target)
|
|
if state == "free":
|
|
if not args.quiet:
|
|
print(f"no claim on {args.host}/{args.target}")
|
|
return 0
|
|
if state == "theirs" and not args.steal:
|
|
sys.stderr.write(f"ops-log: {args.host}/{args.target} is held by "
|
|
f"{holder.get('agent')}, not you — use --steal to force.\n")
|
|
return 3
|
|
_drop_claim(args.host, args.target)
|
|
if not args.quiet:
|
|
print(f"released {args.host}/{args.target}")
|
|
return 0
|
|
|
|
|
|
def cmd_check(args) -> int:
|
|
state, holder = claim_state(args.host, args.target)
|
|
if state == "theirs":
|
|
print(f"HELD by {holder.get('agent')} since {holder.get('since')} "
|
|
f"({holder_age(holder)}): {holder.get('why') or '(no reason)'}")
|
|
return 3
|
|
print({"free": "free", "mine": "held by you", "stale": "stale (reclaimable)"}[state])
|
|
return 0
|
|
|
|
|
|
def cmd_claims(args) -> int:
|
|
if not CLAIMS_DIR.exists():
|
|
print("no claims")
|
|
return 0
|
|
rows = []
|
|
for d in sorted(CLAIMS_DIR.iterdir()):
|
|
if not d.is_dir():
|
|
continue
|
|
host, _, target = d.name.partition("__")
|
|
state, holder = claim_state(host, target)
|
|
if state == "free":
|
|
continue
|
|
rows.append((state, holder, host, target))
|
|
if not rows:
|
|
print("no claims")
|
|
return 0
|
|
for state, holder, host, target in rows:
|
|
agent = (holder or {}).get("agent", "?")
|
|
why = (holder or {}).get("why") or "(no reason)"
|
|
flag = " STALE" if state == "stale" else ""
|
|
print(f"{host}/{target:<24} {agent:<14} {holder_age(holder):<10}{flag} {why}")
|
|
return 0
|
|
|
|
|
|
# ─── read paths ────────────────────────────────────────────────────────────
|
|
|
|
def cmd_record(args) -> int:
|
|
if args.outcome not in OUTCOMES:
|
|
raise SystemExit(f"ops-log: --outcome must be one of {', '.join(OUTCOMES)}")
|
|
rec = record(args.host, args.action, args.target, args.outcome, args.detail)
|
|
if not args.quiet:
|
|
print(f"logged: {rec['ts']} {rec['agent']} {rec['host']} {rec['action']} "
|
|
f"{rec['target']} [{rec['outcome']}]")
|
|
return 0
|
|
|
|
|
|
def cmd_tail(args) -> int:
|
|
recs = read_records(limit=args.n,
|
|
since_s=parse_duration(args.since) if args.since else None,
|
|
host=args.host, target=args.target)
|
|
if not recs:
|
|
print("(no records)")
|
|
return 0
|
|
for r in recs:
|
|
detail = f" — {r['detail']}" if r.get("detail") else ""
|
|
where = r.get("host", "?") + (f"/{r['target']}" if r.get("target") else "")
|
|
print(f"{r.get('ts','?')} {r.get('agent','?'):<14} {where:<30} "
|
|
f"{r.get('action','?'):<16} [{r.get('outcome','?')}]{detail}")
|
|
return 0
|
|
|
|
|
|
# ─── the detector ──────────────────────────────────────────────────────────
|
|
#
|
|
# The manual path — someone ssh'ing in and editing a compose file by hand —
|
|
# is the one the automatic writers cannot cover, and a rule that says
|
|
# "remember to log it" is exactly the instrument that passes in both states.
|
|
# So instead of trusting discipline, ask the host: what changed on disk more
|
|
# recently than the newest log line about it?
|
|
|
|
def cmd_audit(args) -> int:
|
|
if args.hosts == ["all"]:
|
|
hosts = all_audit_hosts()
|
|
else:
|
|
hosts = args.hosts or default_audit_hosts()
|
|
findings = 0
|
|
unreachable: list[str] = []
|
|
for host in hosts:
|
|
newest = newest_stack_mtimes(host, args.since)
|
|
if newest is None:
|
|
print(f"{host}: UNREACHABLE (not audited)")
|
|
unreachable.append(host)
|
|
continue
|
|
if not newest:
|
|
continue
|
|
for stack, mtime in sorted(newest.items()):
|
|
logged = last_record_for(host, stack)
|
|
if logged and logged >= mtime:
|
|
continue
|
|
who = "no log line at all" if not logged else f"newest log line {logged}"
|
|
print(f"UNLOGGED {host}/{stack}: changed on host {mtime} — {who}")
|
|
findings += 1
|
|
|
|
reached = len(hosts) - len(unreachable)
|
|
if findings:
|
|
print(f"\n{findings} unlogged change(s) across {reached} host(s). Record them "
|
|
f"(`ops-log record --host <h> --action <what> --target <stack>`) "
|
|
f"or find out who did it.")
|
|
if unreachable:
|
|
print(f"INCOMPLETE: {len(unreachable)} host(s) not audited "
|
|
f"({', '.join(unreachable)}) — there may be more.")
|
|
return 4
|
|
# An unreachable host is NOT a clean host. Saying "clean" here would make
|
|
# this exactly the instrument it exists to replace: one that passes
|
|
# whether or not it actually looked.
|
|
if unreachable:
|
|
print(f"audit INCOMPLETE: {reached} host(s) clean, "
|
|
f"{len(unreachable)} NOT audited ({', '.join(unreachable)}).")
|
|
return 5
|
|
print(f"audit clean: {reached} host(s) reached, every on-host stack change "
|
|
f"has a log line at or after it.")
|
|
return 0
|
|
|
|
|
|
# The hosts `deploy-stack.sh` actually pushes stacks to. Deliberately an
|
|
# explicit list rather than "every dir under servers/": an audit that quietly
|
|
# ssh's into all ~25 hosts — hypervisors, the NAS, and the SureFire TENANT
|
|
# boxes we are contractually meant to coordinate on — is a surprise, not a
|
|
# feature. `ops-log audit all` opts into the wider sweep and still excludes
|
|
# sf-*.
|
|
#
|
|
# corviduo-dev is excluded on purpose: Worldtree deploys there are CI/CD-driven
|
|
# and rewrite the tree constantly, so it would report unlogged changes forever
|
|
# and train us to ignore the output.
|
|
STACK_HOSTS = ["ana-docker", "nh3-docker", "esh-docker-vm", "vm-esh-nas",
|
|
"fv-ml1", "irv-ml1"]
|
|
|
|
|
|
def default_audit_hosts() -> list[str]:
|
|
return list(STACK_HOSTS)
|
|
|
|
|
|
def all_audit_hosts() -> list[str]:
|
|
servers = REPO_ROOT / "servers"
|
|
if not servers.is_dir():
|
|
return default_audit_hosts()
|
|
return sorted(d.name for d in servers.iterdir()
|
|
if d.is_dir() and (d / "README.md").exists()
|
|
and not d.name.startswith("sf")
|
|
and d.name != "corviduo-dev")
|
|
|
|
|
|
def ssh_target(host: str) -> str:
|
|
fb = REPO_ROOT / "servers" / host / "ssh-target"
|
|
if fb.is_file():
|
|
for line in fb.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
return line.split()[0]
|
|
return host
|
|
|
|
|
|
def newest_stack_mtimes(host: str, since: str | None) -> dict[str, str] | None:
|
|
"""Per stack under /opt/docker/compose, the newest file mtime inside it.
|
|
`since=None` means every stack, unfiltered (used to lay the baseline).
|
|
|
|
Returns None when the host could not be reached — an unreachable host is
|
|
NOT an empty result, and conflating the two is how an audit reports clean
|
|
for a box it never talked to."""
|
|
window = ""
|
|
if since is not None:
|
|
secs = parse_duration(since)
|
|
window = f"-newermt '@'$(( $(date +%s) - {secs} )) "
|
|
remote = (
|
|
"find /opt/docker/compose -mindepth 2 -type f "
|
|
+ window +
|
|
r"-printf '%T@ %p\n' 2>/dev/null || true"
|
|
)
|
|
try:
|
|
proc = subprocess.run(
|
|
["ssh", "-n", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
|
|
ssh_target(host), remote],
|
|
capture_output=True, text=True, timeout=45)
|
|
except (subprocess.TimeoutExpired, OSError):
|
|
return None
|
|
if proc.returncode != 0:
|
|
return None
|
|
newest: dict[str, float] = {}
|
|
for line in proc.stdout.splitlines():
|
|
parts = line.strip().split(" ", 1)
|
|
if len(parts) != 2:
|
|
continue
|
|
try:
|
|
epoch = float(parts[0])
|
|
except ValueError:
|
|
continue
|
|
rel = parts[1][len("/opt/docker/compose/"):]
|
|
stack = rel.split("/", 1)[0]
|
|
if epoch > newest.get(stack, 0.0):
|
|
newest[stack] = epoch
|
|
return {k: datetime.datetime.fromtimestamp(v, datetime.timezone.utc)
|
|
.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
for k, v in newest.items()}
|
|
|
|
|
|
def last_record_for(host: str, stack: str) -> str | None:
|
|
best = None
|
|
for rec in read_records(host=host):
|
|
if rec.get("target") != stack:
|
|
continue
|
|
ts = rec.get("ts")
|
|
if ts and (best is None or ts > best):
|
|
best = ts
|
|
return best
|
|
|
|
|
|
def cmd_baseline(args) -> int:
|
|
"""Lay the epoch: one record per stack that already exists, marking it as
|
|
pre-ops-log.
|
|
|
|
Without this, the detector's first useful run is a week away and until then
|
|
it reports every stack on the fleet as unlogged — 66 of them on 2026-09-19.
|
|
Nobody acts on 66 findings, so the instrument gets ignored, which is the
|
|
exact failure it was built to avoid.
|
|
|
|
The line says attribution is UNAVAILABLE rather than pretending the change
|
|
was accounted for. It closes the detector, not the question."""
|
|
hosts = all_audit_hosts() if args.hosts == ["all"] else (args.hosts or default_audit_hosts())
|
|
written = 0
|
|
unreachable = []
|
|
for host in hosts:
|
|
stacks = newest_stack_mtimes(host, None)
|
|
if stacks is None:
|
|
print(f"{host}: UNREACHABLE — NOT baselined, it will keep reporting.")
|
|
unreachable.append(host)
|
|
continue
|
|
for stack, mtime in sorted(stacks.items()):
|
|
if last_record_for(host, stack) and not args.force:
|
|
continue
|
|
record(host, "baseline", stack, outcome="ok",
|
|
detail=f"pre-ops-log state; newest file {mtime}; "
|
|
f"attribution for this and everything before it is UNAVAILABLE")
|
|
written += 1
|
|
print(f"{host}: baselined {len(stacks)} stack(s)")
|
|
print(f"\n{written} baseline record(s) written across "
|
|
f"{len(hosts) - len(unreachable)} host(s).")
|
|
if unreachable:
|
|
print(f"INCOMPLETE: {', '.join(unreachable)} not reached.")
|
|
return 5
|
|
return 0
|
|
|
|
|
|
# ─── CLI ───────────────────────────────────────────────────────────────────
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(
|
|
prog="ops-log",
|
|
description="Fleet ops log + lightweight stack claim.",
|
|
epilog=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
quiet = argparse.ArgumentParser(add_help=False)
|
|
quiet.add_argument("-q", "--quiet", action="store_true",
|
|
help="suppress success chatter")
|
|
|
|
r = sub.add_parser("record", parents=[quiet], help="append one line to the ops log")
|
|
r.add_argument("--host", required=True)
|
|
r.add_argument("--action", required=True,
|
|
help="short verb phrase: deploy, restart, edit-compose, playbook, ...")
|
|
r.add_argument("--target", default="", help="stack / service / file the action touched")
|
|
r.add_argument("--outcome", default="changed", help=f"one of: {', '.join(OUTCOMES)}")
|
|
r.add_argument("--detail", default="", help="free text: what actually changed")
|
|
r.set_defaults(func=cmd_record)
|
|
|
|
c = sub.add_parser("claim", parents=[quiet], help="claim a host+target so other agents' tooling refuses it")
|
|
c.add_argument("host")
|
|
c.add_argument("target")
|
|
c.add_argument("--ttl", default="30m", help="how long the claim stays live (default 30m)")
|
|
c.add_argument("--why", default="", help="what you are doing — shown to whoever is refused")
|
|
c.add_argument("--steal", action="store_true", help="break another agent's live claim")
|
|
c.set_defaults(func=cmd_claim)
|
|
|
|
rel = sub.add_parser("release", parents=[quiet], help="drop your claim")
|
|
rel.add_argument("host")
|
|
rel.add_argument("target")
|
|
rel.add_argument("--steal", action="store_true", help="drop someone else's claim")
|
|
rel.set_defaults(func=cmd_release)
|
|
|
|
ck = sub.add_parser("check", help="is this host+target claimed? (exit 3 if held by another)")
|
|
ck.add_argument("host")
|
|
ck.add_argument("target")
|
|
ck.set_defaults(func=cmd_check)
|
|
|
|
cl = sub.add_parser("claims", help="list live claims")
|
|
cl.set_defaults(func=cmd_claims)
|
|
|
|
t = sub.add_parser("tail", help="read recent ops-log lines")
|
|
t.add_argument("--host")
|
|
t.add_argument("--target")
|
|
t.add_argument("--since", default="", help="e.g. 24h, 7d")
|
|
t.add_argument("-n", type=int, default=40)
|
|
t.set_defaults(func=cmd_tail)
|
|
|
|
a = sub.add_parser("audit", help="find on-host stack changes with no log line")
|
|
a.add_argument("hosts", nargs="*",
|
|
help=f"default: the stack hosts ({', '.join(STACK_HOSTS)}); "
|
|
f"pass 'all' to sweep every non-tenant host under servers/")
|
|
a.add_argument("--since", default="7d", help="look back this far on the host (default 7d)")
|
|
a.set_defaults(func=cmd_audit)
|
|
|
|
b = sub.add_parser("baseline",
|
|
help="mark every existing stack as pre-ops-log, so audit "
|
|
"starts from today instead of reporting the whole fleet")
|
|
b.add_argument("hosts", nargs="*", help="default: the stack hosts; 'all' for the wider sweep")
|
|
b.add_argument("--force", action="store_true",
|
|
help="re-baseline stacks that already have a log line")
|
|
b.set_defaults(func=cmd_baseline)
|
|
return p
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv if argv is not None else sys.argv[1:])
|
|
return args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|