Observability Log Analysis Engine

Pulls errors and user/request flows into one canonical pair of master files, from whichever combination of sources a given job is configured to use:

• openreplay_fe — session/error data directly from a self-hosted OpenReplay instance's Postgres, ClickHouse, and MinIO (frontend errors + user journeys).
• cloudrun_be — Cloud Run logs for a backend service via Google Cloud Logging (backend errors + request flows).
• sentry_fe / sentry_be — errors, Session Replay, and performance-trace data from the Sentry API, for an app instrumented with Sentry instead of OpenReplay/Cloud Run.

A "job" is just a Settings object (which sources to pull, connection details, where to write output) built either from the default .env.pipeline (today's original app: openreplay_fe + cloudrun_be, zero config changes needed) or from a job spec JSON file passed via --job (any other app/combination — see "Jobs" below). Each source de-duplicates into the same pair of master files for that job (error_master.json, test_journey_master.json), writes a Run Log per execution, and generates human-readable write-ups in ERROR_MASTER.md/TEST_JOURNEYS.md. Error Master File entries get fed back into Claude Code to fix the underlying bugs via python pipeline/fix_tasks.py — a separate, on-demand script (see "Generating fix tasks" below), not part of the automatic data-pull loop above.

Direct DB access is used for openreplay_fe instead of OpenReplay's REST API because the public API's Sessions endpoint is scoped per-user with no "all sessions since timestamp" endpoint, which doesn't fit the "all users" requirement here.

Journey data comes from MinIO, not a database

events.pages/clicks/inputs/state_actions exist as Postgres tables on this instance but are never written to — clicks/page-navigations/input changes are only ever stored as OpenReplay's proprietary binary "mob" recording files in MinIO (bucket mobs, one dom.mobs/dom.mobe pair per session). Confirmed by reading the chalice API's source directly on the instance: it has no server-side decoding endpoint anywhere, even via OpenReplay's own REST API — the dashboard only ever hands the browser a presigned URL to these same raw files, and decodes them client-side in its JS player.

pipeline/mob_decoder.py is an independent implementation of that wire format, based on the publicly documented schema at openreplay/openreplay/mobs/messages.rb — that mobs/ directory carries its own MIT license, distinct from and NOT the same as the Enterprise-licensed generated decoder under ee/connectors/msgcodec/ in the same repo (which requires a commercial agreement with OpenReplay and is deliberately not used or copied here). See the module docstring for the reverse-engineered framing details (zstd-compressed, varint/zigzag-encoded messages, BatchMetadata messages re-establishing framing mode mid-stream).

Backend logs come from Cloud Logging, not Docker

The test-revizr backend runs on Google Cloud Run, in the same GCP project (jino-poc) as the OpenReplay VM — not a plain Docker host. Cloud Run automatically ships two log streams to Google Cloud Logging: structured per-request logs (method/URL/status/latency) and the app's own stdout/stderr text logs. pipeline/gcp_logging_client.py pulls both via the google-cloud-logging Python client, authenticated with your local gcloud Application Default Credentials — no SSH tunnel needed for this part.

The backend has no app-level session/correlation ID (checked live response headers and app logs — the only ID present anywhere is GCP's own per-request Cloud Trace ID, which doesn't group requests together), so pipeline/backend_journeys.py groups requests into a "journey" by client IP + an idle-time gap (BACKEND_JOURNEY_IDLE_GAP_MINUTES, default 30 min) — an explicit approximation, not a real session boundary.

Sentry — an alternative source (sentry_fe / sentry_be)

For an app instrumented with Sentry instead of OpenReplay/Cloud Run. pipeline/sentry_client.py calls the Sentry Web API directly (no SDK, no tunnel) using a single auth token; pipeline/sentry_errors.py and pipeline/sentry_journeys.py normalize the results into the same NormalizedError/JourneyStep shapes the other two sources produce, so master_files.py/docs.py/fix_tasks.py need no source-specific code at all.

A few things that aren't obvious from the code:

• Sentry SaaS is regionalized per org (sentry.io, us.sentry.io, de.sentry.io, ...). The org's dashboard vanity URL (https://<org-slug>.sentry.io/) does not tell you the region — the API host does matter and calling the wrong one just fails. The correct region is embedded in the auth token itself: decode the base64 segment between the sntrys_/sntryu_ prefix and the trailing signature to find a region_url field, and set it as sentry.region_url in the job spec (defaults to https://sentry.io if omitted).
• Auth token type matters. Sentry's dashboard offers three token kinds under Developer Settings — Org, Personal, OAuth. Org tokens are hardcoded to the org:ci scope (source-map/release upload only) and will 403 on every read call this pipeline makes. Use a Personal Token with project:read / event:read / org:read scopes instead — it acts with the creating user's own project visibility, which is fine as long as that user can see the target FE/BE projects.
• FE journeys come from two different signals, not one:
  • A replay's urls field (from GET /organizations/{org}/replays/) — page-navigation only, no clicks/inputs. Full rrweb recording decode (which would give the same click-level detail pipeline/mob_decoder.py extracts from OpenReplay) is deliberately out of scope — it's a comparable-sized reverse-engineering effort and Sentry's replay recording format isn't documented as a supported API.
  • Each error event's breadcrumb trail (already present in the same payload fetch_events pulls) — Sentry records recent navigation/ui.click/ui.input breadcrumbs leading up to an error, giving click-level detail, but only for the lead-up to whichever errors were actually captured, not arbitrary complete sessions.
• BE journeys are inherently API-request sequences (method + path), the same nature as the existing Cloud Run journeys — there's no page/DOM concept server-side for either source. Sentry's trace_id is a real distributed-tracing correlation ID though (unlike Cloud Run's IP+idle-gap approximation), so transactions are grouped by trace instead.
• If a project's FE journeys or BE journeys come back empty, it usually means Session Replay / Performance tracing isn't actually wired into that app's Sentry SDK init code yet — neither is a dashboard toggle, both are enabled from the app's own code (Sentry.replayIntegration()
  • replaysSessionSampleRate/replaysOnErrorSampleRate for Replay; tracing sample rate config for Performance). Check the dashboard's Replays tab (not Settings) to see if anything is actually being captured for that project before assuming the pipeline is missing something.

Jobs: running this pipeline for more than one app

Everything above assumes one hardcoded app. In practice you may have several — e.g. one app on OpenReplay + Cloud Run, a different app on Sentry — each needing its own connection details and its own isolated output. pipeline/run.py and pipeline/fix_tasks.py both accept:

• no flags — today's original behavior, unchanged: reads .env.pipeline, writes to data/ and the repo root. This is what ./scripts/run_pipeline.sh runs by default.
• --config <path> — same dotenv format as .env.pipeline, just a different file (e.g. a second OpenReplay app with different credentials).
• --job <path.json> — a self-contained JSON job spec instead of a dotenv file. This is the natural fit for Sentry (or any second app), since a job spec bundles connection details, which sources to run, and where to write output all in one file. See jobs/example.sentry.json and jobs/example.openreplay.json for templates — copy one, fill in real values, and it's gitignored automatically (only the example.*.json templates are tracked).

Two fields in a job spec (or their dotenv equivalents) control isolation and behavior per job:

• pipeline_sources — which fetchers run: any combination of openreplay_fe, cloudrun_be, sentry_fe, sentry_be. Determines which credentials are required, and whether the SSH tunnel preflight check runs at all (skipped entirely unless openreplay_fe is present).
• data_dir / docs_dir — where checkpoint.json/ error_master.json/test_journey_master.json/runs/ (data_dir) and ERROR_MASTER.md/TEST_JOURNEYS.md/FIX_TASKS.md (docs_dir) get written. Each job should point these at its own directory (e.g. data/<app-name>) so two jobs never collide on the same files. The default job (no flags) keeps writing to data/ and the repo root, same as before this multi-job support existed.

fix_tasks.py has no source-awareness of its own — it just reads whatever error_master.json lives at settings.data_dir, resolved the same --job/--config/default way. Always pass the same flag you used for the matching run.py call, or it'll silently operate on the wrong job's data instead of erroring.

Setup

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

OpenReplay + Cloud Run app (the default job)

Credentials live in .env.pipeline (gitignored). Copy .env.pipeline.example and fill in the real values if setting this up fresh.

cloudrun_be additionally needs working gcloud credentials with read access to Cloud Logging in the target GCP project:

gcloud auth login
gcloud auth application-default login

Sentry app (a job spec)

1. Copy jobs/example.sentry.json to a new file (e.g. jobs/<app-name>.json — anything under jobs/ other than the example.*.json templates is gitignored automatically).
2. From the Sentry dashboard, collect:
   • Org slug — the segment in your org's dashboard URL.
   • Auth token — Developer Settings → Personal Tokens → create one with scopes project:read, event:read, org:read. Do not use an "Org" token (see the Sentry section above — it's locked to a scope this pipeline can't use and will 403).
   • region_url — decode the base64 segment of the token you just created (between sntrys_/sntryu_ and the trailing signature) to find it, or check Settings → General Settings for a data-region field. Only needed if it isn't the default https://sentry.io.
   • FE/BE project slugs — from each project's URL in the dashboard.
3. Fill those into the job spec's sentry block, and set data_dir/ docs_dir to wherever you want this app's output isolated (e.g. data/<app-name>).
4. No gcloud/SSH tunnel setup needed for a Sentry-only job.

Running (manual trigger)

Default job (today's original OpenReplay + Cloud Run app) — a single run pulls both sources. Postgres, ClickHouse, and MinIO only run inside the OpenReplay k3s cluster (ClusterIP services) on the GCE VM behind replay.cosmaya.co.in — none are reachable directly, so a run needs an SSH tunnel up first; Cloud Logging just needs gcloud auth (see Setup above), no tunnel. Not run on a schedule yet — trigger it by hand when you want fresh data:

./scripts/run_pipeline.sh

This clears any stale remote kubectl port-forward from a previous interrupted run, opens a fresh SSH tunnel (Postgres 5432, ClickHouse 8123/9000, MinIO 9010), waits for it to become reachable, runs python pipeline/run.py, then always tears the tunnel down again (both locally and on the remote VM) whether the run succeeded or failed. Safe to run repeatedly on demand. pipeline/tunnel.py's preflight check (Postgres/ClickHouse/MinIO reachability) runs whenever openreplay_fe is one of the job's pipeline_sources — which it always is for the default job, so a Cloud-Run-only debugging run of the default job still needs the tunnel up. A job whose pipeline_sources omits openreplay_fe entirely (e.g. a cloudrun_be-only or Sentry job via --job/--config) skips the tunnel check completely instead.

If you'd rather manage the tunnel yourself (e.g. to run the pipeline manually a few times in a row, or to poke at the DBs directly), use ./scripts/open_tunnels.sh instead — it opens the same tunnel but leaves it running in the foreground until you Ctrl+C it, and then run python pipeline/run.py yourself in another terminal. pipeline/tunnel.py checks reachability first either way and fails fast with a remediation message if the tunnel isn't up.

Each run:

1. Phase 1: pulls sessions/errors from Postgres and ClickHouse since the last successful checkpoint (data/checkpoint.json). For each session in that batch, fetches its dom.mobs/dom.mobe files from MinIO and decodes them into ordered click/page/input steps (see above) — this is the only source of journey data on this instance.
2. Phase 2: pulls Cloud Run request logs + app logs for test-revizr since the last checkpoint, keeps request errors (status >= 500) and ERROR/CRITICAL-severity app log lines, and groups requests into IP+idle-gap journeys (see above).
3. Normalizes and de-duplicates errors from both phases against data/error_master.json, and journeys from both phases against data/test_journey_master.json — one canonical pair of master files regardless of source. Every error message string is passed through pipeline/redact.py first, which strips PEM key blocks and JSON blobs containing a private_key field (added after a real Cloud Run traceback captured a live service-account key verbatim — see the module docstring).
4. Writes data/runs/run_<timestamp>.json recording what was pulled and which entries were new vs. already known.
5. Writes up any not-yet-documented error/journey entries as prose in ERROR_MASTER.md / TEST_JOURNEYS.md (via a claude -p call — see below), and a per-run findings audit.
6. Only on success: saves both master files and advances the checkpoint. A failed run leaves the checkpoint and master files untouched, so it's safe to simply re-run.

Sentry job (or any other job spec / second dotenv config) — no tunnel to manage, so call pipeline/run.py directly instead of going through run_pipeline.sh:

source .venv/bin/activate
python pipeline/run.py --job jobs/<app-name>.json

(./scripts/run_pipeline.sh --job jobs/<app-name>.json also works — it detects the job doesn't use openreplay_fe and skips the SSH tunnel entirely — but is unnecessary overhead for a Sentry-only job.) Everything else behaves identically to the default job: incremental pulls via that job's own checkpoint.json, de-dup against that job's own master files, descriptive docs generated the same way, safe to re-run on failure. First run pulls the job spec's first_run_lookback_hours (default 168h/7 days) of history; every run after that only pulls what's new.

Descriptive docs (ERROR_MASTER.md / TEST_JOURNEYS.md)

The JSON master files are the deterministic source of truth for de-duplication; pipeline/docs.py adds a human-readable narrative layer on top, mirroring the DEFECTS.md/TEST_CONDITIONS.md style from the sibling observability_poc project. For any record not yet marked documented in the JSON (brand new this run, or left over from a run where the doc step failed), it calls claude -p once with the structured entries and asks for a prose write-up per entry, then appends those sections to ERROR_MASTER.md (## E-N: ...) / TEST_JOURNEYS.md (### TC-N — ...) — sections are never rewritten once written. A record is only marked documented: true after its section is successfully appended, so a claude failure just retries next run rather than losing the write-up. If nothing is pending, the claude call is skipped entirely (no cost, no findings audit written).

Generating fix tasks (FIX_TASKS.md)

python pipeline/fix_tasks.py                       # default job
python pipeline/fix_tasks.py --job jobs/<app-name>.json   # any other job

A separate, on-demand script — not run automatically by run.py/ run_pipeline.sh, and needs no SSH tunnel or gcloud auth, just that job's error_master.json and ERROR_MASTER.md. It has no source-awareness of its own — always pass the same --job/--config you used for the matching run.py call, or it'll silently read the wrong job's data instead of erroring (see "Jobs" above). For every error not yet marked fix_task_generated in the JSON, it calls claude -p once (one call per error, not batched, so one bad response only costs that error and each prompt can include a full traceback without truncation risk) and writes a ready-to-run task prompt into FIX_TASKS.md (## Fix Task — E-N: ..., with a fenced ```text block meant to be copy-pasted as claude -p "<task>" inside the source application's own repo — the frontend or test-revizr backend, not this observability repo). Tasks are written to be honest about thin data: if there's no stack trace, the task says the source location is unknown and describes what to investigate instead of guessing. fix_task_generated is independent of documented (one tracks ERROR_MASTER.md narrative, the other tracks FIX_TASKS.md), and behaves the same way: only set after its section is successfully written, so a claude failure just retries that one error next run.

Output artifacts

Paths below are for the default job (data_dir/docs_dir = data/ / repo root). Any other job's equivalents live under its own configured data_dir/docs_dir instead (e.g. data/<app-name>/error_master.json).

• data/error_master.json — canonical, deduplicated error registry across all runs (tracked in git).
• data/test_journey_master.json — canonical, deduplicated user-journey registry, meant as a growing test-case repository (tracked in git).
• ERROR_MASTER.md / TEST_JOURNEYS.md — human-readable narrative view generated from the above (tracked in git).
• FIX_TASKS.md — ready-to-run claude -p task prompts per error, generated on demand by pipeline/fix_tasks.py (tracked in git).
• data/runs/*.json — one run log per execution (gitignored).
• data/runs/*_findings.md — per-run findings audit, written only when something new was documented (gitignored).
• data/checkpoint.json — last-processed cursor per data source (gitignored, local mutable state).
• jobs/*.json — job spec files (gitignored, may contain real credentials); only the jobs/example.*.json templates are tracked.

Security: redacting secrets from ingested logs

pipeline/redact.py strips PEM private-key blocks and JSON blobs containing a private_key field out of every error message before it's stored or sent to claude -p. This exists because a real Cloud Run traceback once captured a full plaintext GCP service-account key (the app tried to load credentials from a value that resolved to a raw JSON string instead of a file path) — it reached data/error_master.json and a claude -p prompt before being caught and redacted by hand. It's applied at two points: where each source normalizes into NormalizedError (errors.py, backend_errors.py), and again in docs.py right before anything is serialized into an LLM prompt. It's a blunt, best-effort regex filter, not a substitute for fixing an app that logs credentials in the first place.