diff --git a/benchmark/README.md b/benchmark/README.md new file mode 100644 index 0000000..ba4e672 --- /dev/null +++ b/benchmark/README.md @@ -0,0 +1,283 @@ +# Benchmark + +What does auditing cost? This benchmark measures two things on PostgreSQL: + +- **Write overhead**: the same insert or update, and the same request with a + `log()` call, with and without the library, and with both `on_error` + settings. +- **Read latency** on a synthetic log of 5 million entries over 12 months: + `list_groups`, `object_history` and `access_summary`, and how many + partitions each one actually reads. + +It is **not** a comparison with other libraries (sqlalchemy-continuum, +triggers, pgaudit), and it is not a tuned setup. It uses one connection and +one machine, with PostgreSQL in a local Docker container on stock settings. + +## Headline + +Measured 2026-09-26 on a laptop (PostgreSQL 18.6 in Docker, stock settings; +details below): + +- **Writes:** auditing a flush of 1 object adds **2.4 ms** at p50 (a plain + insert takes 3.4 ms; audited, 5.8 ms). With `on_error="raise"`, which + skips the savepoint, it adds **1.5 ms**. A flush of 100 objects adds + **11.4 ms**, about **114 µs per object** (9.8 ms with `"raise"`). A + durable `log()` adds **6.7 ms**. +- **Installed but switched off** (`audit_enabled=False`) stays within + 0.6 ms of plain at p50. +- **Reads on 5M entries:** the first page of `list_groups` (50 groups) takes + **18.5 ms** at p50 without a query window and **15.1 ms** with a 30-day + window. PostgreSQL spends under 1 ms executing the queries of either. The + window saves planning time: its plans hold 27 partition scans instead of + 82. +- Pagination reads **5 of 80 partitions** for a next page, including one + six months back. Filters with no time bound (`actor_id`, `object_history`, + `access_summary` over the whole history) read all 64 activity + partitions, and come in at 10 to 22 ms here. + +## Results + +Raw JSON: `benchmark/results/run_20260926T104637Z_7e088fba.json` +(gitignored). Every number in this README comes from that run. The full run +took 677 s, including 302 s of seeding and 127 s of `VACUUM ANALYZE`. + +### Environment + +| | | +| --- | --- | +| CPU | Intel Core i7-8850H @ 2.60GHz, 12 threads | +| RAM | 15.5 GB visible to the WSL2 VM | +| OS | Linux 6.18 (WSL2 on Windows), Docker Engine 29.8.1 | +| PostgreSQL | 18.6 (`postgres:18` image via testcontainers), **stock config**: `shared_buffers` 128 MB, `work_mem` 4 MB, `jit` on, `synchronous_commit` on, `fsync` on | +| Client | Python 3.12.11, SQLAlchemy 2.1.0, psycopg 3.3.6 (sync `Session`) | +| Library | sqlalchemy-audit-trail 0.1.0 at commit 953974f | +| Round trip | `SELECT 1` on an open connection: p50 0.32 ms, p95 0.48 ms | + +### Writes + +Milliseconds per transaction, from `add_all`/change through `commit()`. There +were 1000 iterations for 1 and 10 objects and 300 for 100 objects, after 100 +warm-up iterations per variant. +- `audited` uses the default `on_error="log"`; `"raise"` is the same + capture without the savepoint. +- "Statements" counts what goes through the cursor per transaction, for + plain / log / raise. BEGIN and COMMIT are not included. + +| Operation | Objects | plain p50 / p95 | disabled p50 / p95 | audited p50 / p95 | audited, `"raise"` p50 / p95 | Added p50, `"log"` | Added p50, `"raise"` | Per object, log / raise | Statements | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| insert | 1 | 3.37 / 5.17 | 3.34 / 4.98 | 5.80 / 8.51 | 4.88 / 7.33 | +2.43 ms (+72%) | +1.52 ms (+45%) | 2429 / 1516 µs | 1 / 5 / 3 | +| insert | 10 | 3.98 / 6.69 | 4.05 / 6.59 | 7.68 / 11.86 | 6.34 / 9.63 | +3.70 ms (+93%) | +2.36 ms (+59%) | 370 / 236 µs | 1 / 5 / 3 | +| insert | 100 | 8.04 / 12.80 | 7.84 / 10.91 | 19.42 / 36.04 | 17.87 / 27.18 | +11.38 ms (+141%) | +9.83 ms (+122%) | 114 / 98 µs | 1 / 5 / 3 | +| update | 1 | 2.83 / 3.71 | 2.84 / 4.20 | 5.26 / 7.24 | 4.46 / 6.15 | +2.43 ms (+86%) | +1.63 ms (+58%) | 2427 / 1633 µs | 1 / 5 / 3 | +| update | 10 | 3.37 / 4.42 | 3.51 / 5.28 | 7.13 / 10.53 | 6.04 / 8.89 | +3.76 ms (+112%) | +2.68 ms (+80%) | 376 / 268 µs | 1 / 5 / 3 | +| update | 100 | 7.07 / 9.96 | 7.64 / 15.78 | 18.75 / 34.57 | 17.34 / 22.91 | +11.68 ms (+165%) | +10.27 ms (+145%) | 117 / 103 µs | 1 / 5 / 3 | + +The `log()` rows are one plain insert plus the event, compared with the same +insert without it (1000 iterations each): + +| Operation | plain p50 / p95 | `"log"` p50 / p95 | `"raise"` p50 / p95 | Added p50, `"log"` | Added p50, `"raise"` | Statements | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | +| `log()` | 3.24 / 4.98 | 5.58 / 8.11 | 4.78 / 7.00 | +2.34 ms (+72%) | +1.54 ms (+47%) | 1 / 5 / 3 | +| durable `log()` | 5.63 / 7.04 | 12.37 / 15.05 | 12.31 / 15.31 | +6.73 ms (+119%) | +6.67 ms (+118%) | 1 / 3 / 3 | + +Where the time goes: + +- **The fixed cost is mostly round trips.** With the default + `on_error="log"`, an audited flush sends four more statements than a + plain one: a savepoint, the `audit_transaction` insert, the + `audit_activity` insert, and the savepoint release. At 0.32 ms per + round trip here, the round trips alone are about 1.3 ms of the 2.4 ms. + The rest is the server work behind those statements (the savepoint, and + inserts into two partitioned tables with their indexes), plus a little + Python. The percentages look large because the baseline is small: one + statement plus the commit. +- **What `on_error="raise"` saves:** the savepoint and its release. It + measured 0.8 to 1.5 ms less than `"log"` at p50 in every insert, update + and `log()` case. That is more than the two round trips alone (about + 0.6 ms), so the savepoint's server-side work counts too. The price is + that a failed audit write (for example a missing partition) fails your + transaction instead of being logged. +- **The per-object cost** at 100 objects is about 100 to 120 µs. A cProfile + of the 100-object insert, taken separately from this run (and inflating + Python time), put roughly half of the listener's time in the executemany + of the 100 JSONB activity rows and about a third in building the diffs + in Python. +- **A durable `log()`** commits on its own connection, so a request pays + for a second transaction and a second commit. `on_error` changes nothing + there (−0.06 ms): a durable write never uses the savepoint. +- **`disabled`** stays within 0.6 ms of `plain` at p50 (−2.5% to +8.0%). + The differences go in both directions, which is noise at this scale. The + mixin's `active_history` listeners have nothing to load here, because the + rows were loaded before the change (see the protocol). + +### Reads + +5,000,000 `audit_activity` rows (4.38 GB with indexes) and 2,500,000 +`audit_transaction` rows (0.56 GB), spread over 365 days. There are +4 severities × 16 months = 64 activity partitions (12 past months, the +current one and 3 ahead) and 16 transaction partitions. Each case ran 20 +warm-up and 200 measured calls, each in its own session. Limit is 50 +groups. + +The last four columns come from one `EXPLAIN (ANALYZE)` run of every +statement the call sent: +- "Partitions read" counts distinct partitions actually executed. +- "Planned" sums the partition scans in the plans, including those pruned + at run time. +- "Planning" and "Execution" are server-side milliseconds under EXPLAIN's + instrumentation. + +| Case | p50 ms | p95 ms | Groups | Statements | Partitions read (of 80) | Planned | Planning ms | Execution ms | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| `list_groups` first page | 18.5 | 25.8 | 50 | 7 | 17 | 82 | 5.3 | 0.7 | +| `list_groups` first page, 30-day window | 15.1 | 19.7 | 50 | 7 | 17 | 27 | 2.3 | 0.4 | +| `list_groups` next page | 22.1 | 28.6 | 50 | 8 | 5 | 74 | 5.5 | 0.7 | +| `list_groups` next page, 30-day window | 17.4 | 23.2 | 50 | 8 | 5 | 19 | 2.8 | 0.7 | +| `list_groups` page 6 months back | 18.3 | 22.4 | 50 | 8 | 5 | 44 | 4.7 | 0.7 | +| `list_groups(actor_id=...)` | 22.1 | 33.7 | 50 | 4 | 65 | 121 | 7.8 | 2.7 | +| `object_history` (with children) | 21.5 | 25.8 | 50 | 4 | 68 | 148 | 9.4 | 1.9 | +| `access_summary`, whole history | 10.0 | 13.9 | — (367 entries counted) | 1 | 64 | 64 | 3.1 | 1.9 | + +What this shows: + +- **Data volume barely matters for pages.** `list_groups` runs one + `LIMIT 51` query per severity. PostgreSQL reads each severity's monthly + partitions newest first and stops once it has enough rows. The 17 + partitions read on a first page are 4 per severity plus one transaction + partition. Of the 4 per severity, 3 are the empty partitions created + ahead of time and 1 is the current month. A next page, or a page six + months back, reads one partition per severity, because the cursor bounds + `created_at` from above. +- **The window saves planning, not execution.** Without a lower bound, each + per-severity query is planned over all 16 of its months and pruned at run + time. With the 30-day window it is planned over 5. That is the 5.3 ms + versus 2.3 ms planning, and most of the 3.4 ms latency difference on the + first page. Planning cost grows with the number of partitions a query + cannot exclude, so it grows with retention. +- **Most of the latency is client side.** The server executes these pages + in under 1 ms. The rest is 7 or 8 round trips (about 2.5 ms here), + planning, and Python: row decoding, grouping and compaction. +- **Where pruning cannot help.** `actor_id` without `severities`, + `object_history` and `access_summary` filter on a column that is not a + partition key, and have no time bound (`object_history` and + `access_summary` ignore the default window by design). They must probe + the index of every activity partition, all 64 of them. It stays fast here + because each probe finds few rows. The cost grows with the number of + partitions (months kept × severities), not with rows. Passing `since` + bounds it. + +## What these numbers are not + +- **Not a comparison with other libraries.** Nothing else was measured. +- **Not a tuned server.** The PostgreSQL configuration is stock (128 MB + `shared_buffers`, `synchronous_commit` on). The write baseline is + dominated by commit fsync inside a WSL2 Docker VM. On real hardware, or + with other durability settings, the plain baseline and the percentages + will differ. The added round trips and per-object costs are the more + portable numbers. +- **A mostly idle machine.** Other workloads were stopped for this run; one + unrelated, idle PostgreSQL container was still running on the host. +- **Not concurrent.** One connection, sequential. Throughput columns in the + JSON (`ops_per_s`, `objects_per_s`) are 1 / mean latency on that one + connection, not server capacity. +- **Only sync psycopg.** `AsyncSession`/asyncpg is not measured. +- **Synthetic data.** The dataset is generated, with this deterministic + skew: INFO 70%, NOTICE 20%, WARNING 8%, CRITICAL 2%. 10% of entries are + `person.viewed` on 1,000 people. The rest are `entity.created` / + `updated` / `deleted` on 9 object types × 100,000 ids, 20% of them with + an `Order` target (5,000 ids). There are 1,000 actors and 50 scopes, and + 1 to 3 entries per transaction. Real logs have other shapes. In + particular, a rare severity makes its per-severity query read further + back. +- **Not cold-cache reads.** Each case runs warm-up calls first. The page + cache holds whatever seeding and warm-up left in it, and cold-cache reads + of the 4.9 GB dataset are not measured. + +## Protocol + +**Write variants.** +- `plain`: a model without `Audited`, on a session class the library is not + installed on. This is the true zero and the baseline for every + percentage. +- `disabled`: the `Audited` model (same columns) on the installed session + class, with `session.info["audit_enabled"] = False`. The mixin's + `active_history` listeners and the session listeners still run, but no + entries are written. This answers "what does leaving it installed but off + cost". +- `audited`: the `Audited` model on the installed session class, capturing + inside an `audit.context(...)`, with the default `on_error="log"`: the + audit inserts run in a savepoint, so a failed audit write does not fail + the transaction. +- `audited_raise`: the same with an `AuditTrail(on_error="raise")`, + installed on its own session class: no savepoint. + +Before measuring, the runner checks that each variant is what it claims to +be, and stops if not: +- the plain model is not `Audited`; +- the plain session class has no trail installed; +- each audited session has the expected trail, `on_error` and capture + flag. + +The checks that passed are listed under `write.preconditions` in the JSON. + +**Write operations.** Each operation is one transaction ended by `commit()`. +- Insert: `add_all(n new objects)` + `commit()`. The objects are built + outside the timer. +- Update: the same n rows are loaded in the session's transaction outside + the timer, then two columns are changed and `commit()` runs, all timed. +- `log()`: one plain insert, then `audit.log(session, event, + payload=...)`, then `commit()`. The events use `benchmark.*` verbs and are + defined only when the write benchmark runs, never at import. +- Durable `log()`: the same, with an event declared `durable=True`. + +The variants of a case run in interleaved blocks of 50, after warm-up, so +drift from table growth or autovacuum falls on all of them. Write tables and +audit tables start empty and are separate from the read dataset. + +**Read dataset.** It is seeded with one `INSERT ... SELECT` over +`generate_series` per table, not through the ORM. Each activity row's +`created_at` equals its transaction's `issued_at`, as the library writes +them. `VACUUM (ANALYZE)` runs afterwards. The runner creates its own schemas +and drops them at the end (`--keep` keeps them). + +**Partitions.** A listener captures every `SELECT` a call sends. Each one is +then re-run as `EXPLAIN (ANALYZE, FORMAT JSON)` with the same parameters, on +the same connection and in the same transaction, after the library's +`SET LOCAL plan_cache_mode = 'force_custom_plan'`. This needs no superuser. + +## Rerun + +Needs Docker, or an existing server: + +```bash +uv sync --frozen +uv run python -m benchmark.run_benchmark # postgres:18 in Docker, ~12 min +uv run python -m benchmark.run_benchmark --image postgres:14 +uv run python -m benchmark.run_benchmark --database-url postgresql://user:pw@host/db +``` + +With `--database-url`, the runner creates schemas named `bench__*` and +drops them afterwards. The URL's driver is replaced with psycopg. + +Quick check (seconds): + +```bash +uv run python -m benchmark.run_benchmark --iterations 20 --iterations-large 10 \ + --warmup 5 --activity-rows 20000 --read-iterations 5 +``` + +| Flag | Default | | +| --- | --- | --- | +| `--database-url` | none | use this server instead of a container | +| `--image` | `postgres:18` | container image | +| `--batch-sizes` | `1,10,100` | objects per flush | +| `--iterations` / `--iterations-large` | 1000 / 300 | measured writes; the second for batches ≥ `--large-batch` (100) | +| `--warmup` / `--block` | 100 / 50 | warm-up writes per variant; interleaving block | +| `--activity-rows` | 5,000,000 | read dataset size (transactions = half) | +| `--read-iterations` / `--read-warmup` | 200 / 20 | calls per read case | +| `--skip-write` / `--skip-read` | off | run one half only | +| `--keep` | off | keep the benchmark schemas | +| `--output-dir` | `benchmark/results/` | where the JSON goes | + +`tests/db/test_benchmark.py` runs the whole runner at tiny sizes in the test +suite, so the script keeps working between full runs. diff --git a/benchmark/__init__.py b/benchmark/__init__.py new file mode 100644 index 0000000..5fd4dd0 --- /dev/null +++ b/benchmark/__init__.py @@ -0,0 +1,4 @@ +"""Benchmark of sqlalchemy-audit-trail: write overhead and read latency. + +Run with ``uv run python -m benchmark.run_benchmark``; see ``benchmark/README.md``. +""" diff --git a/benchmark/env.py b/benchmark/env.py new file mode 100644 index 0000000..6b2fda2 --- /dev/null +++ b/benchmark/env.py @@ -0,0 +1,125 @@ +"""The machine, server and library versions a run was measured on.""" + +from __future__ import annotations + +import os +import platform +import subprocess +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import TypedDict + +from sqlalchemy import Engine, text + +ROOT = Path(__file__).resolve().parents[1] + +PG_SETTINGS = ( + "shared_buffers", + "work_mem", + "effective_cache_size", + "maintenance_work_mem", + "max_parallel_workers_per_gather", + "jit", + "random_page_cost", + "synchronous_commit", + "fsync", +) + + +class Postgres(TypedDict): + """The PostgreSQL server.""" + + version: str + server: str + settings: dict[str, str] + + +class Environment(TypedDict): + """Where a run was measured.""" + + cpu: str + threads: int | None + ram_gb: float | None + os: str + database: str + postgres: Postgres + python: str + sqlalchemy: str + psycopg: str + audit_trail: str + commit: str | None + + +def _cpu() -> str: + try: + for line in Path("/proc/cpuinfo").read_text().splitlines(): + if line.startswith("model name"): + return line.split(":", 1)[1].strip() + except OSError: + pass + return platform.processor() or platform.machine() + + +def _ram_gb() -> float | None: + try: + for line in Path("/proc/meminfo").read_text().splitlines(): + if line.startswith("MemTotal:"): + return round(int(line.split()[1]) / 1024 / 1024, 1) + except OSError: + pass + return None + + +def _package(name: str) -> str: + try: + return version(name) + except PackageNotFoundError: + return "not installed" + + +def _commit() -> str | None: + try: + result = subprocess.run( + ["git", "-C", str(ROOT), "rev-parse", "--short", "HEAD"], + capture_output=True, + text=True, + check=True, + ) + except (OSError, subprocess.CalledProcessError): + return None + return result.stdout.strip() or None + + +def environment(engine: Engine, database: str) -> Environment: + """Describe the machine, the server and the versions in use. + + Args: + engine: Engine connected to the benchmark database. + database: How the database was provided, e.g. the container image. + """ + with engine.connect() as conn: + server = str(conn.execute(text("SELECT version()")).scalar_one()) + short = str(conn.execute(text("SHOW server_version")).scalar_one()) + rows = conn.execute( + text( + "SELECT name, setting, unit FROM pg_settings WHERE name = ANY(:names)" + ), + {"names": list(PG_SETTINGS)}, + ) + settings = { + name: setting if unit is None else f"{setting} ({unit})" + for name, setting, unit in rows + } + return Environment( + cpu=_cpu(), + threads=os.cpu_count(), + ram_gb=_ram_gb(), + os=platform.platform(), + database=database, + postgres=Postgres(version=short, server=server, settings=settings), + python=platform.python_version(), + sqlalchemy=_package("sqlalchemy"), + psycopg=_package("psycopg"), + audit_trail=_package("sqlalchemy-audit-trail"), + commit=_commit(), + ) diff --git a/benchmark/explain.py b/benchmark/explain.py new file mode 100644 index 0000000..78fda19 --- /dev/null +++ b/benchmark/explain.py @@ -0,0 +1,113 @@ +"""Partitions scanned, from ``EXPLAIN (ANALYZE, FORMAT JSON)``. + +The statements a library call sends are captured with a +``before_cursor_execute`` listener, then each ``SELECT`` is explained with +the same parameters on the same connection, in the same transaction (after +the library's ``SET LOCAL plan_cache_mode``). Needs no superuser, unlike +``auto_explain``. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator +from typing import Any, NamedTuple, TypedDict + +from sqlalchemy import Connection, Engine, event + + +class StatementPlan(TypedDict): + """One statement, the partitions its plan touched, and its server time. + + The times are those ``EXPLAIN ANALYZE`` reports for one run, with its + instrumentation overhead; they show where the time goes, not latency. + """ + + sql: str + partitions_scanned: list[str] + partitions_in_plan: int + planning_ms: float + execution_ms: float + + +class _Captured(NamedTuple): + statement: str + parameters: object + + +def nodes(plan: dict[str, Any]) -> Iterator[dict[str, Any]]: + """Every node of a JSON plan, depth first.""" + yield plan + for child in plan.get("Plans", []): + yield from nodes(child) + + +def scanned(plan: dict[str, Any]) -> tuple[list[str], int]: + """Relations a plan executed, and the number of scan nodes it holds. + + A scan node that the executor pruned at run time stays in the plan with + ``Actual Loops`` 0 ("never executed"); partitions pruned at plan time are + not in it at all. + """ + executed: set[str] = set() + planned = 0 + for node in nodes(plan): + relation = node.get("Relation Name") + if relation is None: + continue + planned += 1 + if node.get("Actual Loops", 0) > 0: + executed.add(str(relation)) + return sorted(executed), planned + + +def explain_call( + engine: Engine, + connection_of: Callable[[], Connection], + call: Callable[[], object], +) -> list[StatementPlan]: + """Run ``call`` and explain every ``SELECT`` it sent. + + Args: + engine: Engine whose statements are captured. + connection_of: Returns the connection ``call`` ran on, still in its + transaction. + call: The library call. + """ + captured: list[_Captured] = [] + + def capture( + conn: Connection, + cursor: object, + statement: str, + parameters: object, + context: object, + executemany: bool, + ) -> None: + if statement.lstrip().upper().startswith("SELECT") and not executemany: + captured.append(_Captured(statement, parameters)) + + event.listen(engine, "before_cursor_execute", capture) + try: + call() + finally: + event.remove(engine, "before_cursor_execute", capture) + + connection = connection_of() + plans: list[StatementPlan] = [] + for item in captured: + result = connection.exec_driver_sql( + "EXPLAIN (ANALYZE, FORMAT JSON) " + item.statement, + item.parameters, # type: ignore[arg-type] + ) + document: list[dict[str, Any]] = result.scalar_one() + executed, planned = scanned(document[0]["Plan"]) + plans.append( + StatementPlan( + sql=" ".join(item.statement.split()), + partitions_scanned=executed, + partitions_in_plan=planned, + planning_ms=float(document[0]["Planning Time"]), + execution_ms=float(document[0]["Execution Time"]), + ) + ) + return plans diff --git a/benchmark/models.py b/benchmark/models.py new file mode 100644 index 0000000..6e712d3 --- /dev/null +++ b/benchmark/models.py @@ -0,0 +1,71 @@ +"""The models and events the write benchmark uses. + +``BenchPlain`` and ``BenchAudited`` have the same columns; only the second +mixes in ``Audited``. Their tables have no schema: the runner maps them into +its own schema with ``schema_translate_map``. +""" + +from __future__ import annotations + +from decimal import Decimal +from functools import cache + +from sqlalchemy import Numeric +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + +from audit_trail import Audited, AuditEvent, Severity, event + + +class Base(DeclarativeBase): + """Declarative base of the benchmark models.""" + + +class _Columns: + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] + status: Mapped[str] + counter: Mapped[int] + amount: Mapped[Decimal] = mapped_column(Numeric(12, 2)) + note: Mapped[str | None] + + +class BenchPlain(_Columns, Base): + """Not audited: the baseline.""" + + __tablename__ = "bench_plain" + + +class BenchAudited(_Columns, Base, Audited): + """Audited, with the same columns as ``BenchPlain``.""" + + __tablename__ = "bench_audited" + + +@cache +def bench_events() -> type[AuditEvent]: + """The events written with ``log()``: ``VIEWED`` and durable ``EXPORTED``. + + Built on first use, not at import. ``AuditTrail(events=None)`` registers + every ``AuditEvent`` subclass in the process, and an event with the + built-in ``Severity`` would break that for a host with its own severity + enum. Importing the benchmark therefore defines none; only running the + write benchmark does, in its own process. The verbs use the + ``benchmark.`` prefix so they cannot clash with anyone's. + """ + + class BenchEvent(AuditEvent): + VIEWED = event("benchmark.viewed", Severity.INFO) + EXPORTED = event("benchmark.exported", Severity.WARNING, durable=True) + + return BenchEvent + + +def row_values(index: int) -> dict[str, object]: + """Column values of the ``index``-th benchmark object.""" + return { + "name": f"item {index}", + "status": "draft", + "counter": 0, + "amount": Decimal(index % 1000) + Decimal("0.99"), + "note": None if index % 2 else f"note {index}", + } diff --git a/benchmark/read.py b/benchmark/read.py new file mode 100644 index 0000000..020c738 --- /dev/null +++ b/benchmark/read.py @@ -0,0 +1,410 @@ +"""Read latency on a large, synthetic audit log. + +The log is seeded with plain SQL (``generate_series``), not through the ORM: +``audit_transaction`` rows spread evenly over the last 365 days, and one to +three ``audit_activity`` rows per transaction (two on average) with the +transaction's ``issued_at`` as ``created_at``, as the library writes them. + +Row mix (deterministic, from a hash of the row number): + +- severity: INFO 70%, NOTICE 20%, WARNING 8%, CRITICAL 2%; +- verb: 10% ``person.viewed`` on 1,000 ``Person`` ids; otherwise + ``entity.created`` (10%), ``entity.deleted`` (5%) or ``entity.updated`` + on nine object types with 100,000 ids each; +- 20% of the entity rows have an ``Order`` target (5,000 ids); +- 1,000 actors, 50 scopes. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from datetime import datetime, timedelta +from typing import NamedTuple, TypedDict + +from sqlalchemy import Engine, text +from sqlalchemy.orm import Session, sessionmaker + +from audit_trail import ALL_HISTORY, AuditTrail, Cursor, Page +from audit_trail.maintenance import ensure_partitions +from audit_trail.migrations import create_audit_tables +from benchmark.explain import StatementPlan, explain_call +from benchmark.stats import summarize, timed + +WINDOW = timedelta(days=30) +SPAN = timedelta(days=365) + +HISTORY_OBJECT = ("Order", "42") +ACCESS_OBJECT = ("Person", "7") +ACCESS_VERB = "person.viewed" +ACTOR = "user-7" + +SEVERITY_MIX = {"INFO": 0.70, "NOTICE": 0.20, "WARNING": 0.08, "CRITICAL": 0.02} + +_TRANSACTIONS = """ +INSERT INTO {transaction} ( + id, issued_at, actor_type, actor_id, actor_label, remote_addr, user_agent, + method, path, channel, auth_method, request_id, correlation_id +) +OVERRIDING SYSTEM VALUE +SELECT + g, + CAST(:start AS timestamptz) + (g - 1) * CAST(:step AS interval), + 'user', + 'user-' || (g % 1000), + 'user' || (g % 1000) || '@example.com', + CAST('10.0.0.0' AS inet) + (g % 65536), + 'Mozilla/5.0 (benchmark)', + 'POST', + '/api/orders/' || (g % 100000), + 'api', + 'session', + CAST(md5('r' || g) AS uuid), + CAST(md5('c' || g) AS uuid) +FROM generate_series(1, :n) AS g +""" + +_ACTIVITIES = """ +INSERT INTO {activity} ( + transaction_id, verb, severity, object_type, object_id, object_label, + target_type, target_id, actor_id, scope_id, correlation_id, created_at, data +) +SELECT + g, v.verb, v.severity, v.object_type, v.object_id, + v.object_type || ' ' || v.object_id, + CASE WHEN v.targeted THEN 'Order' END, + CASE WHEN v.targeted THEN CAST((r.h2 / 5) % 5000 AS text) END, + 'user-' || (g % 1000), + 'tenant-' || (g % 50), + CAST(md5('c' || g) AS uuid), + CAST(:start AS timestamptz) + (g - 1) * CAST(:step AS interval), + CASE WHEN v.viewed + THEN jsonb_build_object( + 'v', 1, + 'payload', jsonb_build_object( + 'fields', jsonb_build_array('email', 'phone')), + 'context', v.context) + ELSE jsonb_build_object( + 'v', 1, + 'changes', jsonb_build_object( + 'status', jsonb_build_array('draft', 'sent'), + 'amount', jsonb_build_array( + (r.h % 1000) || '.00', (r.h % 1000 + 1) || '.00')), + 'context', v.context) + END +FROM generate_series(1, :n) AS g +CROSS JOIN LATERAL generate_series(1, g % 3 + 1) AS j +CROSS JOIN LATERAL ( + SELECT + CAST(hashint8(CAST(g * 4 + j AS bigint)) AS bigint) & 2147483647 AS h, + CAST(hashint8(CAST(g * 4 + j AS bigint) + 1000000000000) AS bigint) + & 2147483647 AS h2 +) AS r +CROSS JOIN LATERAL ( + SELECT (r.h / 100) % 20 < 2 AS viewed +) AS k +CROSS JOIN LATERAL ( + SELECT + k.viewed, + CASE + WHEN r.h % 100 < 70 THEN 10 + WHEN r.h % 100 < 90 THEN 20 + WHEN r.h % 100 < 98 THEN 30 + ELSE 40 + END AS severity, + CASE + WHEN k.viewed THEN 'person.viewed' + WHEN (r.h / 100) % 20 < 4 THEN 'entity.created' + WHEN (r.h / 100) % 20 = 4 THEN 'entity.deleted' + ELSE 'entity.updated' + END AS verb, + CASE + WHEN k.viewed THEN 'Person' + ELSE (ARRAY['Order', 'OrderLine', 'Invoice', 'Customer', + 'Product', 'Payment', 'Shipment', 'Ticket', + 'Comment'])[1 + (r.h / 2000) % 9] + END AS object_type, + CASE + WHEN k.viewed THEN CAST((r.h / 18000) % 1000 AS text) + ELSE CAST((r.h / 18000) % 100000 AS text) + END AS object_id, + NOT k.viewed AND r.h2 % 5 = 0 AS targeted, + jsonb_build_object( + 'actor_type', 'user', + 'actor_label', 'user' || (g % 1000) || '@example.com', + 'remote_addr', '10.0.' || (g % 256) || '.' || (g / 256 % 256), + 'channel', 'api', + 'method', 'POST', + 'path', '/api/orders/' || (g % 100000)) AS context +) AS v +""" + + +class Sizes(TypedDict): + """On-disk size of the seeded tables, indexes included.""" + + activity_bytes: int + transaction_bytes: int + + +class Partitions(TypedDict): + """Leaf partitions of the audit tables.""" + + activity: int + transaction: int + + +class Dataset(TypedDict): + """The seeded log.""" + + activity_rows: int + transaction_rows: int + days: int + severity_mix: dict[str, float] + partitions: Partitions + sizes: Sizes + seed_s: float + vacuum_analyze_s: float + + +class ReadCase(TypedDict): + """Latency and partitions of one read.""" + + name: str + description: str + iterations: int + p50_ms: float + p95_ms: float + mean_ms: float + min_ms: float + groups: int + rows: int + partitions_scanned: int + explain_planning_ms: float + explain_execution_ms: float + statements: list[StatementPlan] + + +class ReadResult(TypedDict): + """Everything the read benchmark measured.""" + + dataset: Dataset + cases: list[ReadCase] + + +class ReadConfig(NamedTuple): + """Sizes of the read benchmark.""" + + activity_rows: int + iterations: int + warmup: int + + +class _Scenario(NamedTuple): + name: str + description: str + call: Callable[[Session], object] + + +def seed(engine: Engine, audit: AuditTrail, activity_rows: int) -> Dataset: + """Create the audit tables in ``audit``'s schema and fill them.""" + transactions = max(activity_rows // 2, 1) + with engine.begin() as conn: + end: datetime = conn.execute( + text("SELECT date_trunc('second', now()) - interval '1 minute'") + ).scalar_one() + start = end - SPAN + create_audit_tables(conn, audit.tables, audit.severities) + ensure_partitions( + conn, audit.tables, audit.severities, months_ahead=15, now=start + ) + params = {"start": start, "step": SPAN / transactions, "n": transactions} + started = time.perf_counter() + with engine.begin() as conn: + conn.execute( + text(_TRANSACTIONS.format(transaction=audit.tables.transaction.fullname)), + params, + ) + conn.execute( + text(_ACTIVITIES.format(activity=audit.tables.activity.fullname)), params + ) + seed_s = time.perf_counter() - started + started = time.perf_counter() + with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn: + for table in (audit.tables.transaction, audit.tables.activity): + conn.execute(text(f"VACUUM (ANALYZE) {table.fullname}")) + vacuum_s = time.perf_counter() - started + with engine.connect() as conn: + + def count(sql: str, table: str) -> int: + return int(conn.execute(text(sql), {"t": table}).scalar_one()) + + leaves = "SELECT count(*) FROM pg_partition_tree(:t) WHERE isleaf" + size = "SELECT sum(pg_total_relation_size(relid)) FROM pg_partition_tree(:t)" + rows = "SELECT count(*) FROM {}" + activity = audit.tables.activity.fullname + transaction = audit.tables.transaction.fullname + return Dataset( + activity_rows=int(conn.execute(text(rows.format(activity))).scalar_one()), + transaction_rows=int( + conn.execute(text(rows.format(transaction))).scalar_one() + ), + days=SPAN.days, + severity_mix=SEVERITY_MIX, + partitions=Partitions( + activity=count(leaves, activity), transaction=count(leaves, transaction) + ), + sizes=Sizes( + activity_bytes=count(size, activity), + transaction_bytes=count(size, transaction), + ), + seed_s=seed_s, + vacuum_analyze_s=vacuum_s, + ) + + +def _scenarios( + audit: AuditTrail, windowed: AuditTrail, sessions: sessionmaker[Session] +) -> list[_Scenario]: + with sessions() as session: + first = audit.query.list_groups(session) + first_windowed = windowed.query.list_groups(session) + newest = first.groups[0].transaction.issued_at if first.groups else None + deep = None + if newest is not None: + deep = Cursor(newest - timedelta(days=182), 2**62, ALL_HISTORY) + + scenarios = [ + _Scenario( + "list_groups first page", + "list_groups(), no window: all severities, one query per severity", + audit.query.list_groups, + ), + _Scenario( + "list_groups first page, 30-day window", + "list_groups() with default_query_window=30 days", + windowed.query.list_groups, + ), + ] + if first.next_cursor is not None: + cursor = first.next_cursor + scenarios.append( + _Scenario( + "list_groups next page", + "list_groups(cursor=page 1's next_cursor), no window", + lambda s: audit.query.list_groups(s, cursor=cursor), + ) + ) + if first_windowed.next_cursor is not None: + windowed_cursor = first_windowed.next_cursor + scenarios.append( + _Scenario( + "list_groups next page, 30-day window", + "list_groups(cursor=page 1's next_cursor), 30-day window", + lambda s: windowed.query.list_groups(s, cursor=windowed_cursor), + ) + ) + if deep is not None: + deep_cursor = deep + scenarios.append( + _Scenario( + "list_groups page 6 months back", + "list_groups(cursor=a position 182 days back), no window", + lambda s: audit.query.list_groups(s, cursor=deep_cursor), + ) + ) + scenarios += [ + _Scenario( + "list_groups by actor", + f"list_groups(actor_id={ACTOR!r}), no window: one query over all " + "severities", + lambda s: audit.query.list_groups(s, actor_id=ACTOR), + ), + _Scenario( + "object_history", + f"object_history{HISTORY_OBJECT}, with children, first page", + lambda s: audit.query.object_history(s, *HISTORY_OBJECT), + ), + _Scenario( + "access_summary", + f"access_summary{(*ACCESS_OBJECT, ACCESS_VERB)}, whole history", + lambda s: audit.query.access_summary(s, *ACCESS_OBJECT, ACCESS_VERB), + ), + ] + return scenarios + + +def _size(result: object) -> tuple[int, int]: + # Groups and entries of a page; rows of a list (access_summary). + if isinstance(result, Page): + return len(result.groups), sum(len(g.activities) for g in result.groups) + if isinstance(result, list): + return 0, len(result) + raise TypeError(f"unexpected result {type(result).__name__}") + + +def _measure( + engine: Engine, + sessions: sessionmaker[Session], + scenario: _Scenario, + config: ReadConfig, +) -> ReadCase: + def once() -> float: + # A session (and so a transaction) per call, as a request would. + with sessions() as session: + return timed(lambda: scenario.call(session)) + + for _ in range(config.warmup): + once() + summary = summarize([once() for _ in range(config.iterations)]) + with sessions() as session: + results: list[object] = [] + plans = explain_call( + engine, + session.connection, + lambda: results.append(scenario.call(session)), + ) + groups, rows = _size(results[0]) + partitions = {name for plan in plans for name in plan["partitions_scanned"]} + return ReadCase( + name=scenario.name, + description=scenario.description, + iterations=summary.iterations, + p50_ms=summary.p50_ms, + p95_ms=summary.p95_ms, + mean_ms=summary.mean_ms, + min_ms=summary.min_ms, + groups=groups, + rows=rows, + partitions_scanned=len(partitions), + explain_planning_ms=sum(plan["planning_ms"] for plan in plans), + explain_execution_ms=sum(plan["execution_ms"] for plan in plans), + statements=plans, + ) + + +def run( + engine: Engine, + audit_schema: str, + config: ReadConfig, + progress: Callable[[str], None], +) -> ReadResult: + """Seed the log, then time every read scenario and explain it.""" + audit = AuditTrail(engine, schema=audit_schema, events=[]) + windowed = AuditTrail( + engine, schema=audit_schema, events=[], default_query_window=WINDOW + ) + progress(f"read: seeding {config.activity_rows:,} activity rows") + dataset = seed(engine, audit, config.activity_rows) + progress( + f"read: seeded in {dataset['seed_s']:.0f} s, " + f"VACUUM ANALYZE {dataset['vacuum_analyze_s']:.0f} s" + ) + sessions = sessionmaker(engine) + cases: list[ReadCase] = [] + for scenario in _scenarios(audit, windowed, sessions): + progress(f"read: {scenario.name} ({config.iterations} it)") + cases.append(_measure(engine, sessions, scenario, config)) + audit.dispose() + windowed.dispose() + return ReadResult(dataset=dataset, cases=cases) diff --git a/benchmark/results/.gitkeep b/benchmark/results/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/benchmark/run_benchmark.py b/benchmark/run_benchmark.py new file mode 100644 index 0000000..cd8a951 --- /dev/null +++ b/benchmark/run_benchmark.py @@ -0,0 +1,303 @@ +"""What does auditing cost? Write overhead and read latency on PostgreSQL. + +Full run (starts a throwaway postgres:18 container with testcontainers): + + uv run python -m benchmark.run_benchmark + +Against a server you already have (the runner creates and drops its own +schemas there): + + uv run python -m benchmark.run_benchmark --database-url postgresql+psycopg://... + +Raw results are written as JSON to ``benchmark/results/`` (gitignored). +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +import uuid +from collections.abc import Iterator, Sequence +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import TypedDict + +from sqlalchemy import Engine, create_engine, make_url, text + +from benchmark import read, write +from benchmark.env import Environment, environment +from benchmark.stats import summarize, timed + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_OUTPUT_DIR = ROOT / "benchmark" / "results" +SCHEMA_VERSION = 1 + + +class Config(TypedDict): + """The sizes a run used.""" + + batch_sizes: list[int] + iterations: int + iterations_large: int + large_batch: int + warmup: int + block: int + activity_rows: int + read_iterations: int + read_warmup: int + + +class RoundTrip(TypedDict): + """Latency of ``SELECT 1`` on an open connection: the cost of one round trip.""" + + iterations: int + p50_ms: float + p95_ms: float + + +class Report(TypedDict): + """The JSON a run writes.""" + + schema_version: int + created_at: str + duration_s: float + config: Config + environment: Environment + round_trip: RoundTrip + write: write.WriteResult | None + read: read.ReadResult | None + + +def _sizes(value: str) -> list[int]: + try: + sizes = [int(part) for part in value.split(",") if part.strip()] + except ValueError as error: + raise argparse.ArgumentTypeError(f"not a list of integers: {value}") from error + if not sizes or min(sizes) < 1: + raise argparse.ArgumentTypeError("batch sizes must be >= 1") + return sizes + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + """Parse the command line.""" + parser = argparse.ArgumentParser( + description="Measure sqlalchemy-audit-trail write overhead and read latency." + ) + parser.add_argument( + "--database-url", + help="PostgreSQL URL (driver is set to psycopg). Default: start a " + "testcontainers container.", + ) + parser.add_argument("--image", default="postgres:18", help="Container image.") + parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR) + parser.add_argument("--batch-sizes", type=_sizes, default=[1, 10, 100]) + parser.add_argument("--iterations", type=int, default=1000) + parser.add_argument( + "--iterations-large", + type=int, + default=300, + help="Iterations for batches of --large-batch objects or more.", + ) + parser.add_argument("--large-batch", type=int, default=100) + parser.add_argument("--warmup", type=int, default=100) + parser.add_argument( + "--block", type=int, default=50, help="Iterations per interleaved block." + ) + parser.add_argument("--activity-rows", type=int, default=5_000_000) + parser.add_argument("--read-iterations", type=int, default=200) + parser.add_argument("--read-warmup", type=int, default=20) + parser.add_argument("--skip-write", action="store_true") + parser.add_argument("--skip-read", action="store_true") + parser.add_argument( + "--keep", action="store_true", help="Keep the benchmark schemas." + ) + parser.add_argument("--quiet", action="store_true") + args = parser.parse_args(argv) + for name in ("iterations", "iterations_large", "block", "read_iterations"): + if getattr(args, name) < 1: + parser.error(f"--{name.replace('_', '-')} must be >= 1") + for name in ("warmup", "read_warmup"): + if getattr(args, name) < 0: + parser.error(f"--{name.replace('_', '-')} must be >= 0") + if args.activity_rows < 2: + parser.error("--activity-rows must be >= 2") + return args + + +@contextmanager +def _database(args: argparse.Namespace) -> Iterator[tuple[str, str]]: + # (psycopg URL, description of where the database runs) + if args.database_url: + url = ( + make_url(args.database_url) + .set(drivername="postgresql+psycopg") + .render_as_string(hide_password=False) + ) + yield url, "given by --database-url" + return + from testcontainers.community.postgres import PostgresContainer + + with PostgresContainer(args.image, driver="psycopg") as pg: + yield ( + pg.get_connection_url(), + f"Docker {args.image} (testcontainers), stock config", + ) + + +def _config(args: argparse.Namespace) -> Config: + return Config( + batch_sizes=list(args.batch_sizes), + iterations=args.iterations, + iterations_large=args.iterations_large, + large_batch=args.large_batch, + warmup=args.warmup, + block=args.block, + activity_rows=args.activity_rows, + read_iterations=args.read_iterations, + read_warmup=args.read_warmup, + ) + + +def _round_trip(engine: Engine, iterations: int) -> RoundTrip: + with engine.connect() as conn: + for _ in range(10): + conn.execute(text("SELECT 1")) + summary = summarize( + [timed(lambda: conn.execute(text("SELECT 1"))) for _ in range(iterations)] + ) + return RoundTrip( + iterations=summary.iterations, p50_ms=summary.p50_ms, p95_ms=summary.p95_ms + ) + + +def _drop(engine: Engine, schemas: list[str]) -> None: + with engine.begin() as conn: + for name in schemas: + conn.execute(text(f'DROP SCHEMA IF EXISTS "{name}" CASCADE')) + + +def _print_tables(report: Report) -> None: + trip = report["round_trip"] + print(f"\nround trip (SELECT 1): p50 {trip['p50_ms']:.3f} ms") + result = report["write"] + if result is not None: + print("\nwrite (ms per transaction)") + print( + f"{'operation':<12}{'n':>5}{'variant':>14}{'p50':>9}{'p95':>9}{'obj/s':>10}" + ) + for case in result["cases"]: + print( + f"{case['operation']:<12}{case['objects']:>5}{case['variant']:>14}" + f"{case['p50_ms']:>9.3f}{case['p95_ms']:>9.3f}" + f"{case['objects_per_s']:>10.0f}{case['statements_per_op']:>7.1f}" + ) + print("\noverhead vs plain") + for row in result["overhead"]: + print( + f"{row['operation']:<12}{row['objects']:>5}{row['variant']:>14}" + f" p50 {row['p50_pct']:+7.1f}% p95 {row['p95_pct']:+7.1f}%" + f" {row['added_p50_ms']:+.3f} ms" + f" {row['added_p50_us_per_object']:.0f} us/object" + ) + reads = report["read"] + if reads is not None: + print("\nread (ms)") + for read_case in reads["cases"]: + print( + f"{read_case['name']:<40}{read_case['p50_ms']:>9.3f}" + f"{read_case['p95_ms']:>9.3f}" + f" partitions {read_case['partitions_scanned']}" + f" statements {len(read_case['statements'])}" + f" plan {read_case['explain_planning_ms']:.2f}" + f" exec {read_case['explain_execution_ms']:.2f}" + ) + + +def run(args: argparse.Namespace) -> tuple[Report, Path]: + """Run the benchmark and write its JSON; return the report and the path.""" + + def progress(message: str) -> None: + if not args.quiet: + print(message, flush=True) + + started = time.perf_counter() + run_id = uuid.uuid4().hex[:8] + app_schema = f"bench_{run_id}_app" + write_schema = f"bench_{run_id}_write" + read_schema = f"bench_{run_id}_read" + with _database(args) as (url, database): + engine = create_engine(url) + try: + progress(f"database: {database}") + env = environment(engine, database) + round_trip = _round_trip(engine, args.iterations) + with engine.begin() as conn: + conn.execute(text(f'CREATE SCHEMA "{app_schema}"')) + write_result = None + read_result = None + try: + if not args.skip_write: + app_engine = engine.execution_options( + schema_translate_map={None: app_schema} + ) + write_result = write.run( + engine, + app_engine, + write_schema, + write.WriteConfig( + batch_sizes=tuple(args.batch_sizes), + iterations=args.iterations, + iterations_large=args.iterations_large, + large_batch=args.large_batch, + warmup=args.warmup, + block=args.block, + ), + progress, + ) + if not args.skip_read: + read_result = read.run( + engine, + read_schema, + read.ReadConfig( + activity_rows=args.activity_rows, + iterations=args.read_iterations, + warmup=args.read_warmup, + ), + progress, + ) + finally: + if not args.keep: + _drop(engine, [app_schema, write_schema, read_schema]) + finally: + engine.dispose() + + report = Report( + schema_version=SCHEMA_VERSION, + created_at=datetime.now(timezone.utc).isoformat(), + duration_s=time.perf_counter() - started, + config=_config(args), + environment=env, + round_trip=round_trip, + write=write_result, + read=read_result, + ) + args.output_dir.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + path = args.output_dir / f"run_{stamp}_{run_id}.json" + path.write_text(json.dumps(report, indent=2, default=str) + "\n", encoding="utf-8") + if not args.quiet: + _print_tables(report) + print(f"\nWrote {path} in {report['duration_s']:.0f} s") + return report, path + + +def main(argv: Sequence[str] | None = None) -> None: + """Command-line entry point.""" + run(parse_args(argv)) + + +if __name__ == "__main__": + main(sys.argv[1:]) diff --git a/benchmark/stats.py b/benchmark/stats.py new file mode 100644 index 0000000..bdb9909 --- /dev/null +++ b/benchmark/stats.py @@ -0,0 +1,42 @@ +"""Latency summaries.""" + +from __future__ import annotations + +import statistics +import time +from collections.abc import Callable +from typing import NamedTuple + + +class Summary(NamedTuple): + """Latency of one case, in milliseconds.""" + + iterations: int + p50_ms: float + p95_ms: float + mean_ms: float + min_ms: float + + +def summarize(seconds: list[float]) -> Summary: + """Summarize measured durations (seconds) as milliseconds. + + Percentiles use ``statistics.quantiles`` with the inclusive method, so a + percentile always lies between the smallest and the largest sample. + """ + if not seconds: + raise ValueError("no samples") + ms = [value * 1000 for value in seconds] + if len(ms) == 1: + p50 = p95 = ms[0] + else: + cuts = statistics.quantiles(ms, n=100, method="inclusive") + p50, p95 = cuts[49], cuts[94] + return Summary(len(ms), p50, p95, statistics.fmean(ms), min(ms)) + + +def timed(call: Callable[[], object]) -> float: + """Run ``call`` once and return its duration in seconds.""" + started = time.perf_counter() + call() + return time.perf_counter() - started diff --git a/benchmark/write.py b/benchmark/write.py new file mode 100644 index 0000000..b274907 --- /dev/null +++ b/benchmark/write.py @@ -0,0 +1,509 @@ +"""Write overhead: the same flush with and without auditing. + +Variants: + +- ``plain``: ``BenchPlain`` (not ``Audited``) on a session class the library + is not installed on. Nothing of the library runs: the true zero. +- ``disabled``: ``BenchAudited`` on the installed session class with + ``session.info["audit_enabled"] = False``. No entries are written, but the + mixin's ``active_history`` set listeners and the session listeners still + run. +- ``audited``: ``BenchAudited`` on the installed session class, capturing + with the default ``on_error="log"``: the audit inserts run inside a + savepoint, so a failed audit write does not fail the transaction. +- ``audited_raise``: the same with ``on_error="raise"``, installed on its + own session class: no savepoint, and a failed audit write fails the + transaction. + +``check_setup`` verifies these preconditions before anything is measured. + +Every timed operation is one database transaction, ended by ``commit()``. +The variants of a case run in interleaved blocks, so drift (table growth, +autovacuum, a noisy neighbour) spreads over all of them. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator, Sequence +from contextlib import contextmanager +from typing import Literal, NamedTuple, TypedDict, cast + +from sqlalchemy import Engine, event, select +from sqlalchemy.orm import Session, sessionmaker + +from audit_trail import Audited, AuditEvent, AuditTrail +from audit_trail._typing import assert_never +from audit_trail.listener import AUDIT_ENABLED_KEY, installed_trail +from audit_trail.maintenance import ensure_partitions +from audit_trail.migrations import create_audit_tables +from benchmark.models import Base, BenchAudited, BenchPlain, bench_events, row_values +from benchmark.stats import summarize, timed + +Operation = Literal["insert", "update", "log", "log_durable"] +Variant = Literal["plain", "disabled", "audited", "audited_raise"] + + +class WriteCase(TypedDict): + """Latency of one operation, batch size and variant.""" + + operation: Operation + objects: int + variant: Variant + iterations: int + p50_ms: float + p95_ms: float + mean_ms: float + min_ms: float + ops_per_s: float + objects_per_s: float + statements_per_op: float + + +class Overhead(TypedDict): + """A variant compared with the ``plain`` baseline.""" + + operation: Operation + objects: int + variant: Variant + baseline: Variant + p50_pct: float + p95_pct: float + added_p50_ms: float + added_p50_us_per_object: float + + +class WriteResult(TypedDict): + """Everything the write benchmark measured.""" + + preconditions: list[str] + cases: list[WriteCase] + overhead: list[Overhead] + + +class WriteConfig(NamedTuple): + """Sizes of the write benchmark.""" + + batch_sizes: tuple[int, ...] + iterations: int + iterations_large: int + large_batch: int + warmup: int + block: int + + +class Sessions(NamedTuple): + """Session factories of the variants.""" + + plain: sessionmaker[Session] + audited: sessionmaker[Session] + audited_raise: sessionmaker[Session] + + +class Setup(NamedTuple): + """The trails and session factories of a write run.""" + + audit: AuditTrail + audit_raise: AuditTrail + sessions: Sessions + + +class SetupError(RuntimeError): + """A variant is not what it claims to be; measuring it would mislead.""" + + +Step = Callable[[], "_Timing"] + + +def setup( + engine: Engine, + app_engine: Engine, + audit_schema: str, + events: Sequence[type[AuditEvent]] | None = None, +) -> Setup: + """Create the model tables and the audit tables; install the listeners. + + ``events`` defaults to ``bench_events()``; pass ``()`` to set up without + defining them (the test suite does, to keep its process free of them). + """ + registered = [bench_events()] if events is None else list(events) + audit = AuditTrail(engine, schema=audit_schema, events=registered) + audit_raise = AuditTrail( + engine, schema=audit_schema, events=registered, on_error="raise" + ) + with engine.begin() as conn: + create_audit_tables(conn, audit.tables, audit.severities) + ensure_partitions(conn, audit.tables, audit.severities) + Base.metadata.create_all(app_engine) + + def factory(name: str) -> sessionmaker[Session]: + # A fresh Session subclass per run: install() is process-wide per class. + return sessionmaker(app_engine, class_=type(name, (Session,), {})) + + sessions = Sessions( + plain=factory("PlainSession"), + audited=factory("AuditedSession"), + audited_raise=factory("AuditedRaiseSession"), + ) + audit.install(sessions.audited) + audit_raise.install(sessions.audited_raise) + return Setup(audit, audit_raise, sessions) + + +class _Statements: + """Counts the statements sent through the cursor (not BEGIN/COMMIT).""" + + def __init__(self, engines: list[Engine]) -> None: + self.count = 0 + self.engines = engines + + def _seen(self, *args: object) -> None: + self.count += 1 + + @contextmanager + def listening(self) -> Iterator[None]: + for engine in self.engines: + event.listen(engine, "before_cursor_execute", self._seen) + try: + yield + finally: + for engine in self.engines: + event.remove(engine, "before_cursor_execute", self._seen) + + +class _Timing(NamedTuple): + seconds: float + statements: int + + +def _timed(statements: _Statements, work: Callable[[], object]) -> _Timing: + # Only the timed work counts: the untimed SELECT of an update does not. + before = statements.count + seconds = timed(work) + return _Timing(seconds, statements.count - before) + + +class _Counter: + def __init__(self) -> None: + self.value = 0 + + def take(self, count: int) -> list[int]: + start = self.value + self.value += count + return list(range(start, start + count)) + + +_AUDITED: tuple[Variant, ...] = ("disabled", "audited", "audited_raise") + + +def _session(sessions: Sessions, variant: Variant) -> Session: + match variant: + case "plain": + return sessions.plain() + case "disabled": + session = sessions.audited() + session.info[AUDIT_ENABLED_KEY] = False + return session + case "audited": + return sessions.audited() + case "audited_raise": + return sessions.audited_raise() + case _: + assert_never(variant) + + +def _model(variant: Variant) -> type[BenchPlain] | type[BenchAudited]: + return BenchPlain if variant == "plain" else BenchAudited + + +def _trail(setup: Setup, variant: Variant) -> AuditTrail | None: + # The trail a variant's log() goes through: the one installed on its + # session class. + return setup.audit_raise if variant == "audited_raise" else setup.audit + + +PRECONDITIONS = ( + "plain model is not Audited", + "other variants use the Audited model", + "plain session has no trail installed", + "disabled session has the on_error='log' trail and capture off", + "audited session has the on_error='log' trail and capture on", + "audited_raise session has the on_error='raise' trail and capture on", +) +"""What ``check_setup`` verifies, in order.""" + + +def check_setup(setup: Setup) -> list[str]: + """Verify that every variant is what it claims to be, before measuring. + + Checks the models and sessions the steps actually use (``_model`` and + ``_session``), so a mistake there fails the run instead of producing a + wrong baseline. + + Args: + setup: What ``setup`` returned. + + Returns: + ``PRECONDITIONS``, all verified. + + Raises: + SetupError: A precondition does not hold. + """ + + def installed(variant: Variant, trail: AuditTrail | None, on: bool) -> bool: + with _session(setup.sessions, variant) as session: + enabled = session.info.get(AUDIT_ENABLED_KEY) is not False + return installed_trail(session) is trail and enabled is on + + checks = ( + not issubclass(_model("plain"), Audited), + all(issubclass(_model(variant), Audited) for variant in _AUDITED), + installed("plain", None, True), + installed("disabled", setup.audit, False) and setup.audit.on_error == "log", + installed("audited", setup.audit, True), + installed("audited_raise", setup.audit_raise, True) + and setup.audit_raise.on_error == "raise", + ) + for name, ok in zip(PRECONDITIONS, checks, strict=True): + if not ok: + raise SetupError(f"benchmark precondition failed: {name}") + return list(PRECONDITIONS) + + +def _insert_step( + sessions: Sessions, + variant: Variant, + objects: int, + counter: _Counter, + statements: _Statements, +) -> Step: + model = _model(variant) + + def step() -> _Timing: + rows = [model(**row_values(i)) for i in counter.take(objects)] + with _session(sessions, variant) as session: + + def work() -> None: + session.add_all(rows) + session.commit() + + return _timed(statements, work) + + return step + + +def _update_step( + sessions: Sessions, variant: Variant, objects: int, statements: _Statements +) -> Step: + model = _model(variant) + with sessions.plain() as session: + rows = [model(**row_values(i)) for i in range(objects)] + session.add_all(rows) + session.commit() + ids = [row.id for row in rows] + + def step() -> _Timing: + with _session(sessions, variant) as session: + # Loaded in their own transaction first, untimed: the timed part + # is the change and its flush, not the SELECT. + loaded = cast( + "Sequence[BenchPlain | BenchAudited]", + session.scalars(select(model).where(model.id.in_(ids))).all(), + ) + + def work() -> None: + for row in loaded: + row.counter += 1 + row.status = "sent" if row.status == "draft" else "draft" + session.commit() + + return _timed(statements, work) + + return step + + +def _log_step( + setup: Setup, + variant: Variant, + durable: bool, + counter: _Counter, + statements: _Statements, +) -> Step: + # One plain row, as a request would write, then the event. + event = bench_events()["EXPORTED" if durable else "VIEWED"] + + def step() -> _Timing: + row = BenchPlain(**row_values(counter.take(1)[0])) + trail = None if variant == "plain" else _trail(setup, variant) + with _session(setup.sessions, variant) as session: + + def work() -> None: + session.add(row) + if trail is not None: + trail.log(session, event, payload={"format": "csv"}) + session.commit() + + return _timed(statements, work) + + return step + + +class _Samples(NamedTuple): + seconds: list[float] + statements: int + + +def _measure( + steps: dict[Variant, Step], + iterations: int, + warmup: int, + block: int, +) -> dict[Variant, _Samples]: + for step in steps.values(): + for _ in range(warmup): + step() + seconds: dict[Variant, list[float]] = {variant: [] for variant in steps} + counts: dict[Variant, int] = dict.fromkeys(steps, 0) + done = 0 + while done < iterations: + size = min(block, iterations - done) + for variant, step in steps.items(): + for _ in range(size): + timing = step() + seconds[variant].append(timing.seconds) + counts[variant] += timing.statements + done += size + return {variant: _Samples(seconds[variant], counts[variant]) for variant in steps} + + +def _case( + operation: Operation, objects: int, variant: Variant, samples: _Samples +) -> WriteCase: + summary = summarize(samples.seconds) + ops = 1000 / summary.mean_ms + return WriteCase( + operation=operation, + objects=objects, + variant=variant, + iterations=summary.iterations, + p50_ms=summary.p50_ms, + p95_ms=summary.p95_ms, + mean_ms=summary.mean_ms, + min_ms=summary.min_ms, + ops_per_s=ops, + objects_per_s=ops * objects, + statements_per_op=samples.statements / summary.iterations, + ) + + +def overhead(cases: list[WriteCase]) -> list[Overhead]: + """Compare every non-``plain`` case with the ``plain`` case of its kind.""" + base = { + (case["operation"], case["objects"]): case + for case in cases + if case["variant"] == "plain" + } + result: list[Overhead] = [] + for case in cases: + plain = base.get((case["operation"], case["objects"])) + if case["variant"] == "plain" or plain is None: + continue + added = case["p50_ms"] - plain["p50_ms"] + result.append( + Overhead( + operation=case["operation"], + objects=case["objects"], + variant=case["variant"], + baseline="plain", + p50_pct=100 * added / plain["p50_ms"], + p95_pct=100 * (case["p95_ms"] - plain["p95_ms"]) / plain["p95_ms"], + added_p50_ms=added, + added_p50_us_per_object=1000 * added / case["objects"], + ) + ) + return result + + +@contextmanager +def _context(audit: AuditTrail) -> Iterator[None]: + with audit.context( + actor_type="user", + actor_id="42", + actor_label="bench@example.com", + channel="api", + path="/bench", + method="POST", + ): + yield + + +def run( + engine: Engine, + app_engine: Engine, + audit_schema: str, + config: WriteConfig, + progress: Callable[[str], None], +) -> WriteResult: + """Measure insert, update and ``log()`` with and without auditing.""" + run_setup = setup(engine, app_engine, audit_schema) + preconditions = check_setup(run_setup) + audit, sessions = run_setup.audit, run_setup.sessions + durable_engines = [audit.durable_engine, run_setup.audit_raise.durable_engine] + statements = _Statements( + [engine, *(e for e in durable_engines if isinstance(e, Engine))] + ) + counter = _Counter() + cases: list[WriteCase] = [] + variants: tuple[Variant, ...] = ("plain", "disabled", "audited", "audited_raise") + try: + with _context(audit), statements.listening(): + for operation in ("insert", "update"): + for objects in config.batch_sizes: + iterations = ( + config.iterations_large + if objects >= config.large_batch + else config.iterations + ) + progress(f"write: {operation} x{objects} ({iterations} it)") + steps: dict[Variant, Step] = { + variant: ( + _insert_step( + sessions, variant, objects, counter, statements + ) + if operation == "insert" + else _update_step(sessions, variant, objects, statements) + ) + for variant in variants + } + samples = _measure(steps, iterations, config.warmup, config.block) + cases.extend( + _case(operation, objects, variant, samples[variant]) + for variant in variants + ) + log_ops: tuple[tuple[Operation, bool], ...] = ( + ("log", False), + ("log_durable", True), + ) + for log_op, durable in log_ops: + progress(f"write: {log_op} ({config.iterations} it)") + log_variants: tuple[Variant, ...] = ( + "plain", + "audited", + "audited_raise", + ) + steps = { + variant: _log_step(run_setup, variant, durable, counter, statements) + for variant in log_variants + } + samples = _measure( + steps, config.iterations, config.warmup, config.block + ) + cases.extend( + _case(log_op, 1, variant, samples[variant]) + for variant in log_variants + ) + finally: + audit.dispose() + run_setup.audit_raise.dispose() + return WriteResult( + preconditions=preconditions, cases=cases, overhead=overhead(cases) + ) diff --git a/pyproject.toml b/pyproject.toml index 60acea7..ab60afd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,7 +94,7 @@ typing-modules = ["audit_trail._typing"] [tool.mypy] python_version = "3.10" -files = ["audit_trail", "tests"] +files = ["audit_trail", "benchmark", "tests"] strict = true warn_unreachable = true show_error_codes = true diff --git a/tests/db/test_benchmark.py b/tests/db/test_benchmark.py new file mode 100644 index 0000000..d05657e --- /dev/null +++ b/tests/db/test_benchmark.py @@ -0,0 +1,197 @@ +"""The benchmark runner, at tiny sizes, so it does not rot between full runs. + +The run itself goes through the command line in a subprocess: it defines the +benchmark's events, and ``AuditTrail(events=None)`` would discover them in +this process. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import uuid +from collections.abc import Iterator +from pathlib import Path + +import pytest +from sqlalchemy import Engine, text + +from audit_trail import AuditTrail +from benchmark import write +from benchmark.models import BenchAudited +from benchmark.run_benchmark import ROOT, SCHEMA_VERSION, Report + +BATCHES = [1, 3] + + +def test_benchmark_runs_and_writes_its_json( + database_url: str, engine: Engine, tmp_path: Path +) -> None: + command = [ + sys.executable, + "-m", + "benchmark.run_benchmark", + f"--database-url={database_url}", + f"--output-dir={tmp_path}", + "--batch-sizes=1,3", + "--iterations=3", + "--iterations-large=2", + "--large-batch=3", + "--warmup=1", + "--block=2", + "--activity-rows=2000", + "--read-iterations=2", + "--read-warmup=1", + "--quiet", + ] + subprocess.run(command, cwd=ROOT, check=True, timeout=300) + + [path] = tmp_path.glob("run_*.json") + written = json.loads(path.read_text(encoding="utf-8")) + assert written["schema_version"] == SCHEMA_VERSION + assert set(written) == set(Report.__annotations__) + assert written["environment"]["postgres"]["version"] + assert written["round_trip"]["p50_ms"] > 0 + + writes = written["write"] + assert writes["preconditions"] == list(write.PRECONDITIONS) + kinds = {(c["operation"], c["objects"], c["variant"]) for c in writes["cases"]} + expected = { + (operation, objects, variant) + for operation in ("insert", "update") + for objects in BATCHES + for variant in ("plain", "disabled", "audited", "audited_raise") + } | { + (operation, 1, variant) + for operation in ("log", "log_durable") + for variant in ("plain", "audited", "audited_raise") + } + assert kinds == expected + for case in writes["cases"]: + large = case["objects"] >= 3 and case["operation"] in ("insert", "update") + assert case["iterations"] == (2 if large else 3) + assert 0 < case["min_ms"] <= case["p50_ms"] <= case["p95_ms"] + assert case["statements_per_op"] >= 1 + by_kind = {(c["operation"], c["objects"], c["variant"]): c for c in writes["cases"]} + # Capturing writes the transaction row and the entries: more statements. + for operation in ("insert", "update"): + for objects in BATCHES: + plain = by_kind[operation, objects, "plain"]["statements_per_op"] + disabled = by_kind[operation, objects, "disabled"]["statements_per_op"] + audited = by_kind[operation, objects, "audited"]["statements_per_op"] + raising = by_kind[operation, objects, "audited_raise"] + assert disabled == plain + assert audited > plain + # on_error="raise" skips the savepoint and its release. + assert raising["statements_per_op"] == audited - 2 + overhead = { + (o["operation"], o["objects"], o["variant"]) for o in writes["overhead"] + } + assert overhead == {kind for kind in expected if kind[2] != "plain"} + + reads = written["read"] + dataset = reads["dataset"] + assert dataset["activity_rows"] >= 2000 + assert dataset["transaction_rows"] == 1000 + assert dataset["partitions"]["activity"] == 4 * 16 + assert dataset["seed_s"] > 0 + names = [case["name"] for case in reads["cases"]] + assert "list_groups first page" in names + assert "list_groups first page, 30-day window" in names + assert "list_groups next page" in names + assert "list_groups by actor" in names + assert "object_history" in names + assert "access_summary" in names + for case in reads["cases"]: + assert 0 < case["p50_ms"] <= case["p95_ms"] + assert case["statements"], case["name"] + assert case["partitions_scanned"] >= 1 + for statement in case["statements"]: + assert statement["sql"].startswith("SELECT") + assert statement["planning_ms"] >= 0 + assert statement["execution_ms"] >= 0 + assert ( + len(statement["partitions_scanned"]) <= statement["partitions_in_plan"] + ) + first = next(c for c in reads["cases"] if c["name"] == "list_groups first page") + assert first["groups"] == 50 + + run_id = path.stem.rsplit("_", 1)[1] + with engine.connect() as conn: + left = conn.execute( + text("SELECT nspname FROM pg_namespace WHERE nspname LIKE :p"), + {"p": f"bench_{run_id}_%"}, + ).all() + assert left == [] + + +@pytest.fixture +def write_setup(engine: Engine) -> Iterator[write.Setup]: + name = f"bench_{uuid.uuid4().hex[:8]}" + with engine.begin() as conn: + conn.execute(text(f'CREATE SCHEMA "{name}_app"')) + app_engine = engine.execution_options(schema_translate_map={None: f"{name}_app"}) + setup = write.setup(engine, app_engine, f"{name}_write", events=()) + yield setup + setup.audit.dispose() + setup.audit_raise.dispose() + with engine.begin() as conn: + for schema in (f"{name}_app", f"{name}_write"): + conn.execute(text(f'DROP SCHEMA "{schema}" CASCADE')) + + +def test_check_setup_accepts_the_real_setup(write_setup: write.Setup) -> None: + assert write.check_setup(write_setup) == list(write.PRECONDITIONS) + + +@pytest.mark.parametrize( + ("swap", "failed"), + [ + ("plain", "plain session has no trail installed"), + # disabled and audited share a factory; disabled is checked first. + ("audited", "disabled session has the on_error='log' trail"), + ("audited_raise", "audited_raise session has the on_error='raise' trail"), + ], +) +def test_check_setup_rejects_a_wrong_session( + write_setup: write.Setup, swap: str, failed: str +) -> None: + # Each variant gets another variant's session factory. + sessions = write_setup.sessions + others = { + "plain": sessions.audited, + "audited": sessions.plain, + "audited_raise": sessions.audited, + } + broken = write_setup._replace(sessions=sessions._replace(**{swap: others[swap]})) + with pytest.raises(write.SetupError, match=failed): + write.check_setup(broken) + + +def test_check_setup_rejects_an_audited_baseline( + write_setup: write.Setup, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(write, "_model", lambda variant: BenchAudited) + with pytest.raises(write.SetupError, match="plain model is not Audited"): + write.check_setup(write_setup) + + +def test_importing_the_benchmark_defines_no_events(engine: Engine) -> None: + # This module imported every benchmark module. Discovery must not find + # benchmark events in this process: with a host's own severity enum, + # their built-in Severity would make AuditTrail(events=None) fail. + trail = AuditTrail(engine, events=None) + assert not [v for v in trail.registry.events if v.startswith("benchmark.")] + + +def test_bench_events_use_their_own_prefix_and_the_default_severity() -> None: + check = ( + "from audit_trail import Severity\n" + "from benchmark.models import bench_events\n" + "members = list(bench_events())\n" + "assert {m.value for m in members} == " + "{'benchmark.viewed', 'benchmark.exported'}\n" + "assert all(isinstance(m.severity, Severity) for m in members)\n" + ) + subprocess.run([sys.executable, "-c", check], cwd=ROOT, check=True, timeout=60)