fix(logs): unwedge the OOM-failing consolidation job, and mirror tiers to rustfs (#116) (#117)

* feat(logs): mirror consolidated + sessions tiers to rustfs (#116)

geo-agent-ops minted the scoped credential pair, but rustfs
logs-open-llm-proxy is empty, so logs-open-llm-proxy-reader can't
answer anything yet. This is the step that fills it.

Both consolidation CronJobs now copy consolidated/** and sessions/**
to rustfs after the tiers are written and verified.

Credentials bind under RUSTFS_* rather than AWS_*, per the collision
called out in #116: the jobs already bind AWS_ACCESS_KEY_ID /
AWS_SECRET_ACCESS_KEY to the `aws` secret for the NRP source, and
reusing those names would overwrite the source credential and break the
job before it mirrored anything. Endpoint and bucket come from the
Secret too, so moving rustfs doesn't mean editing this repo.

All four bindings are optional:true — a cluster without the Secret still
consolidates and prints the mirror as skipped.

Copy-only, never delete: an accidental source deletion must not
propagate. Re-copies when source LastModified or size differs, which is
what catches the reflatten pass rewriting files in place — size alone is
not a witness, since a re-flatten can land on an identical byte count.

Ordered last so a mirror failure cannot cost the consolidation work, but
it does fail the Job: a mirror that quietly stops is a stale mirror
nobody notices, and the reader would keep serving stale answers.

Verified: both manifests parse as YAML and their embedded Python passes
ast.parse (heredoc indentation intact). Mirror logic exercised against
stub clients over six cases — absent credential skips; empty dest
copies; matching mtime+size skips; same-size-newer-mtime re-copies (the
reflatten case); a 403 on HEAD raises instead of being swallowed as
"missing"; and no path deletes at the destination.

Not run against live rustfs — needs the Secret, which exists only
in-cluster. First scheduled run is the real test.

* fix(logs): raise CronJob memory + isolate backfill — daily job OOM-wedged 9 days

Discovered while checking whether the rustfs mirror was live: it can't be,
because the daily job never reaches the mirror step. It has failed every
night since 2026-08-07.

  OutOfMemoryException: failed to allocate 256.0 KiB (819.0 MiB/819.1 MiB used)
  in build_session_view, backfilling 2026-08-07

Self-perpetuating: the day fails, keeps its slot on the backfill list, and
re-breaks the job the next night. threads=2 and preserve_insertion_order
=false were already set, so the cheap mitigations were spent. New days were
still consolidating (that loop runs first), but session views for
2026-08-07/08/12 are missing and the reflatten pass has not run in 9 days.

These are CronJobs, not persistent pods, so the 2Gi rule for long-lived
workloads doesn't bind. Daily 1Gi -> 4Gi with DuckDB capped at 3GB; monthly
2Gi -> 6Gi capped at 4GB; both get temp_directory plus an ephemeral-storage
request to back the spill.

Headroom is the primary fix and the comment says so. Measured on a
session-view-shaped build (window function + self-join over a wide
payload): spill off failed at 500MB and 1GB, passed at 2GB; spill on
passed at 500MB; at 200MB both failed. DuckDB does not spill every
operator, so temp_directory is a margin, not a cure. An earlier draft of
this change claimed spilling made the failure impossible -- the test above
refuted that and the comment was corrected before commit.

The monthly job is raised although it has not failed: same code over a
whole month is strictly more exposed, and 2026-09-02 rolls up an August
holding the benchmark sweeps.

Backfill now isolates each day. One oversized day is recorded and skipped
so reflatten and the rustfs mirror still run; the job then raises with the
failed-day list, so it stays loud rather than quietly tolerating a gap.

Both manifests parse as YAML and their embedded Python passes ast.parse.

---------

Co-authored-by: Default User <email@example.com>

Author: cboettig

CHANGELOG.md

@@ -96,6 +96,47 @@ See [Releases](README.md#releases) for how a release is cut.
   `deepseek/…` models) route instead of silently falling back to NRP. Enables the
   fleet-wide "DeepSeek V4 Flash (OpenRouter)" picker option.
 
+### Fixed
+- **Daily consolidation had been failing for 9 days (OOM), wedged on one day.** Last
+  success 2026-08-07; every run since died with
+  `OutOfMemoryException: failed to allocate 256.0 KiB (819.0 MiB/819.1 MiB used)` inside
+  `build_session_view`, backfilling `2026-08-07`. Self-perpetuating: the day failed, kept
+  its place on the backfill list, and re-broke the job nightly. `threads=2` and
+  `preserve_insertion_order=false` were already set, so the easy mitigations were spent.
+  These are CronJobs, not persistent pods, so the 2Gi ceiling for long-lived workloads
+  doesn't apply: daily 1Gi → 4Gi (DuckDB capped at 3GB), monthly 2Gi → 6Gi (capped at
+  4GB), both with `temp_directory` set so DuckDB can spill and an `ephemeral-storage`
+  request to back it. Headroom is the primary fix — on a session-view-shaped build,
+  spilling turned failure into success at 500MB–1GB but not at 200MB, since DuckDB does
+  not spill every operator. The monthly job is raised too although it has not failed yet:
+  it runs the same code over a whole month, so it is strictly more exposed, and its next
+  run (2026-09-02) rolls up an August containing the heavy benchmark sweeps.
+  The backfill loop now isolates each day — one oversized day is recorded and skipped so
+  the reflatten pass and the rustfs mirror still run — and the job raises at the end with
+  the failed days, so it stays loud instead of silently tolerating the gap.
+  Missing session views for `2026-08-07`, `08` and `12` should rebuild on the next run.
+
+### Added
+- **Mirror `consolidated/**` and `sessions/**` to rustfs from the consolidation CronJobs
+  (#116).** Log analysis has required the single NRP credential, which carries
+  read/write/delete on *every* NRP bucket, for a read-only task against one bucket of a
+  few MiB (#113). `geo-agent-ops` has minted a scoped pair — `logs-open-llm-proxy-reader`
+  (Get/List only) and `…-writer` (plus object Put/Delete, no bucket create/delete) — but
+  the rustfs bucket was empty, so the reader was useless. Both CronJobs now copy the
+  query-ready tiers there after the tiers are written and verified. Credentials come from
+  the `rustfs-logs-write` Secret under **`RUSTFS_*`** names, deliberately not `AWS_*`:
+  those are already bound to the `aws` secret for the NRP source, and reusing them would
+  clobber the source credential and break the job before it mirrored anything. All four
+  bindings are `optional: true`, so a cluster without the Secret still consolidates and
+  reports the mirror skipped. Copy-only, never delete — an accidental source deletion
+  must not propagate; re-copies when the source `LastModified` or size changes, which is
+  what catches the in-place rewrites the reflatten pass performs (size alone is not a
+  witness). Runs last so a mirror failure cannot cost the consolidation work, but it does
+  fail the Job, because a mirror that quietly stops is a stale mirror nobody notices.
+  NRP Ceph stays the system of record: rustfs shares the same rook Ceph, so this is a
+  convenience copy, not a second failure domain. Consumer-side retarget of `sync-logs.sh`
+  stays in #113 and deliberately does **not** land until the mirror is confirmed non-empty.
+
 ### Fixed
 - **`geo-agent-training` skill: log collection was broken and over-privileged.** Its
   Step 1 selector was `app=llm-proxy`, which matches **no pods** — the label is

LOGGING.md

@@ -117,6 +117,19 @@ request/response interleaving:
 
 ## Access pattern
 
+> 🔜 **A read-only rustfs mirror is being stood up (#116).** The consolidation CronJobs now
+> copy `consolidated/**` and `sessions/**` to `logs-open-llm-proxy` on rustfs after each run,
+> so those tiers can be read with a credential scoped to that one bucket — Get/List only —
+> instead of the single NRP key that carries read/write/delete on *every* NRP bucket (#113).
+> The credential lives in the `rustfs-logs-read` Secret in `biodiversity`; reference it by
+> **name**, never by value, so rotation touches nothing here.
+>
+> **Not yet the recommended path.** The sections below still describe the NRP-key workflow,
+> and stay that way until the mirror has run and been confirmed non-empty — retargeting
+> sooner would make the recommended path return *nothing* rather than too much. NRP Ceph
+> remains the system of record either way; rustfs shares the same rook Ceph, so the mirror
+> is a convenience copy, not a second failure domain.
+
 ### Local sync (recommended for interactive analysis)
 
 The bucket is **private**, but `rclone` already has credentials configured under the `nrp` remote. Sync the bucket to a local scratch dir once per session, then query the local files — no S3 secret, no shell-expanded credentials, and orders of magnitude faster iteration:

consolidate-daily-cronjob.yaml

@@ -37,6 +37,24 @@ spec:
                   valueFrom: { secretKeyRef: { name: aws, key: AWS_ACCESS_KEY_ID } }
                 - name: AWS_SECRET_ACCESS_KEY
                   valueFrom: { secretKeyRef: { name: aws, key: AWS_SECRET_ACCESS_KEY } }
+                # rustfs mirror target (#116). Deliberately NOT named AWS_* — the
+                # consolidation work above binds those to the `aws` secret for the NRP
+                # source, and reusing the names would overwrite the source credential
+                # and break the job before it ever mirrored anything.
+                # `optional: true` throughout: a cluster without the secret still
+                # consolidates, and the mirror step reports itself skipped.
+                - name: RUSTFS_KEY
+                  valueFrom:
+                    secretKeyRef: { name: rustfs-logs-write, key: AWS_ACCESS_KEY_ID, optional: true }
+                - name: RUSTFS_SECRET
+                  valueFrom:
+                    secretKeyRef: { name: rustfs-logs-write, key: AWS_SECRET_ACCESS_KEY, optional: true }
+                - name: RUSTFS_ENDPOINT
+                  valueFrom:
+                    secretKeyRef: { name: rustfs-logs-write, key: AWS_S3_ENDPOINT, optional: true }
+                - name: RUSTFS_BUCKET
+                  valueFrom:
+                    secretKeyRef: { name: rustfs-logs-write, key: BUCKET, optional: true }
               command:
                 - /bin/bash
                 - -c
@@ -46,6 +64,7 @@ spec:
                   python - <<'PY'
                   import os, datetime, boto3, duckdb
                   from botocore.client import Config
+                  from botocore.exceptions import ClientError
 
                   BUCKET   = 'logs-open-llm-proxy'
                   ENDPOINT = 'http://rook-ceph-rgw-nautiluss3.rook'
@@ -103,6 +122,18 @@ spec:
                   # memory bounded (the data is re-sorted by ts on read anyway).
                   con.execute("SET threads=2")
                   con.execute("SET preserve_insertion_order=false")
+                  # Cap DuckDB below the pod limit, and give it somewhere to spill.
+                  #
+                  # The daily job wedged for 9 days OOMing at 819/819 MiB building one
+                  # session view, which blocked the reflatten pass and everything after
+                  # it. Headroom is the primary fix; `temp_directory` is a margin, not a
+                  # cure. Measured on a session-view-shaped build (window function +
+                  # self-join over a wide payload): with spill off it failed at 500MB
+                  # and 1GB and passed at 2GB; with spill on it passed at 500MB. At a
+                  # hard-tight budget (200MB) both still failed — DuckDB does not spill
+                  # every operator. So raise the ceiling *and* allow spilling.
+                  con.execute("SET memory_limit='3GB'")
+                  con.execute("SET temp_directory='/tmp/duckdb-spill'")
                   con.execute(f"""
                       CREATE SECRET s3_logs (
                           TYPE S3,
@@ -266,10 +297,21 @@ spec:
                           s3.delete_objects(Bucket=BUCKET, Delete={'Objects': to_delete[i:i+1000]})
                       print(f"  ✓ {len(to_delete)} JSONL chunks removed; session view written")
 
+                  # Isolate per day. A single oversized day used to abort the whole
+                  # run before the reflatten pass and the rustfs mirror, and it stayed
+                  # on the backfill list, so it re-broke the job every night. Now one
+                  # bad day is recorded and skipped; the rest of the pipeline completes
+                  # and the job still exits non-zero at the end (see `failed` below),
+                  # so the failure is loud rather than silently tolerated.
+                  failed_days = []
                   for day in backfill_sessions:
                       print(f"↺ backfill session view: {day}")
-                      build_session_view(f's3://{BUCKET}/consolidated/daily/{day}.parquet',
-                                         f's3://{BUCKET}/sessions/daily/{day}.parquet')
+                      try:
+                          build_session_view(f's3://{BUCKET}/consolidated/daily/{day}.parquet',
+                                             f's3://{BUCKET}/sessions/daily/{day}.parquet')
+                      except Exception as e:
+                          print(f"  ✗ session view failed for {day}: {type(e).__name__}: {e}")
+                          failed_days.append(day)
 
                   # Schema-upgrade pass: bring any legacy-schema consolidated daily file
                   # (current month — older months live in monthly files, upgraded by the
@@ -280,12 +322,77 @@ spec:
                           upgraded += 1
                           print(f"  ⬆ re-flattened {day} to wide schema")
                   print(f"Legacy daily files upgraded to wide schema: {upgraded}")
+                  # -----------------------------------------------------------------
+                  # Mirror the query-ready tiers to rustfs (#116).
+                  #
+                  # Exists so log analysis can run on the read-only, single-bucket
+                  # `logs-open-llm-proxy-reader` credential instead of the one NRP key,
+                  # which carries read/write/delete on every NRP bucket (#113).
+                  #
+                  # NRP Ceph stays the system of record. rustfs sits on the *same* rook
+                  # Ceph, so this is a convenience mirror, not a second failure domain,
+                  # and must never become the only copy.
+                  #
+                  # Copy-only, never delete: an accidental source deletion must not
+                  # propagate. True mirror semantics would need a --max-delete-style
+                  # guard first (the writer identity can delete objects; this does not).
+                  #
+                  # Runs last, after every tier is written and verified, so a mirror
+                  # failure cannot cost the consolidation work. It does fail the Job
+                  # though — a mirror that quietly stops is a stale mirror nobody
+                  # notices, and the reader would keep serving yesterday's answers.
+                  # -----------------------------------------------------------------
+                  def mirror_to_rustfs(prefixes=('consolidated/', 'sessions/')):
+                      key = os.environ.get('RUSTFS_KEY')
+                      if not key:
+                          print("ℹ️  RUSTFS_KEY not set — skipping rustfs mirror (#116)")
+                          return
+                      dest_bucket = os.environ.get('RUSTFS_BUCKET') or BUCKET
+                      rustfs = boto3.client(
+                          's3',
+                          endpoint_url=os.environ['RUSTFS_ENDPOINT'],
+                          aws_access_key_id=key,
+                          aws_secret_access_key=os.environ['RUSTFS_SECRET'],
+                          config=Config(s3={'addressing_style': 'path'}),
+                      )
+                      copied = skipped = 0
+                      for prefix in prefixes:
+                          for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
+                              for obj in page.get('Contents', []):
+                                  k, stamp = obj['Key'], obj['LastModified'].isoformat()
+                                  # Re-upload whenever the source changed. Size alone is
+                                  # not a witness: the reflatten pass rewrites files in
+                                  # place and can land on an identical byte count.
+                                  try:
+                                      head = rustfs.head_object(Bucket=dest_bucket, Key=k)
+                                      if (head['Metadata'].get('src-mtime') == stamp
+                                              and head['ContentLength'] == obj['Size']):
+                                          skipped += 1
+                                          continue
+                                  except ClientError as e:
+                                      if e.response['Error']['Code'] not in ('404', 'NoSuchKey', 'NotFound'):
+                                          raise   # 403 etc. is a real problem — be loud
+                                  body = s3.get_object(Bucket=BUCKET, Key=k)['Body'].read()
+                                  rustfs.put_object(Bucket=dest_bucket, Key=k, Body=body,
+                                                    Metadata={'src-mtime': stamp})
+                                  copied += 1
+                      print(f"rustfs mirror → {dest_bucket}: {copied} copied, {skipped} already current")
+
+                  mirror_to_rustfs()
+
+                  if failed_days:
+                      raise SystemExit(
+                          f"session view build failed for {len(failed_days)} day(s): "
+                          f"{failed_days} — consolidation, reflatten and the rustfs "
+                          f"mirror completed for everything else")
                   print("Done.")
                   PY
               resources:
                 requests:
                   cpu: "500m"
-                  memory: "1Gi"
+                  memory: "4Gi"
+                  ephemeral-storage: "8Gi"
                 limits:
                   cpu: "500m"
-                  memory: "1Gi"
+                  memory: "4Gi"
+                  ephemeral-storage: "8Gi"