diff --git a/.serena/memories/audit-log.md b/.serena/memories/audit-log.md index 898e95e413..bc33a4a190 100644 --- a/.serena/memories/audit-log.md +++ b/.serena/memories/audit-log.md @@ -12,14 +12,15 @@ The system has: - **Three producers** (they create events): the Penpot backend (RPC layer), the Penpot frontend (browser app), and the Nitrate admin-console, which sends both server-side and client-side events. - **One central buffer**: the Postgres `audit_log` table in Penpot. It is temporary: it holds events for minutes/hours until they are archived, then they are deleted. -- **Four consumers** (they read events): webhooks, error reporters, telemetry, and the external archive in Nexus. -- **One long-term archive**: **Nexus**, a separate service that stores events for the long term. Archiving to **Nitrate** is in development and **does not exist on the current branch** (see section 11). +- **Four consumers** (they read events): webhooks, error reporters, telemetry, and the external archives in Nexus and in the admin console. +- **Two long-term archives**: **Nexus**, a separate service that stores events for the long term, and the **admin console** (Nitrate), which ingests its own allowlisted events through a second archive job (see section 12). ``` Penpot frontend ─ POST push-audit-events (public) ─────────▶┐ Nitrate admin-console (browser) ─ POST push-audit-events ──▶│ -Nitrate admin-console (server) ─ POST push-audit-events ──▶│ ─▶ audit_log table (buffer) ── cron every 5m ──▶ Nexus (archive) -Penpot backend (RPC wrap-audit) ─ submit ───────────────────┘ │ ├──▶ webhooks +Nitrate admin-console (server) ─ POST push-audit-events ──▶│ ─▶ audit_log table (buffer) ──┬─ cron 0s every 5m ──▶ Nexus (archive) +Penpot backend (RPC wrap-audit) ─ submit ───────────────────┘ │ └─ cron 30s every 5m ─▶ admin console (archive) + │ ├──▶ webhooks │ ├──▶ error reporters └──▶ telemetry (shadow rows) ``` @@ -44,7 +45,7 @@ Rule of thumb: if you need to know for sure "did this user do X?", look at `sour ## 3. The `audit_log` table: a buffer, not a store -The Penpot `audit_log` table **is not the final destination**. It is a waiting tray: events sit here until the archive cron sends them to Nexus and the cleanup cron (GC) deletes them. +The Penpot `audit_log` table **is not the final destination**. It is a waiting tray: events sit here until every active archiver (Nexus, admin console) confirms them, and whoever finishes last deletes the row. Main columns (migration `backend/src/app/migrations/sql/0054-add-audit-log-table.sql` and later ones): @@ -56,15 +57,16 @@ Main columns (migration `backend/src/app/migrations/sql/0054-add-audit-log-table - `source` text: `backend` / `frontend` for full rows, `telemetry:backend` / `telemetry:frontend` for anonymized copies (see section 10). - `ip_addr` inet: real IP (or `0.0.0.0` on telemetry rows). - `props` / `context` jsonb holding transit-encoded maps: action data and metadata (versions, browser, request-id...). -- `archived_at` timestamptz: NULL until Nexus confirms receipt; only then the row can be deleted. +- No `archived_at`: the `audit_log_pending` queue (migration `0157`, archiver `1` = Nexus, `2` = admin console, ids never reused) tracks what is left. Every record is born pending only for the archivers enabled at insert time (never a fixed both): cheaper on single-archiver hosts and true to the design doc. Consequence, accepted on purpose: rows born with no send flag on get no pendings, so enabling archiving later does not pick them up — the sweep deletes them unarchived. Indexes (each consumer has its own, so a slow consumer never blocks the others): -- Partial `created_at WHERE archived_at IS NULL`: used by the archive scan. -- Partial `archived_at WHERE archived_at IS NOT NULL`: used by the GC. +- `audit_log_pending` PK `(archiver, audit_log_id)`: each archiver owns its queue at constant cost, even with millions pending. +- `audit_log (created_at, id)` (migration `0157`): the chronological drain order (`ORDER BY created_at ASC, id ASC`; same total order for both archivers, so concurrent drains cannot deadlock). +- `audit_log_pending (audit_log_id)`: serves the "is anything still pending?" check of the final delete. - `(source, created_at)`: used by the telemetry scan. -Life of a row: `INSERT` → (every 5 min) archive sets `archived_at = now()` **only if Nexus answers 204**, in the same transaction → (every 5 min) GC runs `DELETE WHERE archived_at IS NOT NULL` **with no age filter**. So archive must run before GC, or data is deleted without being sent. +Life of a row: `INSERT` (row plus one pending per enabled archiver, same transaction; a rollback takes both) → (every 5 min, Nexus at second 0 and admin console at second 30) each archive job sends its own 128-row batches oldest-first and deletes its pendings in the same transaction → whoever deletes the last pending deletes the row. A batch the destination refuses rolls back and stays for the next tick. Pendings of archivers that are off are reaped at the start of each job run (covers later deactivations), and orphan rows are swept at the end of each run, so there is no separate GC task. The FK is plain `DEFERRABLE` (never `INITIALLY DEFERRED`): every deleter removes pendings before rows, including the telemetry task's own deletes. ## 4. Anatomy of an event @@ -271,17 +273,19 @@ Nexus dispatches on `:type`: `"telemetry-events"` batches are coerced and insert - **No personal data is expected.** Profile ids are random UUIDs with no PII, so the telemetry consumer needs no login guard; grouping is by `instance`, not by person (see posthog-telemetry below). - **The endpoint always answers 200.** Failures are swallowed and only counted (`nexus_telemetry_errors_total`), so a bad batch never breaks the sending instance; successes are counted (`nexus_telemetry_events_total`, `nexus_telemetry_reports_total`). A body the parser cannot read never reaches the handler: it answers 400 and counts in `nexus_http_malformed_requests_total`. Old instances may send a fressian+zstd base64 blob instead of a plain vector; Nexus decodes both (50 MiB cap against decompression bombs). -## 12. Archive to Nexus (today) and to Nitrate (future) +## 12. Archive to Nexus and to the admin console -### Nexus: how it works today +### How it works today -Nexus is the long-term store. Every 5 min the `:audit-log-archive` cron takes 128-row chunks of unarchived rows (`FOR UPDATE SKIP LOCKED`, ordered by `created_at`), POSTs them as transit `{:events [...]}` with `x-shared-key: "nexus "` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri, and marks `archived_at = now()` **only on HTTP 204**, in the same transaction. Anything else is retried next run; flag on but no URI raises `:task-not-configured`. It loops with 100 ms pauses until empty. +Every 5 min each archive job drains its own queue (`:audit-log-archive-nexus` on flag `:nexus`, cron `0 */5 * * * ?`; `:audit-log-archive-admin-console` on flag `:admin-console`, cron `30 */5 * * * ?`, 30s offset so the two drains never start on the same second), in 128-row chunks taken oldest-first (`ORDER BY created_at ASC, id ASC`) with `FOR UPDATE OF audit_log` (`SET LOCAL lock_timeout = '30s'`) inside the same transaction that sends and completes them, with 100 ms pauses between chunks. Batches arrive at Nexus in chronological order (it numbers by arrival for order-sensitive consumers). Nexus receives transit `{:events [...]}` with `x-shared-key: "nexus "` (`:nexus-shared-key`, or derived from the instance secret) and `origin` = public-uri; flag on but no URI raises `:task-not-configured`. The admin console receives only its allowlisted event names (enriched with team names and emails); anything filtered out counts as done for the admin-console queue. A chunk whose destination refuses it (Nexus non-204 raises `:archive-chunk-refused`) or throws (transport failure, admin-console error) rolls back and stays for the next tick, and the job aborts as failed: it archives what it can, and the cron retries the rest. Committed chunks stay sent. The send fns return counts or throw carrying the attempted count as `:events`; `run-single-archiver` records the `failed` metric from it, outside every chunk transaction (a local DB failure carries no `:events` and stays uncounted — it is not a destination failure). No `warn` before the throw: the runner reports the aborted job. Each job touches only its own queue, so a fallen Nexus never starves the admin console (nor vice versa). A chunk is **one POST** and the task advances only on 204, so a single oversized event wedges the whole queue: a crash report whose message embeds a whole payload (a serialized file, for example) can push the request past the proxy body limit and the rejection blocks every later event until it is pruned by hand. Frontend crash reports are therefore size-bounded at the source (`mem:frontend/routing-app-shell-subtleties`): `app.common.exceptions/get-hint` and the CLJS `format-throwable` clamp exception messages (`max-message-length`). On the other side, Nexus checks `origin` against its allowlist (403 if not allowed), coerces the events, and inserts them into its own `audit_log` (partitioned by `received_at`, with an incrementing sequence number and `source`). Watch the rename: Penpot's `created-at` is stored as **`received-at`** in Nexus; Nexus's `created-at` is when it was created there. `tracked-at` keeps the origin time. Telemetry batches arriving at `telemetry.penpot.app` land in the same table through the telemetry endpoint above. -Concurrent runs are safe: rows are locked with `FOR UPDATE SKIP LOCKED` inside the same transaction that sends and marks them, so two archivers running at once each take different rows. Only rows Nexus acknowledges (HTTP 204) get marked; anything else is retried on the next run. +Concurrent jobs are safe: both lock rows in `created_at ASC, id ASC` order (one total order, deliberately blocking — no `SKIP LOCKED`, so archive arrival stays chronological), so the second waits for the first to commit and then sees whether it was last. Only acknowledged work deletes pendings; an admin-console failure never blocks the Nexus send of the same rows (each job owns its queue). + +Observability (`backend/src/app/loggers/audit/archive_task.clj`, definitions in `backend/src/app/main.clj:default-metrics`): one `dbg` summary per job run (`:archiver` label, sent/skipped/chunks, reaped disabled pendings, swept orphans, `:elapsed` as `"1.5s"` via `ct/format-duration`, frozen in a strict binding first because log props evaluate lazily in the logger thread), one `dbg` per admin-console send call (`ready` with totals, `sent`/`failed` with counts), plus `dbg` when the queue is empty or the job's send flag is off. Queue depth is Grafana's business: it queries the database directly, so the run performs no `GROUP BY` for the logs. No `warn` before a throw: the runner reports the aborted job (refusal details travel in the error data). Metrics, all recorded with bounded labels only: `penpot_audit_archive_events_total{ destination = nexus|admin-console, outcome = archived|failed|skipped }` (`skipped` = filtered by the admin-console allowlist) and `penpot_audit_archive_discarded_total{ reason = reaped|swept }` (entries discarded without archiving: `reaped` counts pendings dropped for disabled archivers, `swept` counts rows deleted while nobody waited for them — the silent R3 loss, now dashboard-visible instead of log-only). Timing needs no custom histogram: the jobs substrate already records execution time per job (`jobs-execution-timing`, split by job name), so each archive job is timed as itself. No queue-depth gauge: pending depth is read off the replica database directly. The bulk frontend ingest counts `penpot_audit_ingest_events_total{ kind = full|shadow, outcome = inserted|failed }` (both faces share one transaction, so a failure drops the batch on both). Archive success counts are recorded by `run-single-archiver` on the drain totals: a chunk that returns normally committed its transaction (a rollback always throws instead), so they cannot inflate. Logs never carry event data (`props`, `context`, `profile-id`, `ip-addr`, `team-name`, emails): only counts, statuses and the uri host. ### How Nexus consumes the data (and forwards to PostHog) @@ -291,20 +295,16 @@ Stored rows are not the end: a consumer pipeline reads them back in sequence ord 2. **`posthog`** (runs after `local`): forwards our own SaaS archived events to PostHog. It sends an allowlist of backend track events (teams, members, projects, files, comments, Nitrate/SSO, ...) plus frontend events minus a noise exclusion list, grouped by **team**, with person identify data from the local profile mirror and snake_case props. Both paths include the `initiator` property when the stored context has one, so PostHog can split Penpot vs admin-console traffic. 3. **`posthog-telemetry`**: forwards events with `source = telemetry:frontend|telemetry:backend` to a separate PostHog project. Because these rows are already anonymous (no IPs, truncated times, filtered props), grouping is by **instance** (the self-hosted server, from `context.instance-id`), with one `$groupidentify` event per instance per chunk so PostHog learns per-instance properties (version, URI, flags). Each consumer has its own PostHog credentials. -So the full journey is: Penpot buffer → Nexus `audit_log` (via archive cron or telemetry endpoint) → consumers → PostHog (SaaS events grouped by team, telemetry grouped by instance) plus local analytics views and materialized KPIs. +So the full journey is: Penpot buffer → Nexus `audit_log` (via either archive cron or the telemetry endpoint) → consumers → PostHog (SaaS events grouped by team, telemetry grouped by instance) plus local analytics views and materialized KPIs. -### Nitrate: what exists and what is planned +### Admin console: the second archiver -Today **only Nexus archives**. Archiving Penpot → Nitrate is in development and not on this branch. - -What already exists (do not confuse with archiving): +Penpot → admin-console archiving lives in its own job (`:audit-log-archive-admin-console`, archiver `2`, flag `:admin-console`, params closed empty): it drains its own pending queue and ingests allowlisted events through `nitrate/call :ingest-audit-log`, enriched as described above. What also exists (do not confuse with archiving): - **Nitrate → Penpot**: `POST /api/management/methods/push-audit-events` (`backend/src/app/rpc/management/nitrate.clj`) lets the trusted Nitrate backend inject events into the Penpot audit log (no RPC auth, `::audit/skip`, request context + `submit`). Section 7a is the main user. - **Admin-console browser → collector**: usage events from the admin-console UI traveling through the normal public `push-audit-events` (section 7b). - **Penpot frontend Nitrate bits**: membership/subscription events (`frontend/src/app/main/data/nitrate.cljs`, `nitrate-audit/*`) through the same public endpoint. -The future Penpot → Nitrate archive will follow the Nexus pattern (cron reading the buffer and POSTing batches), but there is no contract or flag on this branch yet. When it lands, document here: endpoint, auth, format, which `source` values travel, and how it shares the GC. - ## 13. Flags and config Flags (`common/src/app/common/flags.cljc`, varia; enabled with `PENPOT_FLAGS=enable-`): @@ -312,8 +312,8 @@ Flags (`common/src/app/common/flags.cljc`, varia; enabled with `PENPOT_FLAGS=ena | Flag | What it does | |---|---| | `:audit-log` | Stores full rows; turns on `wrap-audit` and frontend ingest | -| `:audit-log-archive` | Turns on the cron that sends to Nexus (needs `:audit-log-archive-uri`) | -| `:audit-log-gc` | Turns on the cron that deletes archived rows | +| `:nexus` | Send flag of the Nexus archive job `:audit-log-archive-nexus` (cron `0 */5 * * * ?`; needs `:audit-log-archive-uri`) | +| `:admin-console` | Send flag of the admin-console archive job `:audit-log-archive-admin-console` (cron `30 */5 * * * ?`, 30s after Nexus); also the module flag for the admin-console integration | | `:audit-log-logger` | Structured `app.audit` log per event | | `:error-reporting` | Starts the error reporters (off by default); database reporter on flag alone, Mattermost on flag plus webhook URL | | `:telemetry` | Stores shadow rows and turns on shipping (auto with `:telemetry-enabled`) | @@ -337,6 +337,6 @@ Config (`backend/src/app/config.clj`): `:audit-log-archive-uri`, `:telemetry-uri - Penpot frontend producer: `frontend/src/app/main/data/event.cljs` (collector, buffer, send), `frontend/src/app/main/repo.cljs` + `frontend/src/app/util/http.cljs` (headers), `frontend/src/app/main/data/nitrate.cljs` (Nitrate membership events). - Admin-console producer (private repo, no paths cited here): server-side organization/subscription events through the management `push-audit-events`, browser-side UI events through the public `push-audit-events`. - Consumers: `backend/src/app/loggers/webhooks.clj`, `backend/src/app/loggers/database.clj`, `backend/src/app/loggers/mattermost.clj`, `backend/src/app/tasks/telemetry.clj`. -- Archive/GC: `backend/src/app/loggers/audit/archive_task.clj`, `backend/src/app/loggers/audit/gc_task.clj`, wiring and crons in `backend/src/app/main.clj`. +- Archive (two jobs sharing one namespace + pending queue, no GC): `backend/src/app/loggers/audit/archive_task.clj` (`:audit-log-archive-nexus` / `:audit-log-archive-admin-console` job-defs, shared drain/reap/sweep helpers), producer hook in `backend/src/app/loggers/audit.clj` (`append-audit-entry`, single `queue-pendings` helper) plus the bulk ingest in `backend/src/app/rpc/commands/audit.clj` (`handle-events`), metrics/cron wiring in `backend/src/app/main.clj` (one cron entry per job, Nexus at second 0 and admin console at second 30, each gated by its own send flag), single migration `0157` (queue, drop of `archived_at`, covering `(created_at, id)` index for the chronological scan; no backfill — pre-existing rows are swept unsent, an accepted loss). - Nexus (private repo, no paths cited here): archive ingest, ordered consumers (local profile mirror, PostHog SaaS events, PostHog telemetry), telemetry endpoint, analytics views. - Tests: `backend_tests/rpc-audit-test.clj`, `rpc-team-test/accept-organization-invitation-audit-event`. diff --git a/backend/src/app/loggers/audit.clj b/backend/src/app/loggers/audit.clj index 7c77a710d7..619c367a49 100644 --- a/backend/src/app/loggers/audit.clj +++ b/backend/src/app/loggers/audit.clj @@ -21,6 +21,7 @@ [app.http :as-alias http] [app.http.access-token :as-alias actoken] [app.jobs :as jobs] + [app.loggers.audit.archive-task :as archive] [app.loggers.audit.tasks :as-alias tasks] [app.loggers.webhooks :as-alias webhooks] [app.rpc :as-alias rpc] @@ -241,8 +242,13 @@ (update :props db/tjson) (update :context db/tjson) (update :ip-addr db/inet)) - params (select-keys params event-keys)] - (db/insert! cfg :audit-log params))) + params (select-keys params event-keys) + row (db/insert! cfg :audit-log params)] + ;; The archive queue follows the row in the same transaction: one + ;; pending per enabled archiver. A rollback takes both on its way + ;; out, so a row never outlives its queue. + (archive/queue-pendings cfg [(:id row)]) + row)) (def ^:private xf:filter-telemetry-props "Transducer that keeps only map entries whose values are UUIDs, @@ -582,6 +588,8 @@ logic." [cfg event] (when (contains? cf/flags :audit-log) + ;; Transactional (not a bare connection): the row and its + ;; pendings commit or vanish together (see append-audit-entry). (let [tnow (ct/now) event (-> event (assoc :created-at tnow) @@ -592,4 +600,4 @@ (update :source d/nilv "backend") (select-keys event-keys) (check-event))] - (db/run! cfg append-audit-entry event)))) + (db/tx-run! cfg append-audit-entry event)))) diff --git a/backend/src/app/loggers/audit/archive_task.clj b/backend/src/app/loggers/audit/archive_task.clj index 6c29c29d6b..a4eb14e5e6 100644 --- a/backend/src/app/loggers/audit/archive_task.clj +++ b/backend/src/app/loggers/audit/archive_task.clj @@ -9,20 +9,77 @@ [app.common.exceptions :as ex] [app.common.logging :as l] [app.common.schema :as sm] + [app.common.time :as ct] [app.common.transit :as t] [app.common.uri :as u] [app.common.uuid :as uuid] [app.config :as cf] [app.db :as db] + [app.db.sql :as sql] [app.http.client :as http] [app.jobs :as jobs] + [app.metrics :as mtx] [app.nitrate :as nitrate] [app.setup :as-alias setup] [integrant.core :as ig] [promesa.exec :as px])) -;; This is a task responsible to send the accumulated events to -;; external service for archival. +;; This task drains the audit log buffer into the long-term archives. +;; +;; Two separate jobs share this namespace and the queue helpers, one +;; per destination: `:audit-log-archive-nexus` and +;; `:audit-log-archive-admin-console`. Each job drains only its own queue +;; and deletes its pendings on completion; whoever deletes the last +;; pending also deletes the record ("last one out turns off the +;; light"), so no separate GC task is needed. Housekeeping (reaping +;; disabled pendings, sweeping orphans) lives in both jobs on purpose: +;; only the enabled job runs, so it must clean up for the rest. + +;; Archiver ids are fixed constants, never reuse one: retired +;; pendings are reaped by id, so a reused id would inherit the +;; previous owner's queue. +(def archiver-nexus 1) +(def archiver-admin-console 2) + +(def ^:private archivers + [{:id archiver-nexus :flag :nexus :label "nexus"} + {:id archiver-admin-console :flag :admin-console :label "admin-console"}]) + +(def ^:private xf:enabled-archivers + "Static transducer: keep the archivers whose send flag is on. + Allocated once so hot paths (every audited RPC) do not rebuild + it per call; the flag lookup itself still reads the live flags." + (comp (filter #(contains? cf/flags (:flag %))) + (map :id))) + +(defn enabled-archivers + "Archivers allowed to work right now, from the send flags. Public: + producers queue every record for exactly these archivers." + [] + (into [] xf:enabled-archivers archivers)) + +(defn queue-pendings + "Queue records for every enabled archiver, nothing when there is + nothing to queue or no archiver is on. One multi-row statement for + all ids instead of one per archiver: this runs inside every audited + RPC and the bulk frontend ingest. Runs inside the row transaction: + a rollback takes both." + [cfg ids] + (let [archivers (enabled-archivers)] + (when (and (seq ids) (seq archivers)) + (db/insert-many! cfg :audit-log-pending + [:audit-log-id :archiver] + (for [id ids + archiver archivers] + [id archiver]) + {::sql/on-conflict-do-nothing true})))) + +(def ^:private archiver-labels + (into {} (map (juxt :id :label)) archivers)) + +(defn- archiver-label + [archiver] + (get archiver-labels archiver)) (defn- decode-row [{:keys [props ip-addr context] :as row}] @@ -51,29 +108,95 @@ [row] (select-keys row event-keys)) -(defn- send! +(defn- record-count + [metrics id labels amount] + (mtx/run! metrics :id id :labels labels :inc amount)) + +(defn- truncate-body + [body limit] + (when (some? body) + (let [s (str body)] + (if (> (count s) limit) + (subs s 0 limit) + s)))) + +;; NOTE: deliberately no SKIP LOCKED here. Batches must reach the +;; destinations in chronological order (Nexus numbers by arrival), +;; so every archiver takes the same total order and a second +;; overlapping run waits for the first to commit, then sees whether +;; it was last. The brief blocking is accepted; skipping rows could +;; reorder archive arrival. +(def ^:private sql:pending-chunk + "SELECT a.* + FROM audit_log_pending p + JOIN audit_log a ON a.id = p.audit_log_id + WHERE p.archiver = ?::smallint + ORDER BY a.created_at ASC, a.id ASC + LIMIT 128 + FOR UPDATE OF a") + +(def ^:private sql:delete-pendings + "DELETE FROM audit_log_pending + WHERE archiver = ?::smallint + AND audit_log_id = ANY(?)") + +(def ^:private sql:delete-drained + "DELETE FROM audit_log a + WHERE a.id = ANY(?) + AND NOT EXISTS ( + SELECT 1 FROM audit_log_pending p + WHERE p.audit_log_id = a.id + )") + +(defn- get-pending-rows + [{:keys [::db/conn] :as cfg} archiver] + (->> (db/exec! conn [sql:pending-chunk archiver]) + (not-empty))) + +(defn- complete-batch + "Delete our pendings for the batch, then the records nobody is + waiting for anymore. Two statements on purpose: inside a single + statement the NOT EXISTS would still see the pending just deleted + and never delete a record." + [{:keys [::db/conn]} archiver rows] + (let [ids (db/create-array conn "uuid" (map :id rows))] + (db/exec! conn [sql:delete-pendings archiver ids]) + (db/exec! conn [sql:delete-drained ids]))) + +(defn- send-to-nexus + "POST events to Nexus. Returns the number of events sent; throws on + transport failure or non-204 refusal, carrying the attempted count + as :events in the error data so the caller records the failure. + No metrics or logs here: the runner reports the aborted job." [{:keys [::uri] :as cfg} events] - (let [skey (-> cfg ::setup/shared-keys :nexus) - body (t/encode {:events events}) - headers {"content-type" "application/transit+json" - "origin" (str (cf/get :public-uri)) - "x-shared-key" (str "nexus " skey)} - params {:uri uri - :timeout 12000 - :method :post - :headers headers - :body body} - resp (http/req cfg params {:skip-ssrf-check? true})] - + (let [n (count events) + skey (-> cfg ::setup/shared-keys :nexus) + body (t/encode {:events events}) + params {:uri uri + :timeout 12000 + :method :post + :headers {"content-type" "application/transit+json" + "origin" (str (cf/get :public-uri)) + "x-shared-key" (str "nexus " skey)} + :body body} + resp (try + (http/req cfg params {:skip-ssrf-check? true}) + (catch Throwable cause + (throw (ex/raise :type :internal + :code :nexus-send-failed + :hint "unable to send events to nexus" + :cause cause + :events n))))] (if (= (:status resp) 204) - true - (do - (l/error :hint "unable to archive events" - :resp-status (:status resp) - :resp-body (:body resp)) - false)))) + n + (throw (ex/raise :type :internal + :code :archive-chunk-refused + :hint "nexus refused archive chunk" + :resp-status (:status resp) + :resp-body (truncate-body (:body resp) 500) + :events n))))) -(def ^:private event-names-for-nitrate +(def ^:private event-names-for-admin-console #{"accept-organization-invitation" "accept-team-invitation" "accept-team-invitation-from" @@ -121,7 +244,7 @@ "update-team-photo" "verify-token"}) -;; Props id key -> email prop filled from profile.email (nitrate path only). +;; Props id key -> email prop filled from profile.email (admin-console path only). ;; Some producers store ids as UUID strings (e.g. :user-who-send-invitation). (def ^:private id->email-prop {:profile-id :profile-email @@ -187,8 +310,8 @@ props id->email-prop)))) -(defn- enrich-events-for-nitrate - "Enrich allowlisted events before nitrate ingest: +(defn- enrich-events-for-admin-console + "Enrich allowlisted events before the admin-console ingest: 1. Fill missing :team-name from team id (including soft-deleted). 2. Fill missing emails from matching id keys (including soft-deleted profiles): :profile-email, :user-email, :member-email, @@ -204,103 +327,304 @@ (maybe-assoc-emails emails))) events))) -(defn- send-to-nitrate! +(defn- send-to-admin-console + "Ingest rows into the admin console. Returns {:sent :skipped} + (filtered-out rows count as processed for the queue); throws + carrying the attempted count as :events, like send-to-nexus. Enrich + failures propagate untouched: a local DB/decode failure is ours, not + the destination's, and must not count as a send failure." [cfg rows] - (when-let [events (->> rows - (filterv #(contains? event-names-for-nitrate (:name %))) - (not-empty))] - (nitrate/call cfg :ingest-audit-log - {:events (enrich-events-for-nitrate cfg events)}))) + (let [events (filterv #(contains? event-names-for-admin-console (:name %)) rows) + total (count rows) + skipped (- total (count events)) + uri-host (some-> (cf/get :admin-console-uri) u/uri u/get-domain)] + (l/dbg :hint "archive events chunk ready" + :archiver "admin-console" + :uri-host uri-host + :total total + :skipped skipped) + (if (empty? events) + {:sent 0 :skipped skipped} + ;; Enrich first, outside the try: a local DB/decode failure is + ;; ours, not the destination's, and must not count as + ;; admin-console/failed. + (let [enriched (enrich-events-for-admin-console cfg events) + n (count events)] + (try + (nitrate/call cfg :ingest-audit-log {:events enriched}) + (l/dbg :hint "archive events chunk sent" + :archiver "admin-console" + :uri-host uri-host + :sent n + :skipped skipped) + {:sent n :skipped skipped} + (catch Throwable cause + (l/dbg :hint "archive events chunk failed" + :archiver "admin-console" + :uri-host uri-host + :events n + :code :admin-console-send-failed) + (throw (ex/raise :type :internal + :code :admin-console-send-failed + :hint "unable to archive events to the admin console" + :cause cause + :events n)))))))) -(defn- mark-archived! - [{:keys [::db/conn]} rows] - (let [ids (db/create-array conn "uuid" (map :id rows))] - (db/exec-one! conn ["update audit_log set archived_at=now() where id = ANY(?)" ids]))) +(defn- archive-nexus-chunk + "Send one pending chunk to Nexus and complete it. Returns {:sent n} + or nil when the queue is empty; throws on refusal. Takes a + transactional cfg: the caller opens one transaction per chunk. No + metrics here: a normal return means the transaction committed, so + the caller records the success." + [cfg] + (db/exec! cfg ["SET LOCAL lock_timeout = '30s'"]) + (when-let [rows (get-pending-rows cfg archiver-nexus)] + (let [events (mapv row->event (mapv decode-row rows)) + ;; send-to-nexus returns the count or throws: a refusal + ;; aborts the job, the rows stay for the next tick. + n (send-to-nexus cfg events)] -(def ^:private sql:get-audit-log-chunk - "SELECT * - FROM audit_log - WHERE archived_at IS NULL - ORDER BY created_at ASC - LIMIT 128 - FOR UPDATE - SKIP LOCKED") + (l/trc :hint "archive events chunk" + :archiver "nexus" + :events n) -(defn- get-event-rows - [{:keys [::db/conn] :as cfg}] - (->> (db/exec! conn [sql:get-audit-log-chunk]) - (not-empty))) + (complete-batch cfg archiver-nexus rows) + {:sent n}))) -(defn- archive-events! - [{:keys [::uri] :as cfg}] - (db/tx-run! cfg (fn [cfg] - (when-let [rows (get-event-rows cfg)] - (let [decoded (mapv decode-row rows) - events (mapv row->event decoded)] - (l/trc :hint "archive events chunk" :uri uri :events (count events)) +(defn- archive-admin-console-chunk + "Send one pending chunk to the admin console and complete it. Returns + {:sent :skipped} or nil when the queue is empty; throws on failure. + Takes a transactional cfg, no metrics here: same contract as + archive-nexus-chunk." + [cfg] + (db/exec! cfg ["SET LOCAL lock_timeout = '30s'"]) + (when-let [rows (get-pending-rows cfg archiver-admin-console)] + (let [decoded (mapv decode-row rows) + _ (l/trc :hint "archive events chunk" + :archiver "admin-console" + :events (count decoded)) + admin-console-res (send-to-admin-console cfg decoded)] + ;; Filtered-out rows count as processed for the + ;; admin-console queue: their pending is done, the + ;; record stays until every archiver is done. + (complete-batch cfg archiver-admin-console rows) + admin-console-res))) - (when (contains? cf/flags :admin-console) - (send-to-nitrate! cfg decoded)) +(defn- drain-archiver + "Drain one archiver queue chunk by chunk, one transaction per chunk. + Returns totals; a chunk the destination refuses (or that throws) + aborts the drain with a throw, and the committed chunks so far + stay sent." + [cfg send-chunk] + (loop [sent 0 + skipped 0 + chunks 0] + ;; One transaction per chunk: fetch, send and complete commit (or + ;; roll back) together, and the next chunk sees a fresh snapshot. + (if-let [res (db/tx-run! cfg send-chunk)] + (do + (jobs/heartbeat cfg) + (px/sleep 100) + (recur (+ sent ^long (:sent res)) + (+ skipped ^long (:skipped res 0)) + (inc chunks))) + {:sent sent :skipped skipped :chunks chunks}))) - ;; When :nexus is off, treat Nexus as skipped - ;; success so nitrate-only still marks the chunk. - ;; REPL :enabled true with neither send flag also marks - ;; with no outbound send — always set at least one flag. - (let [nexus-ok? (if (contains? cf/flags :nexus) - (send! cfg events) - true)] - (when nexus-ok? - (mark-archived! cfg rows) - (count events)))))))) +(defn- sql:reap-pendings + [archiver] + (str "DELETE FROM audit_log_pending" + " WHERE archiver = " archiver + " AND audit_log_id IN (SELECT audit_log_id FROM audit_log_pending" + " WHERE archiver = " archiver + " ORDER BY audit_log_id LIMIT 5000)")) -(declare execute-audit-log-archive) +(def ^:private sql:sweep-orphans + "DELETE FROM audit_log a + WHERE a.id IN (SELECT a2.id FROM audit_log a2 + WHERE NOT EXISTS (SELECT 1 FROM audit_log_pending p + WHERE p.audit_log_id = a2.id) + ORDER BY a2.id LIMIT 5000)") -(def schema:audit-log-archive-params - "Optional overrides for the repl invocation defaults. +(defn- delete-batched + "Run a batched DELETE until it removes nothing. Returns the total." + [cfg sql] + (loop [total 0] + (let [n (db/get-update-count (db/exec-one! cfg [sql]))] + (if (pos? n) + (do + (px/sleep 100) + (recur (+ total ^long n))) + total)))) + +(defn- reap-disabled + "Drop the pendings of archivers that are off: a disabled destination + will never drain them, and without this they would pin their records + forever. This also absorbs any later deactivation, with no manual step. + Dropped pendings count as discarded: their rows may still be archived + by an enabled archiver, or swept unsent when nothing waits for them." + [cfg] + (let [total (->> archivers + (remove #(contains? cf/flags (:flag %))) + (map #(delete-batched cfg (sql:reap-pendings (:id %)))) + (reduce + 0))] + (when (pos? total) + (record-count (mtx/instance cfg) :audit-archive-discarded ["reaped"] total)) + total)) + +(defn- sweep-orphans + "Delete records nobody waits for anymore: rows born while no + archiver was on, rows left pending-less by the reap, and leftovers + archived by the old task. Batched and normally a no-op. Deleted rows + count as discarded: on a live system they were never archived." + [cfg] + (let [total (delete-batched cfg sql:sweep-orphans)] + (when (pos? total) + (record-count (mtx/instance cfg) :audit-archive-discarded ["swept"] total)) + total)) + +(declare execute-audit-log-archive-nexus) +(declare execute-audit-log-archive-admin-console) + +(def schema:audit-log-archive-nexus-params + "Overrides for the nexus job repl invocation defaults. Cron submits {}; system deps (pool, shared-keys, http) live on the - job-def cfg closed over by the handler, not in job params." - [:map - [:enabled {:optional true} :boolean] + job-def cfg closed over by the handler, not in job params. + Closed on purpose: a stale key (like the old :enabled) must fail + validation loudly instead of skipping silently." + [:map {:closed true} [:uri {:optional true} ::sm/uri]]) -(defmethod ig/init-key ::job-def +(def schema:audit-log-archive-admin-console-params + "Params of the admin-console job: none. Cron submits {}; system deps + (pool, admin-console client) live on the job-def cfg closed over by + the handler, not in job params. Closed on purpose, same as the nexus + schema." + [:map {:closed true}]) + +(defmethod ig/assert-key ::nexus-job-def [_ cfg] - {::jobs/name :audit-log-archive - ::jobs/schema schema:audit-log-archive-params + (assert (db/pool? (::db/pool cfg)) "expected a valid database pool") + (assert (http/client? (::http/client cfg)) "expected a valid http client") + (assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance")) + +(defmethod ig/init-key ::nexus-job-def + [_ cfg] + ;; Effective destination, once at boot: per-run logs carry counts, + ;; the runner says the job started, this says where it sends. + (l/dbg :hint "audit log archive job initialized" + :archiver "nexus" + :uri-host (some-> (cf/get :audit-log-archive-uri) u/uri u/get-domain)) + {::jobs/name :audit-log-archive-nexus + ::jobs/schema schema:audit-log-archive-nexus-params ::jobs/handler (fn [_context params] - (execute-audit-log-archive cfg params)) - ::jobs/decoder (sm/decoder schema:audit-log-archive-params sm/json-transformer) - ::jobs/validator (sm/validator schema:audit-log-archive-params)}) + (execute-audit-log-archive-nexus cfg params)) + ::jobs/decoder (sm/decoder schema:audit-log-archive-nexus-params sm/json-transformer) + ::jobs/validator (sm/validator schema:audit-log-archive-nexus-params)}) -(defn execute-audit-log-archive - "Plain job handler: archive the accumulated audit events in chunks +(defmethod ig/assert-key ::admin-console-job-def + [_ cfg] + ;; No assert on the admin-console client: it inits to nil without the + ;; :admin-console flag, and the job skips on its flag anyway. + (assert (db/pool? (::db/pool cfg)) "expected a valid database pool") + (assert (mtx/metrics? (::mtx/metrics cfg)) "expected a valid metrics instance")) + +(defmethod ig/init-key ::admin-console-job-def + [_ cfg] + ;; Same as the nexus job-def: destination host once at boot. The + ;; ingest uri is resolved by nitrate/call at send time (it raises + ;; when missing); here it is display-only. + (l/dbg :hint "audit log archive job initialized" + :archiver "admin-console" + :uri-host (some-> (cf/get :admin-console-uri) u/uri u/get-domain)) + {::jobs/name :audit-log-archive-admin-console + ::jobs/schema schema:audit-log-archive-admin-console-params + ::jobs/handler + (fn [_context params] + (execute-audit-log-archive-admin-console cfg params)) + ::jobs/decoder (sm/decoder schema:audit-log-archive-admin-console-params sm/json-transformer) + ::jobs/validator (sm/validator schema:audit-log-archive-admin-console-params)}) + +(defn- run-single-archiver + "Drain one archiver queue, then housekeeping (reap disabled + pendings, sweep orphans) and the per-run summary. Housekeeping lives + in both jobs: only the enabled job runs, so it must reap the queues + of the archivers that are off; both halves are idempotent deletes. + A refused chunk (or a throw) aborts the job: the committed chunks + stay sent, the rest waits for the next tick, and the cron retries. + Every effect is rows and logs (drain metrics land here, outside + every chunk transaction)." + [cfg archiver send-chunk tpoint] + (let [label (archiver-label archiver) + metrics (mtx/instance cfg) + reaped (reap-disabled cfg) + ;; All drain metrics land here, outside every chunk + ;; transaction. Success is recorded on the totals: a normal + ;; chunk return means its transaction committed (a rollback + ;; always throws, taking the failure path instead). The send + ;; fns carry the attempted count as :events; anything else (a + ;; local DB failure) is not a destination failure and stays + ;; uncounted. The throw keeps rolling: a failed chunk aborts + ;; the job, the cron retries. + drained (try + (drain-archiver cfg send-chunk) + (catch Throwable cause + (let [n (:events (ex-data cause))] + (when (and (integer? n) (pos? n)) + (record-count metrics :audit-archive-events [label "failed"] n))) + (throw cause))) + _ (let [{:keys [sent skipped]} drained] + (when (pos? sent) + (record-count metrics :audit-archive-events [label "archived"] sent)) + (when (pos? skipped) + (record-count metrics :audit-archive-events [label "skipped"] skipped))) + swept (sweep-orphans cfg) + sent (:sent drained) + skipped (:skipped drained) + chunks (:chunks drained) + ;; Strict binding on purpose: log props evaluate lazily in + ;; the logger thread, so the elapsed must be frozen here. + elapsed (ct/format-duration (tpoint))] + (if (and (zero? sent) + (zero? skipped) + (zero? reaped) + (zero? swept)) + (l/dbg :hint "no pending audit events" + :archiver label) + (l/dbg :hint "events archived" + :archiver label + :elapsed elapsed + :sent sent + :skipped skipped + :chunks chunks + :reaped reaped + :swept swept)))) + +(defn execute-audit-log-archive-nexus + "Nexus job handler: drain the nexus queue in chunks, then clean up (heartbeat per iteration: the sent chunk batches can be long)." [cfg params] ;; NOTE: this let allows overwrite default configured values from - ;; the repl, when manually invoking the task. Prefer setting a send - ;; flag (:admin-console and/or :nexus); :enabled alone - ;; marks chunks without shipping. - (let [enabled (or (contains? cf/flags :nexus) - (contains? cf/flags :admin-console) - (:enabled params false)) - uri (cf/get :audit-log-archive-uri) - uri (or uri (:uri params)) - ;; Normalize to an uri object; params may carry a plain string - ;; on direct invocations (validation only applies on submit!). - uri (u/uri uri) - cfg (assoc cfg ::uri uri)] - (when (and (contains? cf/flags :nexus) - (not uri)) - (ex/raise :type :internal - :code :task-not-configured - :hint "archive task not configured, missing uri")) - (when enabled - (loop [total 0] - (if-let [n (archive-events! cfg)] - (do - (jobs/heartbeat cfg) - (px/sleep 100) - (recur (+ total ^long n))) - (when (pos? total) - (l/dbg :hint "events archived" :total total))))))) + ;; the repl, when manually invoking the task. + (if-not (contains? cf/flags :nexus) + (l/dbg :hint "audit log archive (nexus) skipped, flag off") + (let [uri (cf/get :audit-log-archive-uri) + uri (or uri (:uri params)) + ;; Normalize to an uri object; params may carry a plain string + ;; on direct invocations (validation only applies on submit!). + uri (u/uri uri) + cfg (assoc cfg ::uri uri)] + (when-not uri + (ex/raise :type :internal + :code :task-not-configured + :hint "archive task not configured, missing uri")) + (run-single-archiver cfg archiver-nexus archive-nexus-chunk (ct/tpoint))))) + +(defn execute-audit-log-archive-admin-console + "Admin-console job handler: drain the admin-console queue in chunks, + then clean up (heartbeat per iteration: the sent chunk batches can + be long)." + [cfg _params] + (if-not (contains? cf/flags :admin-console) + (l/dbg :hint "audit log archive (admin-console) skipped, flag off") + (run-single-archiver cfg archiver-admin-console archive-admin-console-chunk (ct/tpoint)))) diff --git a/backend/src/app/loggers/audit/gc_task.clj b/backend/src/app/loggers/audit/gc_task.clj deleted file mode 100644 index 65cd5bec7f..0000000000 --- a/backend/src/app/loggers/audit/gc_task.clj +++ /dev/null @@ -1,46 +0,0 @@ -;; This Source Code Form is subject to the terms of the Mozilla Public -;; License, v. 2.0. If a copy of the MPL was not distributed with this -;; file, You can obtain one at http://mozilla.org/MPL/2.0/. -;; -;; Copyright (c) KALEIDOS SUBSIDIARY SL - -(ns app.loggers.audit.gc-task - (:require - [app.common.logging :as l] - [app.common.schema :as sm] - [app.db :as db] - [app.jobs :as jobs] - [integrant.core :as ig])) - -(def ^:private sql:clean-archived - "DELETE FROM audit_log - WHERE archived_at IS NOT NULL") - -(defn- clean-archived! - [{:keys [::db/pool]}] - (let [result (db/exec-one! pool [sql:clean-archived]) - result (db/get-update-count result)] - (l/debug :hint "delete archived audit log entries" :deleted result) - result)) - -(declare execute-audit-log-gc) - -(def schema:audit-log-gc-params - "Params map (no params needed; config-derived only)." - [:map {:closed true}]) - -(defmethod ig/init-key ::job-def - [_ cfg] - {::jobs/name :audit-log-gc - ::jobs/schema schema:audit-log-gc-params - ::jobs/handler - (fn [_context params] - (execute-audit-log-gc cfg params)) - ::jobs/decoder (sm/decoder schema:audit-log-gc-params sm/json-transformer) - ::jobs/validator (sm/validator schema:audit-log-gc-params)}) - -(defn execute-audit-log-gc - "Plain job handler: delete the archived audit log entries." - ([cfg] (execute-audit-log-gc cfg nil)) - ([cfg _params] - (clean-archived! cfg))) diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 5ee2c7bbe2..5a387f14be 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -371,6 +371,24 @@ :http-malformed-requests-total {::mdef/name "penpot_http_malformed_requests_total" ::mdef/help "Total number of requests whose body the parser could not read." + ::mdef/type :counter} + + :audit-archive-events + {::mdef/name "penpot_audit_archive_events_total" + ::mdef/help "Audit log events sent by the archive task, by destination and outcome." + ::mdef/labels ["destination" "outcome"] + ::mdef/type :counter} + + :audit-archive-discarded + {::mdef/name "penpot_audit_archive_discarded_total" + ::mdef/help "Audit log entries discarded without archiving, by reason." + ::mdef/labels ["reason"] + ::mdef/type :counter} + + :audit-ingest-events + {::mdef/name "penpot_audit_ingest_events_total" + ::mdef/help "Audit log frontend events ingested, by kind and outcome." + ::mdef/labels ["kind" "outcome"] ::mdef/type :counter}}) (def system-config @@ -604,25 +622,25 @@ ::setup/shared-keys (ig/ref ::setup/shared-keys)} ::jobs/defs - {:sendmail (ig/ref ::email/job-def) - :delete-object (ig/ref :app.tasks.delete-object/job-def) - :demo-purge (ig/ref :app.tasks.demo-purge/job-def) - :run-webhook (ig/ref ::webhooks/run-webhook-job-def) - :process-webhook-event (ig/ref ::webhooks/process-webhook-event-job-def) - :file-gc (ig/ref :app.tasks.file-gc/job-def) - :offload-file-data (ig/ref :app.tasks.offload-file-data/job-def) - :objects-gc (ig/ref :app.tasks.objects-gc/job-def) - :export-binfile (ig/ref :app.tasks.export-binfile/job-def) - :import-binfile (ig/ref :app.tasks.import-binfile/job-def) - :storage-gc-deleted (ig/ref ::sto.gc-deleted/job-def) - :storage-gc-touched (ig/ref ::sto.gc-touched/job-def) - :storage-pending-gc (ig/ref ::sto.pending-gc/job-def) - :jobs-gc (ig/ref :app.jobs.gc/job-def) - :telemetry (ig/ref :app.tasks.telemetry/job-def) - :session-gc (ig/ref ::session/job-def) - :file-gc-scheduler (ig/ref :app.tasks.file-gc-scheduler/job-def) - :audit-log-archive (ig/ref :app.loggers.audit.archive-task/job-def) - :audit-log-gc (ig/ref :app.loggers.audit.gc-task/job-def)} + {:sendmail (ig/ref ::email/job-def) + :delete-object (ig/ref :app.tasks.delete-object/job-def) + :demo-purge (ig/ref :app.tasks.demo-purge/job-def) + :run-webhook (ig/ref ::webhooks/run-webhook-job-def) + :process-webhook-event (ig/ref ::webhooks/process-webhook-event-job-def) + :file-gc (ig/ref :app.tasks.file-gc/job-def) + :offload-file-data (ig/ref :app.tasks.offload-file-data/job-def) + :objects-gc (ig/ref :app.tasks.objects-gc/job-def) + :export-binfile (ig/ref :app.tasks.export-binfile/job-def) + :import-binfile (ig/ref :app.tasks.import-binfile/job-def) + :storage-gc-deleted (ig/ref ::sto.gc-deleted/job-def) + :storage-gc-touched (ig/ref ::sto.gc-touched/job-def) + :storage-pending-gc (ig/ref ::sto.pending-gc/job-def) + :jobs-gc (ig/ref :app.jobs.gc/job-def) + :telemetry (ig/ref :app.tasks.telemetry/job-def) + :session-gc (ig/ref ::session/job-def) + :file-gc-scheduler (ig/ref :app.tasks.file-gc-scheduler/job-def) + :audit-log-archive-nexus (ig/ref :app.loggers.audit.archive-task/nexus-job-def) + :audit-log-archive-admin-console (ig/ref :app.loggers.audit.archive-task/admin-console-job-def)} ::jobs/request-pool {::rds/client (ig/ref ::rds/client) @@ -711,16 +729,16 @@ {::db/pool (ig/ref ::db/pool) ::mtx/metrics (ig/ref ::mtx/metrics)} - :app.loggers.audit.archive-task/job-def + :app.loggers.audit.archive-task/nexus-job-def {::db/pool (ig/ref ::db/pool) ::http.client/client (ig/ref ::http.client/client) ::setup/shared-keys (ig/ref ::setup/shared-keys) - :app.nitrate/client (ig/ref :app.nitrate/client) ::mtx/metrics (ig/ref ::mtx/metrics)} - :app.loggers.audit.gc-task/job-def - {::db/pool (ig/ref ::db/pool) - ::mtx/metrics (ig/ref ::mtx/metrics)} + :app.loggers.audit.archive-task/admin-console-job-def + {::db/pool (ig/ref ::db/pool) + :app.nitrate/client (ig/ref :app.nitrate/client) + ::mtx/metrics (ig/ref ::mtx/metrics)} ::email/blacklist {} @@ -841,15 +859,13 @@ {:cron #penpot/cron "0 30 */3,23 * * ?" :job :telemetry} - (when (or (contains? cf/flags :nexus) - (contains? cf/flags :admin-console)) + (when (contains? cf/flags :nexus) {:cron #penpot/cron "0 */5 * * * ?" ;; every 5m - :job :audit-log-archive}) + :job :audit-log-archive-nexus}) - (when (or (contains? cf/flags :nexus) - (contains? cf/flags :admin-console)) - {:cron #penpot/cron "30 */5 * * * ?" ;; every 5m - :job :audit-log-gc})]} + (when (contains? cf/flags :admin-console) + {:cron #penpot/cron "30 */5 * * * ?" ;; every 5m, offset 30s from nexus + :job :audit-log-archive-admin-console})]} ::wrk/dispatcher {::rds/client (ig/ref ::rds/client) diff --git a/backend/src/app/migrations.clj b/backend/src/app/migrations.clj index 7b25493c8d..104222cc00 100644 --- a/backend/src/app/migrations.clj +++ b/backend/src/app/migrations.clj @@ -529,7 +529,10 @@ :fn (mg/resource "app/migrations/sql/0155-add-idx-to-file-change.sql")} {:name "0156-reset-pg-storage-defaults" - :fn (mg/resource "app/migrations/sql/0156-reset-pg-storage-defaults.sql")}]) + :fn (mg/resource "app/migrations/sql/0156-reset-pg-storage-defaults.sql")} + + {:name "0157-add-audit-log-pending-table" + :fn (mg/resource "app/migrations/sql/0157-add-audit-log-pending-table.sql")}]) (defn apply-migrations! [pool name migrations] diff --git a/backend/src/app/migrations/sql/0157-add-audit-log-pending-table.sql b/backend/src/app/migrations/sql/0157-add-audit-log-pending-table.sql new file mode 100644 index 0000000000..4c6d8d998d --- /dev/null +++ b/backend/src/app/migrations/sql/0157-add-audit-log-pending-table.sql @@ -0,0 +1,39 @@ +-- Work queue for the audit log archivers: one row per (record, +-- archiver) pair still pending. Inserting a record creates its +-- pendings; each archiver deletes its own on completion, and whoever +-- deletes the last pending also deletes the record. Archiver ids are +-- fixed in code and never reused: 1 = nexus, 2 = nitrate. + +CREATE TABLE audit_log_pending ( + -- DEFERRABLE so the test-suite reset (which deletes table by table + -- with SET CONSTRAINTS ALL DEFERRED) keeps working; the guard still + -- fires when a record with pendings is deleted. Plain DEFERRABLE + -- (initially immediate) is enough: every deleter removes pendings + -- before rows, so the check passes at each statement. + audit_log_id uuid NOT NULL REFERENCES audit_log(id) DEFERRABLE, + archiver smallint NOT NULL CHECK (archiver BETWEEN 1 AND 20), + PRIMARY KEY (archiver, audit_log_id)); + +-- Lets the final delete ask "is anything still pending for this +-- record?" without scanning the whole queue. +CREATE INDEX ON audit_log_pending (audit_log_id); + +-- NOTE: no backfill on purpose. Rows predating this queue (sent or +-- not) get no pendings, so the orphan sweep deletes them unsent. +-- Losing the unarchived ones at upgrade time is accepted: this +-- table is a minutes/hours buffer, not a store. + +-- The pending queue replaces the archived_at mark: whoever deletes +-- the last pending deletes the record, so the flag column and the +-- indexes that served the old mark-and-sweep scans go away. +DROP INDEX IF EXISTS audit_log__created_at__idx; +DROP INDEX IF EXISTS audit_log__archived_at__idx; +DROP INDEX IF EXISTS audit_log_id_archived_at_idx; +ALTER TABLE audit_log DROP COLUMN IF EXISTS archived_at; + +-- Serves the archivers' chronological scan (ORDER BY created_at, id): +-- one total order for every archiver, so concurrent drains cannot +-- deadlock on row locks. Covers the whole ORDER BY (bulk ingest +-- shares request-at constantly, so ties are the norm). +CREATE INDEX IF NOT EXISTS audit_log__created_at__idx + ON audit_log (created_at ASC, id); diff --git a/backend/src/app/rpc/commands/audit.clj b/backend/src/app/rpc/commands/audit.clj index 893160a23e..771b321606 100644 --- a/backend/src/app/rpc/commands/audit.clj +++ b/backend/src/app/rpc/commands/audit.clj @@ -16,8 +16,10 @@ [app.db :as db] [app.http :as-alias http] [app.loggers.audit :as audit] + [app.loggers.audit.archive-task :as archive] [app.loggers.database :as loggers.db] [app.loggers.mattermost :as loggers.mm] + [app.metrics :as mtx] [app.rpc :as-alias rpc] [app.rpc.climit :as-alias climit] [app.rpc.doc :as-alias doc] @@ -71,6 +73,9 @@ (def ^:private xf:map-event-row (comp (map adjust-timestamp) + ;; Each row mints its own id: the full row and its telemetry + ;; shadow are distinct records that must not share it. + (map #(assoc % :id (uuid/next))) (map event->row))) (defn- prepare-events @@ -80,8 +85,7 @@ initiator (audit/get-client-initiator request) xform (comp (map (fn [event] - {:id (uuid/next) - :type (:type event) + {:type (:type event) :name (:name event) :props (:props event) ;; The initiator always comes from the @@ -113,6 +117,19 @@ (assoc :source "telemetry:frontend")))) (map event->row))) +(defn- record-ingested + [metrics kind outcome amount] + (when (pos? amount) + (mtx/run! metrics :id :audit-ingest-events :labels [kind outcome] :inc amount))) + +(defn- insert-rows! + [tx rows] + (when (seq rows) + (db/insert-many! tx :audit-log event-columns rows) + ;; Same transaction as the rows: every ingested record is born + ;; pending for every enabled archiver (see archive-task). + (archive/queue-pendings tx (map first rows)))) + (defn- handle-events [{:keys [::db/pool] :as cfg} params] (let [events (prepare-events params)] @@ -124,19 +141,30 @@ (run! (partial loggers.db/emit cfg) events) (run! (partial loggers.mm/emit cfg) events)) - (when (contains? cf/flags :audit-log) - ;; Process and save full audit events when audit-log flag is active - (when-let [rows (-> (sequence xf:map-event-row events) - (not-empty))] - (db/insert-many! pool :audit-log event-columns rows))) - - (when (contains? cf/flags :telemetry) - ;; Store anonymized frontend events so the telemetry task can ship them - ;; in batches. Runs independently from the audit-log insert above so - ;; both modes can be active simultaneously. - (when-let [rows (-> (sequence xf:map-telemetry-event-row events) - (not-empty))] - (db/insert-many! pool :audit-log event-columns rows))))) + ;; Rows and their pendings commit together in one transaction: a + ;; row never outlives its queue. Both faces are best-effort, so a + ;; failure drops the whole batch (counted below as failed); the + ;; endpoint still answers without breaking the request. + (let [metrics (mtx/instance cfg) + full-rows (if (contains? cf/flags :audit-log) + (into [] xf:map-event-row events) + []) + shadow-rows (if (contains? cf/flags :telemetry) + (into [] xf:map-telemetry-event-row events) + [])] + (try + (db/tx-run! pool + (fn [tx] + (insert-rows! tx full-rows) + (insert-rows! tx shadow-rows))) + (record-ingested metrics "full" "inserted" (count full-rows)) + (record-ingested metrics "shadow" "inserted" (count shadow-rows)) + (catch Throwable cause + ;; Whatever never landed counts as failed per attempted + ;; kind; the endpoint logs the cause as before. + (record-ingested metrics "full" "failed" (count full-rows)) + (record-ingested metrics "shadow" "failed" (count shadow-rows)) + (throw cause)))))) (def ^:private valid-event-types #{"action" "identify" "trigger"}) diff --git a/backend/src/app/tasks/telemetry.clj b/backend/src/app/tasks/telemetry.clj index 50029746e3..8088841380 100644 --- a/backend/src/app/tasks/telemetry.clj +++ b/backend/src/app/tasks/telemetry.clj @@ -218,19 +218,32 @@ ;; buffer stays bounded. (def ^:private batch-size 10000) +(def ^:private sql:aged-telemetry + "source LIKE 'telemetry:%' AND created_at < now() - interval '7 days'") + +(def ^:private sql:gc-pendings + (str "DELETE FROM audit_log_pending WHERE audit_log_id IN" + " (SELECT id FROM audit_log WHERE " sql:aged-telemetry ")")) + (def ^:private sql:gc-events - "DELETE FROM audit_log - WHERE source LIKE 'telemetry:%' - AND created_at < now() - interval '7 days'") + (str "DELETE FROM audit_log WHERE " sql:aged-telemetry)) (defn- gc-events "Delete telemetry-mode events older than `telemetry-retention-days` so that the buffer stays bounded." [{:keys [::db/conn]}] - (let [result (db/exec-one! conn [sql:gc-events])] - (when (pos? (:next.jdbc/update-count result)) - (l/warn :hint "purged stale telemetry events" - :count (:next.jdbc/update-count result))))) + ;; One transaction for both deletes: a crash between them would + ;; leave pending-less rows for the archive sweep to delete unsent. + ;; Pendings first: the queue holds a row per archiver for every + ;; record, and the FK would reject deleting a row that still has + ;; pendings. + (db/tx-run! conn + (fn [{:keys [::db/conn]}] + (db/exec-one! conn [sql:gc-pendings]) + (let [result (db/exec-one! conn [sql:gc-events])] + (when (pos? (:next.jdbc/update-count result)) + (l/warn :hint "purged stale telemetry events" + :count (:next.jdbc/update-count result))))))) (def ^:private sql:fetch-telemetry-events "SELECT id, name, type, source, tracked_at, profile_id, props, context @@ -274,10 +287,17 @@ false)))) (defn- delete-sent-events - "Delete rows by their ids after a successful send." + "Delete rows by their ids after a successful send. One transaction + for both deletes: a crash between them would leave pending-less + rows for the archive sweep to delete unsent." [conn ids] - (let [arr (db/create-array conn "uuid" ids)] - (db/exec-one! conn ["DELETE FROM audit_log WHERE id = ANY(?)" arr]))) + (db/tx-run! conn + (fn [tx] + (let [tx-conn (::db/conn tx) + arr (db/create-array tx-conn "uuid" ids)] + ;; Pendings first: same FK reason as in `gc-events`. + (db/exec-one! tx-conn ["DELETE FROM audit_log_pending WHERE audit_log_id = ANY(?)" arr]) + (db/exec-one! tx-conn ["DELETE FROM audit_log WHERE id = ANY(?)" arr]))))) (defn- collect-and-send-audit-events "Collect anonymous telemetry-mode audit events and ship them to the diff --git a/backend/test/backend_tests/loggers_audit_archive_task_test.clj b/backend/test/backend_tests/loggers_audit_archive_task_test.clj index a04c2c2fd1..c0e5a5a0e0 100644 --- a/backend/test/backend_tests/loggers_audit_archive_task_test.clj +++ b/backend/test/backend_tests/loggers_audit_archive_task_test.clj @@ -9,203 +9,244 @@ [app.common.json :as json] [app.common.schema :as sm] [app.common.time :as ct] + [app.common.transit :as transit] [app.common.uri :as u] [app.common.uuid :as uuid] [app.config :as cf] [app.db :as db] [app.http.client :as-alias http] [app.jobs :as jobs] + [app.loggers.audit :as audit] [app.loggers.audit.archive-task :as archive] + [app.main :as main] + [app.metrics :as mtx] + [app.metrics.definition :as-alias mdef] [app.setup :as-alias setup] [backend-tests.helpers :as th] [clojure.test :as t] [cuerdas.core :as str] [integrant.core :as ig] - [mockery.core :refer [with-mocks]])) + [mockery.core :refer [with-mocks]]) + (:import + io.prometheus.client.Counter + io.prometheus.client.Counter$Child)) (t/use-fixtures :once th/state-init) (t/use-fixtures :each th/database-reset) (def ^:private archive-uri "http://nexus.example/archive") +(defn- insert-pendings! + "Queue the row with the real producer helper: pendings for the + enabled archivers, like append-audit-entry does." + [row-id] + (archive/queue-pendings (:app.db/pool th/*system*) [row-id])) + +(defn- pending-count + [] + (:pending (first (th/db-exec! ["SELECT count(*) AS pending FROM audit_log_pending"])))) + (defn- insert-audit-row! ([profile-id name] (insert-audit-row! profile-id name {:team-id (uuid/next)})) ([profile-id name props] - (th/db-insert! :audit-log - {:id (uuid/next) - :name name - :type "action" - :source "backend" - :profile-id profile-id - :ip-addr (db/inet "127.0.0.1") - :props (db/tjson props) - :context (db/tjson {}) - :tracked-at (ct/now) - :created-at (ct/now)}))) + (let [row (th/db-insert! :audit-log + {:id (uuid/next) + :name name + :type "action" + :source "backend" + :profile-id profile-id + :ip-addr (db/inet "127.0.0.1") + :props (db/tjson props) + :context (db/tjson {}) + :tracked-at (ct/now) + :created-at (ct/now)})] + (insert-pendings! (:id row)) + row))) -(defn- nitrate-event-props - [nitrate-mock] - (let [[_ _ params] (:call-args @nitrate-mock)] +(defn- admin-console-event-props + [admin-console-mock] + (let [[_ _ params] (:call-args @admin-console-mock)] (:props (first (:events params))))) -(t/deftest archive-events-sends-to-nitrate-then-nexus - ;; Create profile before enabling :admin-console — system nitrate - ;; client is nil in tests, and team bootstrap would call it. - (let [prof (th/create-profile* 1 {:is-active true}) - order (atom [])] +(defn- run-both! + "Run both archive jobs, Nexus first (the historical single-run + order). The admin-console job takes no params (closed empty schema)." + [params] + (th/run-task! :audit-log-archive-nexus params) + (th/run-task! :audit-log-archive-admin-console {})) + +(t/deftest archive-jobs-drain-their-own-queues + ;; Two separate jobs, one queue each: the nexus job sends without + ;; touching the admin console, and only the admin-console job + ;; deletes the last pending and the row. Create profile before + ;; enabling :admin-console — the system admin-console client is nil + ;; in tests, and team bootstrap would call it. + (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call - :return (fn [& _] - (swap! order conj :nitrate) - nil)} - http-mock {:target 'app.http.client/req - :return (fn [& _] - (swap! order conj :nexus) - {:status 204})}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "create-project") (insert-audit-row! (:id prof) "create-file") - (th/run-task! :audit-log-archive {:uri archive-uri}) - (t/is (:called? @nitrate-mock)) + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}) + (t/is (false? (:called? @admin-console-mock))) (t/is (:called? @http-mock)) - (t/is (= 1 (:call-count @nitrate-mock))) (t/is (= 1 (:call-count @http-mock))) - (t/is (= [:nitrate :nexus] @order)) - (let [[_ method params] (:call-args @nitrate-mock)] + ;; Nexus drained its own queue; both rows wait on the admin console. + (t/is (= 2 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 2 (pending-count))) + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (= 1 (:call-count @admin-console-mock))) + (let [[_ method params] (:call-args @admin-console-mock)] (t/is (= :ingest-audit-log method)) (t/is (= 2 (count (:events params)))) (t/is (= #{"create-project" "create-file"} (into #{} (map :name) (:events params)))) (t/is (every? uuid? (map :id (:events params))))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 2 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) -(t/deftest archive-events-filters-nitrate-event-names +(t/deftest archive-events-filters-admin-console-event-names (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil} + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "create-project") (insert-audit-row! (:id prof) "create-organization-attribute") (insert-audit-row! (:id prof) "unrelated-event") - (th/run-task! :audit-log-archive {:uri archive-uri}) - (t/is (:called? @nitrate-mock)) + (run-both! {:uri archive-uri}) + (t/is (:called? @admin-console-mock)) (t/is (:called? @http-mock)) - (let [[_ _ params] (:call-args @nitrate-mock)] + (let [[_ _ params] (:call-args @admin-console-mock)] (t/is (= #{"create-project" "create-organization-attribute"} (into #{} (map :name) (:events params))))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 3 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) -(t/deftest archive-events-skips-nitrate-when-no-matching-names +(t/deftest archive-events-skips-admin-console-when-no-matching-names (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil} + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "unrelated-event") - (th/run-task! :audit-log-archive {:uri archive-uri}) - (t/is (false? (:called? @nitrate-mock))) + (run-both! {:uri archive-uri}) + (t/is (false? (:called? @admin-console-mock))) (t/is (:called? @http-mock)) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-admin-console-only-marks-without-uri (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil} + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "create-project") - (th/run-task! :audit-log-archive {}) - (t/is (:called? @nitrate-mock)) + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (:called? @admin-console-mock)) (t/is (false? (:called? @http-mock))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-admin-console-only-marks-non-allowlisted (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil} + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "unrelated-event") - (th/run-task! :audit-log-archive {}) - (t/is (false? (:called? @nitrate-mock))) + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (false? (:called? @admin-console-mock))) (t/is (false? (:called? @http-mock))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) -(t/deftest archive-events-propagates-nitrate-error +(t/deftest archive-events-admin-console-error-still-drains-nexus + ;; Mirror of the nexus-throw test: an admin-console failure aborts + ;; only the admin-console job; Nexus already drained its own queue + ;; in its job. The row stays on the admin-console pending. (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call - :throw (ex-info "nitrate down" {})} + (with-mocks [admin-console-mock {:target 'app.nitrate/call + :throw (ex-info "nitrate down" {})} http-mock {:target 'app.http.client/req :return {:status 204}}] (insert-audit-row! (:id prof) "create-project") + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}) (t/is (thrown? clojure.lang.ExceptionInfo - (th/run-task! :audit-log-archive {:uri archive-uri}))) - (t/is (:called? @nitrate-mock)) - (t/is (false? (:called? @http-mock))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is null"])] - (t/is (= 1 (count rows)))))))) + (th/run-task! :audit-log-archive-admin-console {}))) + ;; Nexus already drained its queue; the admin-console failure no + ;; longer blocks it, but the row stays on the admin-console pending. + (t/is (:called? @admin-console-mock)) + (t/is (:called? @http-mock)) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 1 (count rows)))) + (t/is (= 1 (pending-count))))))) (t/deftest archive-events-uses-nexus-without-admin-console (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus}] (with-mocks [http-mock {:target 'app.http.client/req :return {:status 204}} - nitrate-mock {:target 'app.nitrate/call :return nil}] + admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id prof) "create-project") - (th/run-task! :audit-log-archive {:uri archive-uri}) - (t/is (false? (:called? @nitrate-mock))) + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}) + (t/is (false? (:called? @admin-console-mock))) (t/is (:called? @http-mock)) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-skips-mark-when-nexus-fails (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus}] (with-mocks [http-mock {:target 'app.http.client/req :return {:status 500}}] (insert-audit-row! (:id prof) "create-project") - (th/run-task! :audit-log-archive {:uri archive-uri}) + (t/is (thrown? clojure.lang.ExceptionInfo + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}))) (t/is (:called? @http-mock)) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is null"])] + (let [rows (th/db-exec! ["select * from audit_log"])] (t/is (= 1 (count rows)))))))) (t/deftest archive-events-both-flags-skips-mark-when-nexus-fails (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil} + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} http-mock {:target 'app.http.client/req :return {:status 500}}] (insert-audit-row! (:id prof) "create-project") - (th/run-task! :audit-log-archive {:uri archive-uri}) - (t/is (:called? @nitrate-mock)) + (t/is (thrown? clojure.lang.ExceptionInfo + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}))) + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (:called? @admin-console-mock)) (t/is (:called? @http-mock)) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is null"])] + (let [rows (th/db-exec! ["select * from audit_log"])] (t/is (= 1 (count rows)))))))) -(t/deftest archive-events-encodes-db-rows-for-nitrate +(t/deftest archive-events-encodes-db-rows-for-admin-console (let [prof (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:nexus :admin-console} cf/config (assoc cf/config :admin-console-uri "http://ac.example/admin-console/" :admin-console-shared-key "ac-key" :nexus-shared-key "nexus-key")] - (let [nitrate-bodies (atom []) + (let [admin-console-bodies (atom []) shared-keys {:admin-console "ac-key" :nexus "nexus-key"} client (ig/init-key :app.nitrate/client {::http/client (Object.) ::setup/shared-keys shared-keys}) - ;; The archive job is a job-def: the handler closes over the + ;; Each archive job is a job-def: the handler closes over the ;; cfg it was given at init and takes a nil context, because ;; an in-process invocation has no job row to describe. - job-def (ig/init-key :app.loggers.audit.archive-task/job-def - {::db/pool (:app.db/pool th/*system*) - ::setup/shared-keys shared-keys - ::http/client (:app.http.client/client th/*system*) - :app.nitrate/client client})] + nexus-job-def (ig/init-key :app.loggers.audit.archive-task/nexus-job-def + {::db/pool (:app.db/pool th/*system*) + ::setup/shared-keys shared-keys + ::http/client (:app.http.client/client th/*system*) + ::mtx/metrics (th/metrics-stub) + :app.nitrate/client client}) + admin-console-job-def (ig/init-key :app.loggers.audit.archive-task/admin-console-job-def + {::db/pool (:app.db/pool th/*system*) + ::mtx/metrics (th/metrics-stub) + :app.nitrate/client client})] (with-mocks [http-mock {:target 'app.http.client/req :return (fn [_ req & _] (when (str/includes? (str (:uri req)) "api/audit-log") - (swap! nitrate-bodies conj (:body req))) + (swap! admin-console-bodies conj (:body req))) {:status 204})}] (let [team-id (uuid/next) row (th/db-insert! :audit-log @@ -219,9 +260,13 @@ :context (db/tjson {}) :tracked-at (ct/now) :created-at (ct/now)})] - ((::jobs/handler job-def) nil {:uri archive-uri}) - (t/is (= 1 (count @nitrate-bodies))) - (let [body (json/decode (first @nitrate-bodies) :key-fn json/read-kebab-key) + (insert-pendings! (:id row)) + ;; Admin console first so its body is captured before Nexus + ;; deletes the last pending and the row. + ((::jobs/handler admin-console-job-def) nil {}) + ((::jobs/handler nexus-job-def) nil {:uri archive-uri}) + (t/is (= 1 (count @admin-console-bodies))) + (let [body (json/decode (first @admin-console-bodies) :key-fn json/read-kebab-key) event (first (:events body))] (t/is (= 1 (count (:events body)))) (t/is (= (str (:id row)) (str (:id event)))) @@ -230,14 +275,14 @@ (t/is (string? (:created-at event))) (t/is (string? (:tracked-at event))) (t/is (= (str team-id) (str (get-in event [:props :team-id]))))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))))) -(t/deftest archive-events-requires-uri-when-nexus-flag-set +(t/deftest archive-nexus-job-requires-uri-when-nexus-flag-set (with-redefs [cf/flags #{:nexus} cf/config (dissoc cf/config :audit-log-archive-uri)] (let [data (ex-data (try - (th/run-task! :audit-log-archive {}) + (th/run-task! :audit-log-archive-nexus {}) (catch Throwable cause cause)))] (t/is (= :task-not-configured (:code data)))))) @@ -247,7 +292,7 @@ member (th/create-profile* 2 {:is-active true}) team (th/create-team* 1 {:profile-id (:id actor)})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id actor) "create-project" {:team-id (:id team) :profile-id (:id actor) @@ -255,16 +300,16 @@ :member-id (:id member) ;; Producers store this as a string UUID. :user-who-send-invitation (str (:id actor))}) - (th/run-task! :audit-log-archive {}) - (t/is (:called? @nitrate-mock)) - (let [props (nitrate-event-props nitrate-mock)] + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (:called? @admin-console-mock)) + (let [props (admin-console-event-props admin-console-mock)] (t/is (= (:name team) (:team-name props))) (t/is (= (:email actor) (:profile-email props))) (t/is (= (:email member) (:user-email props))) (t/is (= (:email member) (:member-email props))) (t/is (= (:email actor) (:user-who-send-invitation-email props)))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-enriches-soft-deleted-team-and-profile (let [actor (th/create-profile* 1 {:is-active true}) @@ -273,22 +318,22 @@ (th/db-update! :team {:deleted-at (ct/now)} {:id (:id team)}) (th/db-update! :profile {:deleted-at (ct/now)} {:id (:id other)}) (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id actor) "create-project" {:team-id (:id team) :profile-id (:id other)}) - (th/run-task! :audit-log-archive {}) - (let [props (nitrate-event-props nitrate-mock)] + (th/run-task! :audit-log-archive-admin-console {}) + (let [props (admin-console-event-props admin-console-mock)] (t/is (= (:name team) (:team-name props))) (t/is (= (:email other) (:profile-email props)))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-does-not-overwrite-existing-enrich-props (let [actor (th/create-profile* 1 {:is-active true}) team (th/create-team* 1 {:profile-id (:id actor)})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id actor) "create-project" {:team-id (:id team) :team-name "already-set-name" @@ -300,8 +345,8 @@ :member-email "already-member@example.com" :user-who-send-invitation (str (:id actor)) :user-who-send-invitation-email "already-inviter@example.com"}) - (th/run-task! :audit-log-archive {}) - (let [props (nitrate-event-props nitrate-mock)] + (th/run-task! :audit-log-archive-admin-console {}) + (let [props (admin-console-event-props admin-console-mock)] (t/is (= "already-set-name" (:team-name props))) (t/is (= "already-set@example.com" (:profile-email props))) (t/is (= "already-user@example.com" (:user-email props))) @@ -312,29 +357,29 @@ (t/deftest archive-events-enrich-skips-missing-and-non-uuid-ids (let [actor (th/create-profile* 1 {:is-active true})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id actor) "create-project" {:team-id (uuid/next) :profile-id (uuid/next) :user-id "not-a-uuid" :member-id 42}) - (th/run-task! :audit-log-archive {}) - (t/is (:called? @nitrate-mock)) - (let [props (nitrate-event-props nitrate-mock)] + (th/run-task! :audit-log-archive-admin-console {}) + (t/is (:called? @admin-console-mock)) + (let [props (admin-console-event-props admin-console-mock)] (t/is (nil? (:team-name props))) (t/is (nil? (:profile-email props))) (t/is (nil? (:user-email props))) (t/is (nil? (:member-email props))) (t/is (= "not-a-uuid" (:user-id props))) (t/is (= 42 (:member-id props)))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 1 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-events-forwards-export-job-not-legacy-export (let [actor (th/create-profile* 1 {:is-active true}) team (th/create-team* 1 {:profile-id (:id actor)})] (with-redefs [cf/flags #{:admin-console}] - (with-mocks [nitrate-mock {:target 'app.nitrate/call :return nil}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil}] (insert-audit-row! (:id actor) "create-export-binfile-job" {:team-id (:id team) :job-id (uuid/next) @@ -342,20 +387,336 @@ :export-type "detach-libraries"}) (insert-audit-row! (:id actor) "export-binfile" {:file-id (uuid/next)}) - (th/run-task! :audit-log-archive {}) - (let [[_ _ params] (:call-args @nitrate-mock) + (th/run-task! :audit-log-archive-admin-console {}) + (let [[_ _ params] (:call-args @admin-console-mock) events (:events params) props (:props (first events))] (t/is (= ["create-export-binfile-job"] (mapv :name events))) (t/is (= (:id team) (:team-id props))) (t/is (= (:name team) (:team-name props)))) - (let [rows (th/db-exec! ["select * from audit_log where archived_at is not null"])] - (t/is (= 2 (count rows)))))))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) (t/deftest archive-params-accept-empty-cron-props ;; Cron always submits (or props {}) — system deps live on the ;; job-def cfg, not in params. Requiring them here blocks every tick. - (let [valid? (sm/validator archive/schema:audit-log-archive-params)] - (t/is (true? (valid? {}))) - (t/is (true? (valid? {:enabled true}))) - (t/is (true? (valid? {:uri (u/uri archive-uri)}))))) + ;; Both schemas are closed: a stale manual invocation must fail + ;; validation loudly instead of skipping silently. + (let [nexus-valid? (sm/validator archive/schema:audit-log-archive-nexus-params) + admin-console-valid? (sm/validator archive/schema:audit-log-archive-admin-console-params)] + (t/is (true? (nexus-valid? {}))) + (t/is (false? (nexus-valid? {:enabled true}))) + (t/is (true? (nexus-valid? {:uri (u/uri archive-uri)}))) + (t/is (true? (admin-console-valid? {}))) + (t/is (false? (admin-console-valid? {:enabled true}))) + (t/is (false? (admin-console-valid? {:uri (u/uri archive-uri)}))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; METRICS SMOKE TEST +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +;; The job-def cfg of these tests carries an isolated metrics registry +;; (pattern from `jobs-metrics-test`), so assertions read exact values +;; instead of the shared test-system registry. `th/run-task!` would use +;; the latter and its counts would leak across tests. +(def ^:private archive-metric-ids + [:audit-archive-events :audit-archive-discarded]) + +(defn- make-archive-metrics + [] + (ig/init-key :app.metrics/metrics + {:default (select-keys main/default-metrics archive-metric-ids)})) + +(defn- invoke-nexus! + "Invoke the nexus archive handler in-process with an isolated + metrics registry. The handler takes a nil context because there is + no job row to describe (same as `th/run-task!`)." + [metrics params] + (let [job-def (ig/init-key :app.loggers.audit.archive-task/nexus-job-def + {::db/pool (:app.db/pool th/*system*) + ::setup/shared-keys {:admin-console "ac-key" :nexus "nexus-key"} + ::http/client (:app.http.client/client th/*system*) + ::mtx/metrics metrics})] + ((::jobs/handler job-def) nil params))) + +(defn- invoke-admin-console! + "Invoke the admin-console archive handler in-process with an + isolated metrics registry. The admin-console client is the (nil in + tests) system one: tests mock `app.nitrate/call`, so it is never + dereferenced." + [metrics] + (let [job-def (ig/init-key :app.loggers.audit.archive-task/admin-console-job-def + {::db/pool (:app.db/pool th/*system*) + :app.nitrate/client (:app.nitrate/client th/*system*) + ::mtx/metrics metrics})] + ((::jobs/handler job-def) nil {}))) + +(defn- invoke-both! + "Invoke both archive handlers sharing one isolated metrics + registry, Nexus first (the historical single-run order)." + [metrics params] + (invoke-nexus! metrics params) + (invoke-admin-console! metrics)) + +(defn- counter-value + [metrics id labels] + (let [collector (mtx/get-collector metrics id) + instance (::mdef/instance collector) + child (.labels ^Counter instance (into-array String labels))] + (.get ^Counter$Child child))) + +(t/deftest archive-records-metrics-on-success + ;; Smoke test: the run records its domain metrics into the wired + ;; registry (isolated per test, pattern from `jobs-metrics-test`). + ;; Logging itself is not asserted: `dbg` summaries run below the + ;; test log level and failure `warn`s only repeat the outcome the + ;; metrics already prove. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus :admin-console}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 204}}] + (insert-audit-row! (:id prof) "create-project") + (insert-audit-row! (:id prof) "unrelated-event") + (invoke-both! metrics {:uri archive-uri}) + (t/is (= 2.0 (counter-value metrics :audit-archive-events ["nexus" "archived"]))) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["admin-console" "archived"]))) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["admin-console" "skipped"]))) + (t/is (= 0 (pending-count))) + (let [rows (th/db-exec! ["select * from audit_log"])] + (t/is (= 0 (count rows)))))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; PENDING-QUEUE TESTS +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +(defn- full-event + "A complete event for submit*, which (unlike submit) assumes every + required field is filled." + [profile-id] + {:name "create-file" :type "action" :source "backend" + :profile-id profile-id :ip-addr "127.0.0.1" + :props {} :context {} + :created-at (ct/now) :tracked-at (ct/now)}) + +(t/deftest archive-submit-creates-pendings-for-enabled-archivers + ;; The producer hook: a row is born pending for the enabled + ;; archivers, in the same transaction as the row itself. Goes + ;; through submit* because the suite stubs submit itself (see + ;; helpers/init-config). + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:audit-log :nexus :admin-console}] + (audit/submit* {::db/pool (:app.db/pool th/*system*)} + (full-event (:id prof))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 2 (pending-count)))))) + +(t/deftest archive-submit-single-flag-creates-one-pending + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:audit-log :nexus}] + (audit/submit* {::db/pool (:app.db/pool th/*system*)} + (full-event (:id prof))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 1 (pending-count)))))) + +(t/deftest archive-submit-rollback-keeps-no-pendings + ;; A rollback takes the row and its queue together: a row never + ;; outlives its pendings, and no phantom pending pins nothing. + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:audit-log :nexus :admin-console}] + (db/tx-run! {::db/pool (:app.db/pool th/*system*) + ::db/rollback true} + (fn [cfg] + (audit/submit* cfg (full-event (:id prof))) + (t/is (= 1 (count (db/exec! cfg ["select * from audit_log"])))) + (t/is (= 2 (count (db/exec! cfg ["select * from audit_log_pending"])))))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count)))))) + +(t/deftest archive-keeps-row-until-last-archiver-done + ;; Nexus refuses the chunk: the nexus job aborts (failed count + ;; recorded, isolated registry) while the admin-console job still + ;; drains its own queue. The row and the nexus pending stay; once + ;; nexus accepts, everything is gone. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus :admin-console}] + (insert-audit-row! (:id prof) "create-project") + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 500}}] + (let [err (try + (invoke-nexus! metrics {:uri archive-uri}) + nil + (catch clojure.lang.ExceptionInfo cause cause))] + (t/is (= :archive-chunk-refused (:code (ex-data err))))) + (invoke-admin-console! metrics) + (t/is (:called? @admin-console-mock)) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["nexus" "failed"]))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 1 (pending-count)))) + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 204}}] + (run-both! {:uri archive-uri}) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count))))))) + +(t/deftest archive-sweeps-orphan-rows + ;; A row with no pendings (old archived leftover, or born while no + ;; archiver was on) is deleted without any send. + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:nexus :admin-console}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 204}}] + ;; Raw insert, no pendings on purpose. + (th/db-insert! :audit-log + {:id (uuid/next) + :name "create-project" + :type "action" + :source "backend" + :profile-id (:id prof) + :ip-addr (db/inet "127.0.0.1") + :props (db/tjson {}) + :context (db/tjson {}) + :tracked-at (ct/now) + :created-at (ct/now)}) + (run-both! {:uri archive-uri}) + (t/is (false? (:called? @admin-console-mock))) + (t/is (false? (:called? @http-mock))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))))))) + +(t/deftest archive-nexus-transport-failure-records-failed + ;; A transport exception (timeout, DNS, refused) counts the events + ;; as failed and aborts the nexus job: the chunk rolls back and + ;; the rows stay for the next tick. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus}] + (with-mocks [http-mock {:target 'app.http.client/req + :throw (ex-info "timeout" {})}] + (insert-audit-row! (:id prof) "create-project") + (t/is (thrown? clojure.lang.ExceptionInfo + (invoke-nexus! metrics {:uri archive-uri}))) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["nexus" "failed"]))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 1 (pending-count))))))) + +(t/deftest archive-sends-chunks-in-chronological-order + ;; Nexus numbers by arrival for order-sensitive consumers, so + ;; multi-chunk drains must arrive oldest-first no matter how the + ;; queue is scanned. + (let [prof (th/create-profile* 1 {:is-active true}) + base (ct/now) + bodies (atom [])] + (with-redefs [cf/flags #{:nexus}] + (with-mocks [http-mock {:target 'app.http.client/req + :return (fn [_ params & _] + (swap! bodies conj (:body params)) + {:status 204})}] + (doseq [i (range 130)] + (let [row (insert-audit-row! (:id prof) (str "event-" i))] + (th/db-update! :audit-log + {:created-at (.plusSeconds ^java.time.Instant base (long i))} + {:id (:id row)}))) + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}) + (let [received (mapcat #(:events (transit/decode %)) @bodies) + times (map :created-at received)] + (t/is (= 130 (count received))) + (t/is (= (sort times) times)) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"]))))))))) + +(t/deftest archive-reaps-disabled-archiver-pendings + ;; R3 reap by its real path: an admin-console pending born while + ;; admin-console was on is dropped once it is off; nexus still + ;; archives its queue and the row is gone with nothing stuck. The + ;; dropped pending counts as discarded (isolated registry). + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus}] + (with-mocks [http-mock {:target 'app.http.client/req :return {:status 204}} + admin-console-mock {:target 'app.nitrate/call :return nil}] + (let [row (insert-audit-row! (:id prof) "create-project")] + ;; Admin console was on at insert time: both pendings on purpose. + (th/db-exec! ["INSERT INTO audit_log_pending (audit_log_id, archiver) + VALUES (?::uuid, 2) + ON CONFLICT DO NOTHING" + (:id row)]) + (t/is (= 2 (pending-count))) + (invoke-nexus! metrics {:uri archive-uri}) + (t/is (:called? @http-mock)) + (t/is (false? (:called? @admin-console-mock))) + (t/is (= 1.0 (counter-value metrics :audit-archive-discarded ["reaped"]))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count)))))))) + +(t/deftest archive-sweeps-rows-born-with-no-archiver-on + ;; Accepted R3 consequence by its real path (not a hand-made + ;; orphan): ingested with no send flag on, the row is swept + ;; unarchived once the task runs, with zero sends. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:audit-log}] + (audit/submit* {::db/pool (:app.db/pool th/*system*)} + (full-event (:id prof))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count)))) + (with-redefs [cf/flags #{:audit-log :nexus}] + (with-mocks [http-mock {:target 'app.http.client/req :return {:status 204}}] + (invoke-nexus! metrics {:uri archive-uri}) + (t/is (false? (:called? @http-mock))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count))) + (t/is (= 1.0 (counter-value metrics :audit-archive-discarded ["swept"]))) + (t/is (= 0.0 (counter-value metrics :audit-archive-events ["nexus" "archived"]))))))) + +(t/deftest archive-nexus-throw-still-drains-admin-console + ;; Consumer independence, now structural: a Nexus transport throw + ;; aborts only the nexus job, and the admin-console job still drains + ;; its own queue. Nexus keeps its pending (so the row stays); the + ;; admin console completes its own queue. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus :admin-console}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req + :throw (ex-info "timeout" {})}] + (insert-audit-row! (:id prof) "create-project") + (t/is (thrown? clojure.lang.ExceptionInfo + (invoke-nexus! metrics {:uri archive-uri}))) + (invoke-admin-console! metrics) + (t/is (:called? @admin-console-mock)) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["nexus" "failed"]))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 1 (pending-count))))))) + +(t/deftest archive-reap-then-sweeps-without-sends + ;; End to end of the reap path: a row born with only an + ;; admin-console pending is reaped once :admin-console goes off, + ;; then swept with zero sends on either destination. + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:admin-console}] + (insert-audit-row! (:id prof) "create-project") + (t/is (= 1 (pending-count)))) + (with-redefs [cf/flags #{:nexus}] + (with-mocks [http-mock {:target 'app.http.client/req :return {:status 204}} + admin-console-mock {:target 'app.nitrate/call :return nil}] + (th/run-task! :audit-log-archive-nexus {:uri archive-uri}) + (t/is (false? (:called? @http-mock))) + (t/is (false? (:called? @admin-console-mock))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count))))))) + +(t/deftest archive-fully-filtered-admin-console-chunk-counts-skipped + ;; An admin-console chunk with every event outside the allowlist + ;; never calls ingest, but still counts its rows as skipped and + ;; completes its queue. + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-archive-metrics)] + (with-redefs [cf/flags #{:nexus :admin-console}] + (with-mocks [admin-console-mock {:target 'app.nitrate/call :return nil} + http-mock {:target 'app.http.client/req :return {:status 204}}] + (insert-audit-row! (:id prof) "unrelated-event") + (invoke-both! metrics {:uri archive-uri}) + (t/is (false? (:called? @admin-console-mock))) + (t/is (= 1.0 (counter-value metrics :audit-archive-events ["admin-console" "skipped"]))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (pending-count))))))) diff --git a/backend/test/backend_tests/rpc_audit_test.clj b/backend/test/backend_tests/rpc_audit_test.clj index 7c99c0093c..6080c483cf 100644 --- a/backend/test/backend_tests/rpc_audit_test.clj +++ b/backend/test/backend_tests/rpc_audit_test.clj @@ -14,13 +14,24 @@ [app.db :as db] [app.jobs :as-alias jobs] [app.loggers.audit :as audit] + [app.main :as main] + [app.metrics :as mtx] + [app.metrics.definition :as-alias mdef] [app.rpc :as-alias rpc] [app.rpc.climit :as climit] + [app.rpc.commands.audit :as cmd-audit] [app.util.services :as sv] [backend-tests.helpers :as th] + [clojure.string :as str] [clojure.test :as t] + [integrant.core :as ig] [mockery.core :refer [with-mocks]] - [yetti.request])) + [yetti.request]) + (:import + io.prometheus.client.Collector$MetricFamilySamples + io.prometheus.client.CollectorRegistry + io.prometheus.client.Counter + io.prometheus.client.Counter$Child)) (t/use-fixtures :once th/state-init) (t/use-fixtures :each th/database-reset) @@ -97,7 +108,7 @@ (assoc base :app.http/auth-key-id :nexus))))))) (t/deftest push-events-1 - (with-redefs [app.config/flags #{:audit-log}] + (with-redefs [app.config/flags #{:audit-log :nexus}] (let [prof (th/create-profile* 1 {:is-active true}) team-id (:default-team-id prof) proj-id (:default-project-id prof) @@ -127,7 +138,9 @@ (t/is (= 1 (count rows))) (t/is (= (:id prof) (:profile-id row))) (t/is (= "navigate" (:name row))) - (t/is (= "frontend" (:source row))))))) + (t/is (= "frontend" (:source row))) + ;; The row is born pending for the enabled archiver. + (t/is (= 1 (count (th/db-exec! ["select * from audit_log_pending"])))))))) (t/deftest push-events-2 (with-redefs [app.config/flags #{:audit-log}] @@ -228,36 +241,41 @@ (t/deftest push-events-both-flags-creates-two-rows ;; When both :audit-log and :telemetry flags are active, two rows ;; should be stored: one full audit entry and one telemetry entry. - (with-redefs [cf/flags #{:audit-log :telemetry}] - (let [prof (th/create-profile* 1 {:is-active true}) - params {::th/type :push-audit-events - ::rpc/profile-id (:id prof) - :events [{:name "navigate" - :props {:route "dashboard"} - :context {:browser "Chrome" - :version "2.0.0" - :initiator "app"} - :timestamp (ct/now) - :type "action"}]} - params (with-meta params - {:app.http/request http-request}) - out (th/command! params)] + ;; The profile is created before enabling :admin-console: team + ;; bootstrap calls the real nitrate client, which is unconfigured + ;; in tests. + (let [prof (th/create-profile* 1 {:is-active true})] + (with-redefs [cf/flags #{:audit-log :telemetry :nexus :admin-console}] + (let [params {::th/type :push-audit-events + ::rpc/profile-id (:id prof) + :events [{:name "navigate" + :props {:route "dashboard"} + :context {:browser "Chrome" + :version "2.0.0" + :initiator "app"} + :timestamp (ct/now) + :type "action"}]} + params (with-meta params + {:app.http/request http-request}) + out (th/command! params)] - (t/is (nil? (:error out))) + (t/is (nil? (:error out))) - (let [[row1 row2 :as rows] (->> (th/db-exec! ["select * from audit_log order by source"]) - (mapv decode-row))] - (t/is (= 2 (count rows))) - ;; First row: full audit-log entry - (t/is (= "frontend" (:source row1))) - (t/is (contains? (:props row1) :route)) - (t/is (not= "0.0.0.0" (str (:ip-addr row1)))) - ;; Second row: telemetry entry - (t/is (= "telemetry:frontend" (:source row2))) - (t/is (= "0.0.0.0" (str (:ip-addr row2)))) - (let [day-now (ct/truncate (ct/now) :days)] - (t/is (= day-now (:created-at row2))) - (t/is (= day-now (:tracked-at row2)))))))) + (let [[row1 row2 :as rows] (->> (th/db-exec! ["select * from audit_log order by source"]) + (mapv decode-row))] + (t/is (= 2 (count rows))) + ;; First row: full audit-log entry + (t/is (= "frontend" (:source row1))) + (t/is (contains? (:props row1) :route)) + (t/is (not= "0.0.0.0" (str (:ip-addr row1)))) + ;; Second row: telemetry entry + (t/is (= "telemetry:frontend" (:source row2))) + (t/is (= "0.0.0.0" (str (:ip-addr row2)))) + (let [day-now (ct/truncate (ct/now) :days)] + (t/is (= day-now (:created-at row2))) + (t/is (= day-now (:tracked-at row2)))) + ;; Both rows are born pending for the enabled archivers. + (t/is (= 4 (count (th/db-exec! ["select * from audit_log_pending"]))))))))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; BACKEND PROCESS-EVENT PATH (RPC commands) @@ -857,3 +875,89 @@ (t/is (= (:id caller) (:profile-id (prepare-event (:id caller) result))) (str "override " (pr-str bad) " must fall back to the caller")))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; INGEST METRICS (bulk frontend path) +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +;; handle-events resolves its metrics from the cfg (same pattern as +;; the jobs substrate), so these tests drive it directly with an +;; isolated registry instead of the shared test-system one. +(defn- make-ingest-metrics + [] + (ig/init-key :app.metrics/metrics + {:default (select-keys main/default-metrics [:audit-ingest-events])})) + +(defn- ingest-counter + [metrics kind outcome] + (let [collector (mtx/get-collector metrics :audit-ingest-events) + instance (::mdef/instance collector) + child (.labels ^Counter instance (into-array String [kind outcome]))] + (.get ^Counter$Child child))) + +(defn- counter-label-sets + "Exact label sets present for the ingest counter, read off the + registry so phantom series cannot hide. Matches the family by + prefix (Prometheus strips the `_total` suffix from the family + name) and only counts `_total` samples, not `_created` ones." + [metrics] + (let [full (get main/default-metrics :audit-ingest-events) + base (str/replace (::mdef/name full) #"_total$" "")] + (->> (enumeration-seq + (.metricFamilySamples ^CollectorRegistry (mtx/get-registry metrics))) + (filter (fn [^Collector$MetricFamilySamples family] + (str/starts-with? (.-name family) base))) + (mapcat (fn [^Collector$MetricFamilySamples family] + (.samples family))) + (filter (fn [sample] (str/ends-with? (.-name sample) "_total"))) + (into #{} (map (fn [sample] (vec (.-labelValues sample)))))))) + +(defn- push-params + [profile-id] + (with-meta {::rpc/profile-id profile-id + ::rpc/request-at (ct/now) + :events [{:name "navigate" :type "action" + :props {:route "dashboard"} + :timestamp (ct/now)}]} + ;; A map request (like th/command! builds): the reify IRequest + ;; stub only serves the IP headers and blows up on x-client. + {:app.http/request (th/make-dummy-request)})) + +(t/deftest push-handle-events-records-inserted + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-ingest-metrics)] + (with-redefs [cf/flags #{:audit-log :telemetry :nexus :admin-console}] + (@#'cmd-audit/handle-events {::db/pool th/*pool* ::mtx/metrics metrics} + (push-params (:id prof))) + (t/is (= 1.0 (ingest-counter metrics "full" "inserted"))) + (t/is (= 1.0 (ingest-counter metrics "shadow" "inserted"))) + (t/is (= 2 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 4 (count (th/db-exec! ["select * from audit_log_pending"]))))))) + +(t/deftest push-handle-events-records-failed-on-error + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-ingest-metrics)] + (with-redefs [cf/flags #{:audit-log :telemetry :nexus :admin-console}] + (with-mocks [insert-mock {:target 'app.db/insert-many! + :throw (ex-info "boom" {})}] + (t/is (thrown? clojure.lang.ExceptionInfo + (@#'cmd-audit/handle-events {::db/pool th/*pool* ::mtx/metrics metrics} + (push-params (:id prof))))) + (t/is (= 1.0 (ingest-counter metrics "full" "failed"))) + (t/is (= 1.0 (ingest-counter metrics "shadow" "failed"))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log_pending"])))))))) + +(t/deftest push-handle-events-single-face-emits-no-empty-series + ;; With only :audit-log on, exactly one ingest series exists: the + ;; pos? guard in record-ingested emits no phantom shadow series. + ;; (No send flags either, so R3 leaves the row pending-less.) + (let [prof (th/create-profile* 1 {:is-active true}) + metrics (make-ingest-metrics)] + (with-redefs [cf/flags #{:audit-log}] + (@#'cmd-audit/handle-events {::db/pool th/*pool* ::mtx/metrics metrics} + (push-params (:id prof))) + (t/is (= 1.0 (ingest-counter metrics "full" "inserted"))) + (t/is (= #{["full" "inserted"]} (counter-label-sets metrics))) + (t/is (= 1 (count (th/db-exec! ["select * from audit_log"])))) + (t/is (= 0 (count (th/db-exec! ["select * from audit_log_pending"]))))))) diff --git a/backend/test/backend_tests/tasks_telemetry_test.clj b/backend/test/backend_tests/tasks_telemetry_test.clj index 79f29f335b..4585de8c98 100644 --- a/backend/test/backend_tests/tasks_telemetry_test.clj +++ b/backend/test/backend_tests/tasks_telemetry_test.clj @@ -11,6 +11,7 @@ [app.config :as cf] [app.db :as db] [app.loggers.audit :as audit] + [app.loggers.audit.archive-task :as at] [app.tasks.telemetry :as telemetry] [app.util.json :as json] [backend-tests.helpers :as th] @@ -999,3 +1000,45 @@ (th/run-task! :telemetry {:send? true :enabled? true}) (t/is (= 1 @send-count)) ;; still 1, not 2 (t/is (= 0 (count-telemetry-rows))))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; PENDING-QUEUE REGRESSION (telemetry deletes) +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +(defn- insert-pending-pair! + "Queue the row for both archivers (from the archiver constants, not + literals): the delete paths must face pending-ful rows." + [row-id] + (doseq [archiver [at/archiver-nexus at/archiver-admin-console]] + (th/db-exec! ["INSERT INTO audit_log_pending (audit_log_id, archiver) + VALUES (?::uuid, ?::smallint) + ON CONFLICT DO NOTHING" + row-id archiver]))) + +(defn- count-pendings [] + (-> (th/db-exec-one! ["SELECT count(*) AS cnt FROM audit_log_pending"]) + :cnt + long)) + +(t/deftest test-delete-sent-events-removes-pendings-first + ;; Regression: telemetry rows carry pendings for both archivers, so + ;; deleting the row must delete its pendings first, or the FK + ;; rejects the delete and the batch is sent again next tick. + (let [row (insert-telemetry-row! "navigate")] + (insert-pending-pair! (:id row)) + (t/is (= 2 (count-pendings))) + (db/tx-run! th/*pool* + (fn [{:keys [::db/conn]}] + (@#'telemetry/delete-sent-events conn [(:id row)]))) + (t/is (= 0 (count-telemetry-rows))) + (t/is (= 0 (count-pendings))))) + +(t/deftest test-gc-events-removes-pendings-first + ;; Same guarantee for the >7-days purge. + (let [past (ct/in-past {:days 8}) + row (insert-telemetry-row! "navigate" {:created-at past :tracked-at past})] + (insert-pending-pair! (:id row)) + (t/is (= 2 (count-pendings))) + (@#'telemetry/gc-events {::db/conn th/*pool*}) + (t/is (= 0 (count-telemetry-rows))) + (t/is (= 0 (count-pendings)))))