feat(ops-log): attribute host changes across two agents sharing one identity
infra-ops and infra-hermes act as the same OS identity and dockerd does not
log exec per caller, so host-side changes carry no fingerprint. Git cannot
close the gap either: every commit here is attributed to Vuong Hoang by
convention, which is correct for authorship and useless for attribution.
On 2026-09-18 a second session edited the searxng stack mid-deploy, crash-
looping fleet search for ~4 minutes, and the author was unidentifiable.
scripts/ops-log records one line per host-changing action and holds a
lightweight claim so two agents do not deploy the same stack at once.
Four design questions, settled:
* Central on nh3-dev, not per-host and not the post office. Both agents
run as the same unix user there, so one file is shared with zero
provisioning. Per-host needs a writable path on ~25 heterogeneous boxes
and stores "we changed host Y" on host Y. journald looked free but shows
an unprivileged reader only their own _UID, which would have split the
log silently between the infra-ops and lkraven halves of the fleet.
* The claim is advisory and enforced in the tooling. deploy-stack.sh
refuses a foreign claim across the diff, the prompt and the apply -- the
whole review window, which is where the collision happened. Acquire is
mkdir, so it is atomic rather than probably-fine. Stale claims auto-break
and the break is recorded.
* 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.
* There is a detector. `ops-log audit` asks each host what changed on disk
and compares it to the newest log line for that stack, covering the
manual ssh-and-edit path the automatic writers structurally cannot.
ops-log being absent or broken never blocks a deploy; only a live foreign
claim does. `ops-log baseline` marks the 136 stacks that predate the
instrument so the detector starts from today rather than reporting the whole
fleet forever and training us to ignore it.
An unreachable host reports INCOMPLETE and exit 5, never clean.
This commit is contained in:
+55
-1
@@ -28,6 +28,19 @@
|
||||
# Optional environment:
|
||||
# DEPLOY_DEST_STACK=<name> retain a legacy remote stack directory/project
|
||||
# DEPLOY_SUDO=1 use passwordless sudo for remote files and rsync
|
||||
# DEPLOY_NO_CLAIM=1 skip the ops-log claim (see below — use sparingly)
|
||||
# DEPLOY_CLAIM_TTL=<dur> how long the claim stays live (default 30m)
|
||||
#
|
||||
# OPS LOG + CLAIM (added 2026-09-19)
|
||||
# `infra-ops` and `infra-hermes` are two agents sharing ONE OS identity, so
|
||||
# host-side changes are otherwise fingerprint-less. Before touching the
|
||||
# host this script claims <host>/<stack> via scripts/ops-log and holds the
|
||||
# claim across the diff, the y/N prompt and the apply — that whole window is
|
||||
# where the 2026-09-18 searxng collision happened, not just the rsync. It
|
||||
# then records what it pushed.
|
||||
# A REFUSAL (another agent holds the claim) is fatal. ops-log being absent
|
||||
# or broken is NOT: the deploy path must not gain a new single point of
|
||||
# failure just because it grew an audit trail.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
@@ -76,7 +89,7 @@ for a in "$@"; do
|
||||
--yes|-y) ASSUME_YES=1 ;;
|
||||
--compose) DO_CONF=0 ;;
|
||||
--conf) DO_COMPOSE=0 ;;
|
||||
-h|--help) sed -n '2,22p' "$0"; exit 0 ;;
|
||||
-h|--help) sed -n '2,43p' "$0"; exit 0 ;;
|
||||
-*) echo "error: unknown flag $a" >&2; exit 2 ;;
|
||||
*)
|
||||
if [ -z "$HOST" ]; then HOST="$a"
|
||||
@@ -101,6 +114,31 @@ case "$DEST_STACK" in
|
||||
esac
|
||||
[[ "$DEST_STACK" =~ ^[a-zA-Z0-9][a-zA-Z0-9_.-]*$ ]] || { echo "invalid stack name: $DEST_STACK" >&2; exit 2; }
|
||||
|
||||
# --------- Claim the stack before any remote work. --------------------
|
||||
OPS_LOG="$SCRIPT_DIR/ops-log"
|
||||
CLAIMED=0
|
||||
release_claim() {
|
||||
if [ "$CLAIMED" -eq 1 ]; then
|
||||
"$OPS_LOG" release "$HOST" "$STACK" -q >/dev/null 2>&1 || true
|
||||
CLAIMED=0
|
||||
fi
|
||||
return 0
|
||||
}
|
||||
trap release_claim EXIT
|
||||
|
||||
if [ -x "$OPS_LOG" ] && [ "${DEPLOY_NO_CLAIM:-0}" != 1 ]; then
|
||||
claim_rc=0
|
||||
"$OPS_LOG" claim "$HOST" "$STACK" --ttl "${DEPLOY_CLAIM_TTL:-30m}" \
|
||||
--why "deploy-stack.sh $HOST $STACK" -q || claim_rc=$?
|
||||
case "$claim_rc" in
|
||||
0) CLAIMED=1 ;;
|
||||
3) echo "error: refused — see the claim above. Wait for the holder, coordinate" >&2
|
||||
echo " on althing, or override with DEPLOY_NO_CLAIM=1 if it is dead." >&2
|
||||
exit 3 ;;
|
||||
*) echo "warning: ops-log claim failed (exit $claim_rc) — deploying UNCLAIMED." >&2 ;;
|
||||
esac
|
||||
fi
|
||||
|
||||
resolve_target() {
|
||||
# ssh-target file wins when present (may carry user@ or non-default port);
|
||||
# /etc/hosts + ssh_config is the fallback.
|
||||
@@ -315,4 +353,20 @@ for entry in "${PAIRS[@]}"; do
|
||||
"$src" "$dest"
|
||||
done
|
||||
|
||||
# --------- Record what we just did. -----------------------------------
|
||||
# Counted from the dry-run itemize, which is what the operator actually
|
||||
# reviewed and approved — not re-derived after the fact.
|
||||
if [ -x "$OPS_LOG" ]; then
|
||||
summary=""
|
||||
for entry in "${PAIRS[@]}"; do
|
||||
IFS='|' read -r kind _ _ <<<"$entry"
|
||||
n_ch=$(grep -c . <<<"${CHANGED_FILES_BY_KIND[$kind]:-}" || true)
|
||||
n_del=$(grep -c . <<<"${DELETED_FILES_BY_KIND[$kind]:-}" || true)
|
||||
summary+="${summary:+, }$kind ${n_ch:-0} changed/${n_del:-0} deleted"
|
||||
done
|
||||
[ "$DEST_STACK" != "$STACK" ] && summary+=" (remote dir $DEST_STACK)"
|
||||
"$OPS_LOG" record --host "$HOST" --action deploy-stack --target "$STACK" \
|
||||
--outcome changed --detail "$summary" -q || true
|
||||
fi
|
||||
|
||||
echo "done."
|
||||
|
||||
@@ -737,6 +737,54 @@ def print_summary(step_results: list[StepResult], verify_results: list[StepResul
|
||||
return exit_code
|
||||
|
||||
|
||||
# ─── ops log ───────────────────────────────────────────────────────────────
|
||||
#
|
||||
# `infra-ops` and `infra-hermes` are two agents behind one OS identity, so a
|
||||
# change made over ssh is otherwise fingerprint-less. elway is the sanctioned
|
||||
# way to CHANGE things on a host, which makes it the right place to record
|
||||
# them — automatically, because a log you have to remember to write is the
|
||||
# same class of instrument as a health check that passes in both states.
|
||||
# See scripts/ops-log and docs/pfi/ops-log.md.
|
||||
|
||||
OPS_LOG_BIN = Path(__file__).resolve().parent / "ops-log"
|
||||
|
||||
|
||||
def record_to_ops_log(host: str, playbook: Optional[str], adhoc: Optional[str],
|
||||
step_results: list, verify_results: list) -> None:
|
||||
"""Best-effort. A failure here must never change elway's exit code or
|
||||
output — the audit trail is not allowed to become a new failure mode on
|
||||
the path that fixes things."""
|
||||
if not OPS_LOG_BIN.is_file():
|
||||
return
|
||||
tallies = []
|
||||
for phase, res in (("steps", step_results), ("verify", verify_results)):
|
||||
if not res:
|
||||
continue
|
||||
t = _tally(res)
|
||||
tallies.append(f"{phase} {t['ok']} ok/{t['changed']} changed/"
|
||||
f"{t['failed']} failed/{t['skipped']} skipped")
|
||||
all_res = step_results + verify_results
|
||||
if any(r.state == "failed" for r in all_res):
|
||||
outcome = "failed"
|
||||
elif any(r.state == "changed" for r in all_res):
|
||||
outcome = "changed"
|
||||
else:
|
||||
outcome = "ok"
|
||||
target = Path(playbook).stem if playbook else "ad-hoc"
|
||||
detail = "; ".join(tallies) or "no steps ran"
|
||||
if adhoc:
|
||||
# Truncated: the ops log is an index of what happened, not a
|
||||
# transcript — the full command is in the elway run log.
|
||||
detail += f" | {adhoc[:160]}"
|
||||
try:
|
||||
subprocess.run(
|
||||
[str(OPS_LOG_BIN), "record", "--host", host, "--action", "elway",
|
||||
"--target", target, "--outcome", outcome, "--detail", detail, "-q"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ─── CLI ───────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -874,6 +922,10 @@ def main(argv: Optional[list[str]] = None) -> int:
|
||||
dry_run=args.dry_run,
|
||||
)
|
||||
rc = print_summary(step_results, verify_results)
|
||||
if not args.dry_run:
|
||||
record_to_ops_log(args.host, args.playbook,
|
||||
args.shell or args.upload, step_results,
|
||||
verify_results)
|
||||
if perm_log is not None:
|
||||
print(DIM(f"log saved: {perm_log}"))
|
||||
return rc
|
||||
|
||||
Executable
+622
@@ -0,0 +1,622 @@
|
||||
#!/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)
|
||||
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 and is set
|
||||
per pane by dev-launch; fall back to user@box so a human shell is still
|
||||
distinguishable rather than anonymous."""
|
||||
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"{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)
|
||||
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'})")
|
||||
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())
|
||||
Reference in New Issue
Block a user