EIC code displayed by LXR master/​swf-remote/​scripts/​stageout_index.py	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2026-09-28 09:37:31

 #!/usr/bin/env python3
 """Index of the devcloud stage-out bucket, learned from the bucket's own events.
 
 The gateway knows what is in the bucket by being told: S3 emits an
 ObjectCreated event per report object (and per status write, counted by
 the hour rather than indexed) to the epic-stageout-events queue,
 this tool drains the queue into a local index, and the watch and the
 nightly cross-check read the index instead of listing S3. Listing a
 production-scale prefix every few minutes would cost about as much as
 the reporting itself; a queue costs a dollar a month
 (swf-epicprod docs/JOB_REPORTING.md).
 
 Standard library plus boto3 for the drain and the AWS CLI for the rest,
 both on this host's account credentials. No venv, no web-tier coupling:
 the tool owns its state and other things read it. The drain is the one
 hot path: one in-process SQS client, ten messages received and ten
 deleted per call. A CLI process per message cost 0.7 s of CPU each and at
 campaign rate stacked the five-minute runs 25 deep (2026-09-15).
 
 Modes:
   drain      receive events, record objects, delete the messages
   reconcile  authoritative count by listing, compared with the index
   apply      apply the sweeper's spooled pass records to the index
   guard      nightly: is this an explosion, and pull the plug if so
   summary    the state as JSON, for the Capcom tile and for a human
   wipe       by hand: delete the prefix as of now, recorded as a pass
 Cron on ec2dev, admin's crontab: drain every five minutes under
 `flock -n` and `nice -n 19` (a run that finds the lock held exits; runs
 never stack), reconcile at 03:40 UTC, guard at 03:50 UTC.
 """
 
 import argparse
 from datetime import datetime
 import json
 import os
 import re
 import resource
 import sqlite3
 import subprocess
 import sys
 import time
 
 BUCKET = os.environ.get("STAGEOUT_BUCKET", "epic-devcloud-stageout")
 PREFIX = os.environ.get("STAGEOUT_PREFIX", "reports/")
 QUEUE_URL = os.environ.get(
     "STAGEOUT_QUEUE_URL",
     "https://sqs.us-east-1.amazonaws.com/962718900486/epic-stageout-events")
 DB_PATH = os.environ.get(
     "STAGEOUT_INDEX_DB", "/home/admin/data/stageout-index.sqlite")
 REGION = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
 # Objects per hour above which the watch calls it a runaway. A defect that
 # posts in a loop inside one job is the unbounded risk the design names;
 # the fleet itself is finite.
 RATE_CEILING_PER_HOUR = int(os.environ.get("STAGEOUT_RATE_CEILING", "20000"))
 # A job writing more than this many objects is looping, whatever the fleet
 # total says.
 PER_JOB_CEILING = int(os.environ.get("STAGEOUT_PER_JOB_CEILING", "50"))
 
 # The nightly guard's ceilings. Expected production is of the order of
 # 100,000 objects a day at a few kilobytes each; these are the levels at
 # which the traffic stops being production and starts being a defect.
 REPORTING_USER = os.environ.get("STAGEOUT_REPORTING_USER", "epic-job-reporter")
 OBJECT_SOFT_CEILING_PER_DAY = int(
     os.environ.get("STAGEOUT_OBJECT_SOFT", "500000"))
 OBJECT_HARD_CEILING_PER_DAY = int(
     os.environ.get("STAGEOUT_OBJECT_HARD", "1000000"))
 BYTES_SOFT_CEILING_PER_DAY = int(os.environ.get("STAGEOUT_BYTES_SOFT",
                                                 str(5 * 10 ** 9)))
 BYTES_HARD_CEILING_PER_DAY = int(os.environ.get("STAGEOUT_BYTES_HARD",
                                                 str(50 * 10 ** 9)))
 # A day that is this many times the recent norm is an explosion whatever
 # the absolute ceilings say; the floor keeps a quiet week from tripping it.
 RELATIVE_MULTIPLE = int(os.environ.get("STAGEOUT_RELATIVE_MULTIPLE", "20"))
 RELATIVE_FLOOR = int(os.environ.get("STAGEOUT_RELATIVE_FLOOR", "5000"))
 
 # The sweep runs inside the perimeter, on the production operations agent,
 # because every decision it makes needs the inside: the payload digest,
 # the validation that an id names a real ePIC production job, and the
 # record it files into. It deletes what it takes and reports what it
 # deleted here, since S3 tells a bucket's owner nothing about another
 # party's deletions and a count cannot say which rows to retire.
 PASS_SPOOL = os.environ.get("STAGEOUT_PASS_SPOOL",
                             "/var/lib/stageout/passes")
 # The sweeper reports hourly. Three missed hours is comfortably clear of one
 # slow pass and still tight against a dead sweeper (agreed 2026-09-06).
 PASS_STALL_SECONDS = int(os.environ.get("STAGEOUT_PASS_STALL", str(3 * 3600)))
 # A listing sees an object the instant it lands; its event reaches the
 # queue seconds to a minute later and the drain runs every five minutes.
 # An unindexed object younger than this is not yet heard, not unheard
 # (2026-09-11: two objects written in the cross-check's minute read as
 # unheard for a day).
 RECONCILE_GRACE_SECONDS = int(os.environ.get("STAGEOUT_RECONCILE_GRACE",
                                              str(15 * 60)))
 
 SCHEMA = """
 CREATE TABLE IF NOT EXISTS objects (
     key       TEXT PRIMARY KEY,
     size      INTEGER NOT NULL,
     etime     TEXT NOT NULL,
     subject   TEXT,
     seen_at   REAL NOT NULL
 );
 CREATE INDEX IF NOT EXISTS objects_seen ON objects (seen_at);
 CREATE INDEX IF NOT EXISTS objects_subject ON objects (subject);
 CREATE TABLE IF NOT EXISTS state (k TEXT PRIMARY KEY, v TEXT NOT NULL);
 CREATE TABLE IF NOT EXISTS passes (
     pass_id      TEXT PRIMARY KEY,
     reporter     TEXT,
     outcome      TEXT NOT NULL,
     reason       TEXT,
     window_from  TEXT,
     window_to    TEXT,
     jobs_filed   INTEGER NOT NULL,
     keys_read    INTEGER NOT NULL,
     keys_unread  INTEGER NOT NULL,
     retired      INTEGER NOT NULL,
     received_at  TEXT,
     applied_at   REAL NOT NULL
 );
 CREATE INDEX IF NOT EXISTS passes_applied ON passes (applied_at);
 CREATE TABLE IF NOT EXISTS daily (
     day       TEXT PRIMARY KEY,
     objects   INTEGER NOT NULL,
     bytes     INTEGER NOT NULL,
     at        REAL NOT NULL
 );
 CREATE TABLE IF NOT EXISTS status_writes (
     hour      INTEGER PRIMARY KEY,
     writes    INTEGER NOT NULL,
     bytes     INTEGER NOT NULL
 );
 """
 
 # status/<pandaid>.json is one object per job, overwritten while the job
 # runs in PanDA debug mode (payload 0.22.0, call home; at most 150 writes a
 # job). Overwrites are the cost, so these events are counted per hour as
 # writes rather than indexed as objects; the guard adds them to its day.
 STATUS_PREFIX = "status/"
 
 # reports/<subject>/<sequence>.json — the subject is the PanDA job id once
 # the payload carries it; anything else is recorded as it arrives.
 SUBJECT_RE = re.compile(r"^reports/([^/]+)/")
 
 
 def aws(*args, parse=True):
     """Run one AWS CLI call. A failure is raised with its stderr rather
     than swallowed: a silent indexer is worse than none."""
     proc = subprocess.run(("aws",) + args, capture_output=True, text=True)
     if proc.returncode != 0:
         raise RuntimeError(
             f"aws {' '.join(args[:3])} failed ({proc.returncode}): "
             f"{proc.stderr.strip()[:400]}")
     if not parse:
         return proc.stdout
     out = proc.stdout.strip()
     return json.loads(out) if out else {}
 
 
 def db():
     os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
     conn = sqlite3.connect(DB_PATH, timeout=30)
     conn.executescript(SCHEMA)
     return conn
 
 
 def set_state(conn, key, value):
     conn.execute("INSERT INTO state (k, v) VALUES (?, ?) "
                  "ON CONFLICT(k) DO UPDATE SET v = excluded.v",
                  (key, json.dumps(value)))
 
 
 def get_state(conn, key, default=None):
     row = conn.execute("SELECT v FROM state WHERE k = ?", (key,)).fetchone()
     return json.loads(row[0]) if row else default
 
 
 def _epoch(iso):
     """Seconds since the epoch for a listing's LastModified
     (2026-09-12T03:40:01+00:00 or ...Z); an unreadable one counts as old."""
     if not iso:
         return 0.0
     try:
         return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()
     except ValueError:
         return 0.0
 
 
 def subject_of(key):
     match = SUBJECT_RE.match(key)
     return match.group(1) if match else None
 
 
 def drain(conn, max_batches=500):
     """Receive events and record their objects. Returns (recorded, seen).
 
     A message is deleted only after its rows are committed, so a crash
     repeats work rather than losing it; the key is the primary key, so a
     repeat is a no-op. A run is bounded at max_batches rounds of ten, so
     a backlog drains across runs at a few percent of one core rather than
     all at once; the cron line's flock keeps runs from overlapping.
     """
     import boto3  # the drain alone needs it; the other modes stay on the CLI
     sqs = boto3.client("sqs", region_name=REGION)
     recorded = seen = 0
     for _ in range(max_batches):
         result = sqs.receive_message(
             QueueUrl=QUEUE_URL, MaxNumberOfMessages=10, WaitTimeSeconds=1)
         messages = result.get("Messages") or []
         if not messages:
             break
         entries = []
         for n, message in enumerate(messages):
             seen += 1
             entries.append({"Id": str(n), "ReceiptHandle": message["ReceiptHandle"]})
             try:
                 body = json.loads(message["Body"])
             except (KeyError, ValueError) as exc:
                 print(f"undecodable message body: {exc}", file=sys.stderr)
                 continue
             # S3 sends one test event when the notification is configured.
             if body.get("Event") == "s3:TestEvent":
                 continue
             for record in body.get("Records", []):
                 obj = record.get("s3", {}).get("object", {})
                 key = obj.get("key")
                 if not key:
                     continue
                 if key.startswith(STATUS_PREFIX):
                     conn.execute(
                         "INSERT INTO status_writes (hour, writes, bytes) "
                         "VALUES (?, 1, ?) ON CONFLICT(hour) DO UPDATE SET "
                         "writes = writes + 1, bytes = bytes + excluded.bytes",
                         (int(time.time() // 3600), int(obj.get("size") or 0)))
                     recorded += 1
                     continue
                 conn.execute(
                     "INSERT INTO objects (key, size, etime, subject, seen_at) "
                     "VALUES (?, ?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET "
                     "size = excluded.size, etime = excluded.etime",
                     (key, int(obj.get("size") or 0),
                      record.get("eventTime", ""), subject_of(key), time.time()))
                 recorded += 1
         conn.commit()
         reply = sqs.delete_message_batch(QueueUrl=QUEUE_URL, Entries=entries)
         failed = reply.get("Failed") or []
         if failed:
             # An undeleted message returns after its visibility timeout and
             # is re-recorded as a no-op; say so rather than hide it.
             print(f"delete_message_batch: {len(failed)} of {len(entries)} "
                   f"failed: {failed[0].get('Code')} "
                   f"{failed[0].get('Message', '')[:200]}", file=sys.stderr)
     set_state(conn, "last_drain", {"at": time.time(), "recorded": recorded})
     conn.commit()
     return recorded, seen
 
 
 def sweep_state(conn, now):
     """(time of the last applied pass, stall text or None).
 
     Three states, not two. A sweeper reporting normally is silent here. A
     sweeper past its threshold is stalled. A sweeper that has never
     reported is neither, until someone declares that it should be
     running: the commissioning stamp in `sweeper_expected_from` is what
     turns absence into staleness, and it is set by hand when the far side
     says the first real pass is imminent.
     """
     # A gateway wipe is recorded as a pass but says nothing of the sweeper.
     last_pass = conn.execute(
         "SELECT MAX(applied_at) FROM passes "
         "WHERE COALESCE(reporter, '') != 'gateway-wipe'").fetchone()[0]
     if last_pass:
         if now - last_pass > PASS_STALL_SECONDS:
             return last_pass, (f"no sweep pass reported for "
                                f"{(now - last_pass) / 3600:.1f} hours")
         return last_pass, None
     expected = get_state(conn, "sweeper_expected_from")
     if expected and now - float(expected) > PASS_STALL_SECONDS:
         return None, ("the sweeper was expected to be running and has "
                       "never reported a pass")
     return None, None
 
 
 def apply_passes(conn):
     """Apply the sweeper's spooled pass records to the index.
 
     The sweep runs inside the perimeter and deletes what it has taken, and
     S3 tells the owner nothing about another party's deletions. So the
     party that deletes reports it, and this is where those reports land in
     the index (swf-epicprod docs/JOB_REPORTING.md).
 
     A record is applied once and its file removed. An unreadable record is
     moved aside rather than deleted or retried forever: the sweeper treats
     its 202 as final and will never send it again, so losing it silently
     would be losing it for good.
     """
     if not os.path.isdir(PASS_SPOOL):
         return {"applied": 0, "retired": 0, "rejected": 0}
     applied = retired = rejected = 0
     now = time.time()
     for name in sorted(os.listdir(PASS_SPOOL)):
         if not name.endswith(".json"):
             continue
         path = os.path.join(PASS_SPOOL, name)
         try:
             with open(path) as handle:
                 record = json.load(handle)
         except (OSError, ValueError) as exc:
             print(f"unreadable pass record {name}: {exc}", file=sys.stderr)
             try:
                 os.rename(path, path + ".rejected")
             except OSError:
                 pass
             rejected += 1
             continue
         keys_read = [k for k in (record.get("deleted_read") or [])
                      if isinstance(k, str)]
         keys_unread = [k for k in (record.get("deleted_unread") or [])
                        if isinstance(k, str)]
         gone = 0
         for key in keys_read + keys_unread:
             gone += conn.execute(
                 "DELETE FROM objects WHERE key = ?", (key,)).rowcount
         window = record.get("window") or {}
         conn.execute(
             "INSERT INTO passes (pass_id, reporter, outcome, reason, "
             "window_from, window_to, jobs_filed, keys_read, keys_unread, "
             "retired, received_at, applied_at) "
             "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
             "ON CONFLICT(pass_id) DO NOTHING",
             (name[:-5], record.get("reporter"), record.get("outcome", "ok"),
              record.get("reason"), window.get("from"), window.get("to"),
              len(record.get("filed") or record.get("pandaids") or []),
              len(keys_read), len(keys_unread), gone,
              record.get("received_at"), now))
         conn.commit()
         os.remove(path)
         applied += 1
         retired += gone
     if applied:
         set_state(conn, "last_pass_applied", {"at": now, "records": applied,
                                               "retired": retired})
         conn.commit()
     return {"applied": applied, "retired": retired, "rejected": rejected}
 
 
 def reconcile(conn):
     """The nightly cross-check: count the prefix authoritatively and
     compare with the index, then drop rows for objects that are gone.
 
     Listing once a night costs a fraction of a cent. Drift is the signal
     that events were missed or that something writes without telling us;
     an object younger than RECONCILE_GRACE_SECONDS is left to the drain.
     """
     sizes = {}
     modified = {}
     token = None
     while True:
         args = ["s3api", "list-objects-v2", "--bucket", BUCKET,
                 "--prefix", PREFIX, "--output", "json"]
         if token:
             args += ["--starting-token", token]
         page = aws(*args)
         for item in page.get("Contents", []):
             sizes[item["Key"]] = int(item.get("Size") or 0)
             modified[item["Key"]] = item.get("LastModified") or ""
         token = page.get("NextContinuationToken")
         if not (page.get("IsTruncated") and token):
             break
     keys = set(sizes)
 
     indexed = {row[0] for row in conn.execute("SELECT key FROM objects")}
     gone_from_bucket = indexed - keys            # expired or swept
     # Unindexed keys split by age: an event for a fresh object is still
     # in flight, so only the older ones were never heard about.
     horizon = time.time() - RECONCILE_GRACE_SECONDS
     pending = {k for k in keys - indexed if _epoch(modified[k]) > horizon}
     missing_from_index = keys - indexed - pending
 
     for key in gone_from_bucket:
         conn.execute("DELETE FROM objects WHERE key = ?", (key,))
     # The listing carries the size, so an object recorded here is as
     # complete as one announced by an event, minus the event time.
     for key in missing_from_index:
         conn.execute(
             "INSERT INTO objects (key, size, etime, subject, seen_at) "
             "VALUES (?, ?, '', ?, ?) ON CONFLICT(key) DO NOTHING",
             (key, sizes[key], subject_of(key), time.time()))
 
     outcome = {"at": time.time(), "bucket_count": len(keys),
                "index_count": len(indexed),
                "unheard": len(missing_from_index),
                "pending": len(pending),
                "pruned": len(gone_from_bucket)}
     set_state(conn, "last_reconcile", outcome)
     conn.commit()
     return outcome
 
 
 def guard(conn, arm=True):
     """The nightly guard: does the traffic look like an explosion, and
     pull the plug if it does.
 
     The signal is the index, not the bill: objects and bytes are what
     boom first and the invoice is their lagging shadow. Two absolute
     ceilings and one relative one, because a ceiling set against today's
     expectation ages badly, and a sudden multiple of the recent norm is
     the shape of a runaway whatever the norm was.
 
     The plug is the reporting key. Disabling it stops jobs already
     running, since a running job keeps the environment it started with
     and the payload treats a failed write as ordinary (swf-epicprod
     docs/JOB_REPORTING.md). Only that key is touched: the sweep and
     stage-out keys survive, so reading and log stage-out continue.
     """
     now = time.time()
     day_objects, day_bytes = conn.execute(
         "SELECT COUNT(*), COALESCE(SUM(size), 0) FROM objects "
         "WHERE seen_at > ?", (now - 86400,)).fetchone()
     # Status overwrites are writes like any report's, on the same key.
     status_writes, status_bytes = status_since(conn, now, 24)
     day_objects += status_writes
     day_bytes += status_bytes
     # The trailing norm, excluding the last day, over whatever history
     # the index holds. Expiry keeps that to about a week.
     prior_days = conn.execute(
         "SELECT COUNT(*) FROM objects WHERE seen_at <= ? AND seen_at > ?",
         (now - 86400, now - 7 * 86400)).fetchone()[0]
     norm_per_day = prior_days / 6.0 if prior_days else 0.0
 
     breaches = []
     if day_objects > OBJECT_HARD_CEILING_PER_DAY:
         breaches.append(f"{day_objects:,} objects in 24 hours, hard ceiling "
                         f"{OBJECT_HARD_CEILING_PER_DAY:,}")
     if day_bytes > BYTES_HARD_CEILING_PER_DAY:
         breaches.append(f"{day_bytes / 1e9:.1f} GB written in 24 hours, hard "
                         f"ceiling {BYTES_HARD_CEILING_PER_DAY / 1e9:.0f} GB")
     if (norm_per_day >= RELATIVE_FLOOR
             and day_objects > norm_per_day * RELATIVE_MULTIPLE):
         breaches.append(
             f"{day_objects:,} objects in 24 hours against a recent norm of "
             f"{norm_per_day:,.0f} a day, over {RELATIVE_MULTIPLE}x")
 
     warnings = []
     if not breaches:
         if day_objects > OBJECT_SOFT_CEILING_PER_DAY:
             warnings.append(f"{day_objects:,} objects in 24 hours, expected "
                             f"under {OBJECT_SOFT_CEILING_PER_DAY:,}")
         if day_bytes > BYTES_SOFT_CEILING_PER_DAY:
             warnings.append(f"{day_bytes / 1e9:.1f} GB in 24 hours, expected "
                             f"under {BYTES_SOFT_CEILING_PER_DAY / 1e9:.0f} GB")
 
     # A stalled sweeper is not an explosion. The sweep is the bucket's
     # normal drain, so when it stops, objects accumulate for a reason that
     # pulling the write credential would not fix and would make worse: the
     # jobs would go quiet while the backlog stayed. It gets its own verdict
     # and never arms the plug.
     #
     # A sweeper that has never run is a third thing again, and saying
     # "stalled" of something not yet built would be a false alarm standing
     # for days, which is how a tile teaches its reader to ignore it. The
     # threshold therefore means nothing until commissioning is declared.
     last_pass, stalled = sweep_state(conn, now)
     if stalled:
         warnings.append(stalled)
 
     stopped = key_status() == "Inactive"
     if breaches and arm and not stopped:
         disable_reporting_key()
         stopped = True
 
     # The day's totals are kept forever, so growth outlives the objects.
     today = time.strftime("%Y-%m-%d", time.gmtime(now - 43200))
     conn.execute(
         "INSERT INTO daily (day, objects, bytes, at) VALUES (?, ?, ?, ?) "
         "ON CONFLICT(day) DO UPDATE SET objects = excluded.objects, "
         "bytes = excluded.bytes, at = excluded.at",
         (today, day_objects, day_bytes, now))
     history = [
         {"day": row[0], "objects": row[1], "bytes": row[2]}
         for row in conn.execute(
             "SELECT day, objects, bytes FROM daily ORDER BY day DESC LIMIT 14")]
     this_week = sum(r["objects"] for r in history[:7])
     last_week = sum(r["objects"] for r in history[7:14])
     growth = (round((this_week - last_week) / last_week, 2)
               if last_week else None)
 
     outcome = {"at": now, "day_objects": day_objects,
                "day_bytes": day_bytes,
                "day_status_writes": status_writes,
                "history": history[:14],
                "week_over_week": growth,
                "norm_objects_per_day": round(norm_per_day),
                "breaches": breaches, "warnings": warnings,
                "sweep_stalled": stalled,
                "last_pass_at": last_pass,
                "reporting_key_stopped": stopped}
     set_state(conn, "last_guard", outcome)
     conn.commit()
     return outcome
 
 
 def status_since(conn, now, hours):
     """(writes, bytes) to the status prefix over the last `hours` hours."""
     return conn.execute(
         "SELECT COALESCE(SUM(writes), 0), COALESCE(SUM(bytes), 0) "
         "FROM status_writes WHERE hour > ?",
         (int(now // 3600) - hours,)).fetchone()
 
 
 def reporting_keys():
     """The reporting user's access keys, from IAM rather than from the
     credential file, so the guard acts on what is actually enabled. The
     user may hold two: a retired key left Inactive beside its successor
     (2026-09-24, the Event Service key), so every key is considered."""
     return aws("iam", "list-access-keys", "--user-name", REPORTING_USER,
                "--output", "json").get("AccessKeyMetadata", [])
 
 
 def key_status():
     """Active if any of the user's keys can write, else Inactive."""
     try:
         keys = reporting_keys()
     except RuntimeError as exc:
         print(f"key status unreadable: {exc}", file=sys.stderr)
         return "Unknown"
     if not keys:
         return "Missing"
     return ("Active" if any(k["Status"] == "Active" for k in keys)
             else "Inactive")
 
 
 def disable_reporting_key():
     """Pull the plug on every active key. Reversible with one command per
     key, named in the notice."""
     active = [k["AccessKeyId"] for k in reporting_keys()
               if k["Status"] == "Active"]
     if not active:
         raise RuntimeError(f"no active access key found for {REPORTING_USER}")
     for key_id in active:
         aws("iam", "update-access-key", "--user-name", REPORTING_USER,
             "--access-key-id", key_id, "--status", "Inactive", parse=False)
         print(f"DISABLED reporting key {key_id}; re-enable with: "
               f"aws iam update-access-key --user-name {REPORTING_USER} "
               f"--access-key-id {key_id} --status Active", file=sys.stderr)
     return active
 
 
 def wipe(conn, reason):
     """Delete every object under the prefix written before this run began,
     retire their rows, and record the deletion as a pass.
 
     For the gateway's own clearances (2026-09-24: the 9/15-9/18 flood,
     cleared on Torre's word). The record goes where the sweeper's do, so
     the index and the guard read a wipe as a known drain and not as
     missing events; objects written during the run are left alone.
     """
     import boto3
     s3 = boto3.client("s3", region_name=REGION)
     started = time.time()
     deleted = errors = 0
     paginator = s3.get_paginator("list_objects_v2")
     for page in paginator.paginate(Bucket=BUCKET, Prefix=PREFIX):
         keys = [item["Key"] for item in page.get("Contents", [])
                 if item["LastModified"].timestamp() < started]
         if not keys:
             continue
         reply = s3.delete_objects(
             Bucket=BUCKET,
             Delete={"Objects": [{"Key": k} for k in keys], "Quiet": True})
         failed = {e["Key"] for e in reply.get("Errors", [])}
         if failed:
             errors += len(failed)
             first = reply["Errors"][0]
             print(f"delete_objects: {len(failed)} failed: {first.get('Code')} "
                   f"{first.get('Message', '')[:200]}", file=sys.stderr)
         gone = [k for k in keys if k not in failed]
         conn.executemany("DELETE FROM objects WHERE key = ?",
                          [(k,) for k in gone])
         conn.commit()
         deleted += len(gone)
     pass_id = time.strftime("wipe-%Y%m%dT%H%M%SZ", time.gmtime(started))
     window_to = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started))
     conn.execute(
         "INSERT INTO passes (pass_id, reporter, outcome, reason, "
         "window_from, window_to, jobs_filed, keys_read, keys_unread, "
         "retired, received_at, applied_at) "
         "VALUES (?, 'gateway-wipe', ?, ?, NULL, ?, 0, 0, ?, ?, ?, ?)",
         (pass_id, "partial" if errors else "ok", reason, window_to,
          deleted, deleted, window_to, time.time()))
     conn.commit()
     return {"pass_id": pass_id, "deleted": deleted, "failed": errors,
             "seconds": round(time.time() - started)}
 
 
 def summary(conn):
     """The state other things read: totals, the last hour's rate against
     the ceiling, the busiest subject, and the freshness of both passes."""
     now = time.time()
     total = conn.execute("SELECT COUNT(*) FROM objects").fetchone()[0]
     hour = conn.execute("SELECT COUNT(*) FROM objects WHERE seen_at > ?",
                         (now - 3600,)).fetchone()[0]
     day = conn.execute("SELECT COUNT(*) FROM objects WHERE seen_at > ?",
                        (now - 86400,)).fetchone()[0]
     status_hour = status_since(conn, now, 1)[0]
     status_day = status_since(conn, now, 24)[0]
     hour += status_hour
     day += status_day
     busiest = conn.execute(
         "SELECT subject, COUNT(*) c FROM objects WHERE seen_at > ? "
         "GROUP BY subject ORDER BY c DESC LIMIT 1", (now - 86400,)).fetchone()
     last_reconcile = get_state(conn, "last_reconcile", {})
     last_pass, stalled = sweep_state(conn, now)
     verdict, detail = "ok", ""
     if stalled:
         verdict, detail = "warning", "the sweeper has reported no pass: " + stalled
     if hour > RATE_CEILING_PER_HOUR:
         verdict = "alarm"
         detail = f"{hour} objects in the last hour, ceiling {RATE_CEILING_PER_HOUR}"
     elif busiest and busiest[1] > PER_JOB_CEILING:
         verdict = "warning"
         detail = (f"subject {busiest[0]} wrote {busiest[1]} objects today, "
                   f"per-job ceiling {PER_JOB_CEILING}")
     elif last_reconcile.get("unheard"):
         verdict = "warning"
         detail = (f"{last_reconcile['unheard']} objects in the bucket that no "
                   "event announced")
     last_guard = get_state(conn, "last_guard", {})
     if last_guard.get("reporting_key_stopped"):
         verdict, detail = "alarm", (
             "the reporting key is disabled: "
             + "; ".join(last_guard.get("breaches") or ["stopped by the guard"]))
     return {"objects": total, "last_hour": hour, "last_day": day,
             "status_last_hour": status_hour, "status_last_day": status_day,
             "guard": last_guard,
             "last_sweep_pass": last_pass,
             "busiest_subject": busiest[0] if busiest else None,
             "busiest_count": busiest[1] if busiest else 0,
             "rate_ceiling": RATE_CEILING_PER_HOUR,
             "last_drain": get_state(conn, "last_drain", {}),
             "last_reconcile": last_reconcile,
             "verdict": verdict, "detail": detail}
 
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
     parser.add_argument(
         "mode",
         choices=("drain", "reconcile", "apply", "guard", "summary", "wipe"))
     parser.add_argument("--no-arm", action="store_true",
                         help="report a breach without disabling the key")
     parser.add_argument("--reason", default="gateway wipe",
                         help="wipe: why, recorded with the pass")
     args = parser.parse_args()
     conn = db()
     try:
         if args.mode == "drain":
             recorded, seen = drain(conn)
             passes = apply_passes(conn)
             used = resource.getrusage(resource.RUSAGE_SELF)
             print(f"drained {seen} messages, recorded {recorded} objects; "
                   f"applied {passes['applied']} sweep passes retiring "
                   f"{passes['retired']} objects"
                   + (f"; {passes['rejected']} rejected"
                      if passes["rejected"] else "")
                   + f"; cpu {used.ru_utime + used.ru_stime:.1f}s")
         elif args.mode == "reconcile":
             print(json.dumps(reconcile(conn), indent=2))
         elif args.mode == "apply":
             print(json.dumps(apply_passes(conn), indent=2))
         elif args.mode == "guard":
             print(json.dumps(guard(conn, arm=not args.no_arm), indent=2))
         elif args.mode == "wipe":
             print(json.dumps(wipe(conn, args.reason), indent=2))
         else:
             print(json.dumps(summary(conn), indent=2))
     finally:
         conn.close()
     return 0
 
 
 if __name__ == "__main__":
     sys.exit(main())

This page was automatically generated by the 2.3.7 LXR engine. The LXR team