diff --git a/cmd/bench/metrics/Makefile b/cmd/bench/metrics/Makefile new file mode 100644 index 000000000..cbaf96dda --- /dev/null +++ b/cmd/bench/metrics/Makefile @@ -0,0 +1,83 @@ +# Metrics Benchmarks Makefile +# Usage: cd cmd/bench/metrics && make + +OUTPOST_ROOT := $(shell git rev-parse --show-toplevel) +PG_URL := postgres://outpost:outpost@localhost:5488/bench?sslmode=disable +CH_ADDR := localhost:9009 +CH_DB := bench +PG_ROWS := 10000000 +CH_ROWS := 10000000 + +# ── Infra ──────────────────────────────────────────────────────────────────── + +up/pg: + docker compose -f pg/docker-compose.yml up -d + +up/ch: + docker compose -f ch/docker-compose.yml up -d + +up: up/pg up/ch + +down/pg: + docker compose -f pg/docker-compose.yml down -v + +down/ch: + docker compose -f ch/docker-compose.yml down -v + +down: down/pg down/ch + +# ── Migrations ─────────────────────────────────────────────────────────────── + +migrate/pg: + cd $(OUTPOST_ROOT) && BENCH_PG_URL="$(PG_URL)" \ + go test -run='^$$' -bench=BenchmarkPG -benchtime=1x ./cmd/bench/metrics/ + +migrate/ch: + cd $(OUTPOST_ROOT) && BENCH_CH_ADDR="$(CH_ADDR)" BENCH_CH_DB="$(CH_DB)" \ + go test -run='^$$' -bench=BenchmarkCH -benchtime=1x ./cmd/bench/metrics/ + +migrate: migrate/pg migrate/ch + +# ── Seeding ────────────────────────────────────────────────────────────────── + +seed/pg: + psql "$(PG_URL)" -v ROWS=$(PG_ROWS) -f pg/seed.sql + +seed/ch: + clickhouse client --port 9009 --database $(CH_DB) \ + --param_rows $(CH_ROWS) < ch/seed.sql + +# ── Benchmarks ─────────────────────────────────────────────────────────────── + +bench/pg: + cd $(OUTPOST_ROOT) && BENCH_PG_URL="$(PG_URL)" \ + go test -bench=BenchmarkPG -benchtime=1x -count=1 -timeout=30m ./cmd/bench/metrics/ + +bench/ch: + cd $(OUTPOST_ROOT) && BENCH_CH_ADDR="$(CH_ADDR)" BENCH_CH_DB="$(CH_DB)" \ + go test -bench=BenchmarkCH -benchtime=1x -count=1 -timeout=30m ./cmd/bench/metrics/ + +bench/pg/sustained: + cd $(OUTPOST_ROOT) && BENCH_PG_URL="$(PG_URL)" \ + go test -bench=BenchmarkPG -benchtime=10s -count=3 -timeout=30m ./cmd/bench/metrics/ + +bench/ch/sustained: + cd $(OUTPOST_ROOT) && BENCH_CH_ADDR="$(CH_ADDR)" BENCH_CH_DB="$(CH_DB)" \ + go test -bench=BenchmarkCH -benchtime=10s -count=3 -timeout=30m ./cmd/bench/metrics/ + +bench: bench/pg bench/ch + +# ── Full workflow ──────────────────────────────────────────────────────────── + +setup/pg: up/pg migrate/pg seed/pg +setup/ch: up/ch migrate/ch seed/ch +setup: setup/pg setup/ch + +reset/pg: down/pg up/pg migrate/pg seed/pg +reset/ch: down/ch up/ch migrate/ch seed/ch + +.PHONY: up/pg up/ch up down/pg down/ch down \ + migrate/pg migrate/ch migrate \ + seed/pg seed/ch \ + bench/pg bench/ch bench/pg/sustained bench/ch/sustained bench \ + setup/pg setup/ch setup reset/pg reset/ch diff --git a/cmd/bench/metrics/README.md b/cmd/bench/metrics/README.md new file mode 100644 index 000000000..370c5b2eb --- /dev/null +++ b/cmd/bench/metrics/README.md @@ -0,0 +1,63 @@ +# Metrics Benchmarks + +Benchmarks `QueryEventMetrics` / `QueryAttemptMetrics` against PostgreSQL and ClickHouse. + +Shared test cases in `bench_test.go`, backend-specific setup in `pg_test.go` / `ch_test.go`. + +## Quick Start + +```bash +cd outpost/cmd/bench/metrics + +# ── ClickHouse ─────────────────────────────── +make setup/ch # up + migrate + seed (10M) +make bench/ch # single iteration +make bench/ch/sustained # 10s x 3 runs +make down/ch # cleanup + +# ── PostgreSQL ─────────────────────────────── +make setup/pg # up + migrate + seed (10M) +make bench/pg # single iteration +make bench/pg/sustained # 10s x 3 runs +make down/pg # cleanup + +# ── Both ───────────────────────────────────── +make setup # setup both +make bench # bench both +make down # cleanup both +``` + +### Individual steps + +```bash +make up/ch # start container +make migrate/ch # run migrations +make seed/ch # seed data (default 10M, override: make seed/ch CH_ROWS=1000000000) +make bench/ch # run benchmarks +make reset/ch # down + up + migrate + seed (fresh start) +``` + +Same targets available for `/pg`. + +## Structure + +``` +metrics/ + Makefile # all commands + bench_test.go # shared test cases + date ranges + helpers + pg_test.go # PG setup (BENCH_PG_URL) + ch_test.go # CH setup (BENCH_CH_ADDR) + pg/ # PG infra (docker-compose, seed.sql) + ch/ # CH infra (docker-compose, seed.sql, config/) +``` + +## Data Distribution + +Deterministic via modulo arithmetic (identical for both backends): + +- **2 tenants** — `tenant_0` (90%), `tenant_1` (10%) +- **500 destinations** — `dest_0` through `dest_499` +- **3 topics** — `order.created`, `order.updated`, `payment.received` +- **Time** — evenly spread across January 2000 +- **Attempts** — chained retries (1 event -> 1-4 attempts), 0.5% permanently failed +- 10M events -> ~12.6M attempts (22.6M total rows) diff --git a/cmd/bench/metrics/bench_test.go b/cmd/bench/metrics/bench_test.go new file mode 100644 index 000000000..adc6293df --- /dev/null +++ b/cmd/bench/metrics/bench_test.go @@ -0,0 +1,465 @@ +package metrics + +import ( + "context" + "testing" + "time" + + "github.com/hookdeck/outpost/internal/logstore/driver" +) + +// ── Time ranges ───────────────────────────────────────────────────────────── + +var ( + // Full month — all seeded data lives here. + fullMonth = driver.TimeRange{ + Start: time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), + End: time.Date(2000, 2, 1, 0, 0, 0, 0, time.UTC), + } + oneDay = driver.TimeRange{ + Start: time.Date(2000, 1, 15, 0, 0, 0, 0, time.UTC), + End: time.Date(2000, 1, 16, 0, 0, 0, 0, time.UTC), + } + oneWeek = driver.TimeRange{ + Start: time.Date(2000, 1, 8, 0, 0, 0, 0, time.UTC), + End: time.Date(2000, 1, 15, 0, 0, 0, 0, time.UTC), + } +) + +// ── Helpers ───────────────────────────────────────────────────────────────── + +func hourly() *driver.Granularity { return &driver.Granularity{Value: 1, Unit: "h"} } +func daily() *driver.Granularity { return &driver.Granularity{Value: 1, Unit: "d"} } +func twoDays() *driver.Granularity { return &driver.Granularity{Value: 2, Unit: "d"} } +func weekly() *driver.Granularity { return &driver.Granularity{Value: 1, Unit: "w"} } +func monthly() *driver.Granularity { return &driver.Granularity{Value: 1, Unit: "M"} } + +func tenant0() map[string][]string { return map[string][]string{"tenant_id": {"tenant_0"}} } + +func withTenant0(extra map[string][]string) map[string][]string { + m := tenant0() + for k, v := range extra { + m[k] = v + } + return m +} + +// ── Event Benchmarks ──────────────────────────────────────────────────────── + +var eventCases = []struct { + name string + req driver.MetricsRequest +}{ + { + name: "CountAll", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "RateAll", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"rate"}, + Filters: tenant0(), + }, + }, + { + name: "CountAndRate", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count", "rate"}, + Filters: tenant0(), + }, + }, + { + name: "CountByTopic", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"topic"}, + Filters: tenant0(), + }, + }, + { + name: "CountByDestination", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"destination_id"}, + Filters: tenant0(), + }, + }, + { + name: "CountByTenant", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"tenant_id"}, + }, + }, + { + name: "Hourly_1Day", + req: driver.MetricsRequest{ + TimeRange: oneDay, + Granularity: hourly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Hourly_1Week", + req: driver.MetricsRequest{ + TimeRange: oneWeek, + Granularity: hourly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Daily_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: daily(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "TwoDays_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: twoDays(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Weekly_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: weekly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Monthly_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: monthly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "RateHourly_1Day", + req: driver.MetricsRequest{ + TimeRange: oneDay, + Granularity: hourly(), + Measures: []string{"rate"}, + Filters: tenant0(), + }, + }, + { + name: "FilterByTopic", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"topic": {"order.created"}}), + }, + }, + { + name: "FilterByDestination", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"destination_id": {"dest_0"}}), + }, + }, + { + name: "SmallTenant", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: daily(), + Measures: []string{"count"}, + Filters: map[string][]string{"tenant_id": {"tenant_1"}}, + }, + }, +} + +// ── Attempt Benchmarks ────────────────────────────────────────────────────── + +var attemptCases = []struct { + name string + req driver.MetricsRequest +}{ + { + name: "CountAll", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "RateAll", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"rate"}, + Filters: tenant0(), + }, + }, + { + name: "SuccessfulRate", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"successful_rate"}, + Filters: tenant0(), + }, + }, + { + name: "FailedRate", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"failed_rate"}, + Filters: tenant0(), + }, + }, + { + name: "CountByTopic", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"topic"}, + Filters: tenant0(), + }, + }, + { + name: "CountByDestination", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"destination_id"}, + Filters: tenant0(), + }, + }, + { + name: "CountByStatus", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"status"}, + Filters: tenant0(), + }, + }, + { + name: "CountByCode", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"code"}, + Filters: tenant0(), + }, + }, + { + name: "CountByAttemptNumber", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"attempt_number"}, + Filters: tenant0(), + }, + }, + { + name: "Hourly_1Day", + req: driver.MetricsRequest{ + TimeRange: oneDay, + Granularity: hourly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Hourly_1Week", + req: driver.MetricsRequest{ + TimeRange: oneWeek, + Granularity: hourly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Daily_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: daily(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "TwoDays_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: twoDays(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "Weekly_1Month", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: weekly(), + Measures: []string{"count"}, + Filters: tenant0(), + }, + }, + { + name: "AllMeasures", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{ + "count", + "successful_count", + "failed_count", + "error_rate", + "first_attempt_count", + "retry_count", + "manual_retry_count", + "avg_attempt_number", + "rate", + "successful_rate", + "failed_rate", + }, + Filters: tenant0(), + }, + }, + { + name: "AllMeasures_Daily", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: daily(), + Measures: []string{ + "count", + "successful_count", + "failed_count", + "error_rate", + "rate", + "successful_rate", + "failed_rate", + }, + Filters: tenant0(), + }, + }, + { + name: "FilterByStatus", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"status": {"failed"}}), + }, + }, + { + name: "FilterByCode", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"code": {"500"}}), + }, + }, + { + name: "FilterByManual", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"manual": {"true"}}), + }, + }, + { + name: "FilterByAttemptNumber", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"attempt_number": {"0"}}), + }, + }, + { + name: "FilterByTopic", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{"topic": {"order.created"}}), + }, + }, + { + name: "MultiDimension", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Dimensions: []string{"topic", "destination_id", "status"}, + Filters: tenant0(), + }, + }, + { + name: "MultiFilter", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Measures: []string{"count"}, + Filters: withTenant0(map[string][]string{ + "status": {"failed"}, + "topic": {"order.created"}, + }), + }, + }, + { + name: "SmallTenant", + req: driver.MetricsRequest{ + TimeRange: fullMonth, + Granularity: daily(), + Measures: []string{"count"}, + Filters: map[string][]string{"tenant_id": {"tenant_1"}}, + }, + }, +} + +func benchmarkEventMetrics(b *testing.B, store driver.Metrics) { + ctx := context.Background() + + for _, tc := range eventCases { + b.Run(tc.name, func(b *testing.B) { + // Warm up. + if _, err := store.QueryEventMetrics(ctx, tc.req); err != nil { + b.Fatalf("warmup: %v", err) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := store.QueryEventMetrics(ctx, tc.req); err != nil { + b.Fatalf("query: %v", err) + } + } + }) + } +} + +func benchmarkAttemptMetrics(b *testing.B, store driver.Metrics) { + ctx := context.Background() + + for _, tc := range attemptCases { + b.Run(tc.name, func(b *testing.B) { + // Warm up. + if _, err := store.QueryAttemptMetrics(ctx, tc.req); err != nil { + b.Fatalf("warmup: %v", err) + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := store.QueryAttemptMetrics(ctx, tc.req); err != nil { + b.Fatalf("query: %v", err) + } + } + }) + } +} diff --git a/cmd/bench/metrics/ch/README.md b/cmd/bench/metrics/ch/README.md new file mode 100644 index 000000000..77bfff5fc --- /dev/null +++ b/cmd/bench/metrics/ch/README.md @@ -0,0 +1,65 @@ +# CH Metrics Benchmark + +Benchmarks `QueryEventMetrics` / `QueryAttemptMetrics` against ClickHouse (2 CPU, 8GB). + +## Prerequisites + +- Docker (Compose v2) +- Go 1.24+ +- `clickhouse` CLI + +## Quick Start + +```bash +cd outpost + +# 1. Start CH +docker compose -f cmd/bench/metrics/ch/docker-compose.yml up -d + +# 2. Run migrations +BENCH_CH_ADDR="localhost:9009" \ + go test -run='^$' -bench=BenchmarkCH -benchtime=1x ./cmd/bench/metrics/ + +# 3. Seed (default 10M — adjust --param_rows N) +clickhouse client --port 9009 --database bench \ + --param_rows 10000000 < cmd/bench/metrics/ch/seed.sql + +# 4a. Single iteration +BENCH_CH_ADDR="localhost:9009" \ + go test -bench=BenchmarkCH -benchtime=1x -count=1 -timeout=30m ./cmd/bench/metrics/ + +# 4b. Sustained (10s x 3 runs) +BENCH_CH_ADDR="localhost:9009" \ + go test -bench=BenchmarkCH -benchtime=10s -count=3 -timeout=30m ./cmd/bench/metrics/ + +# 5. Cleanup +docker compose -f cmd/bench/metrics/ch/docker-compose.yml down -v +``` + +## Re-seeding + +```bash +docker compose -f cmd/bench/metrics/ch/docker-compose.yml down -v +docker compose -f cmd/bench/metrics/ch/docker-compose.yml up -d +# Repeat steps 2-4 +``` + +## Data Distribution + +Deterministic via modulo arithmetic (shared with PG bench): + +- **2 tenants** — `tenant_0` (90%), `tenant_1` (10%) +- **500 destinations** — `dest_0` through `dest_499` +- **3 topics** — `order.created`, `order.updated`, `payment.received` +- **Time** — evenly spread across January 2000 +- **Attempts** — chained retries (1 event -> 1-4 attempts), 0.5% permanently failed +- 10M events -> ~12.6M attempts + +## Resource Tuning + +| Setting | Default | Purpose | +|---------|---------|---------| +| CPUs | 2 | Parallel query threads | +| Memory | 8GB | Container limit | +| max_memory_usage | 6GB | Per-query memory limit | +| max_threads | 2 | Query parallelism | diff --git a/cmd/bench/metrics/ch/config/users.xml b/cmd/bench/metrics/ch/config/users.xml new file mode 100644 index 000000000..797d914bd --- /dev/null +++ b/cmd/bench/metrics/ch/config/users.xml @@ -0,0 +1,8 @@ + + + + 6000000000 + 2 + + + diff --git a/cmd/bench/metrics/ch/docker-compose.yml b/cmd/bench/metrics/ch/docker-compose.yml new file mode 100644 index 000000000..f7f4cc1e3 --- /dev/null +++ b/cmd/bench/metrics/ch/docker-compose.yml @@ -0,0 +1,26 @@ +services: + clickhouse: + image: clickhouse/clickhouse-server:24 + environment: + CLICKHOUSE_DB: bench + CLICKHOUSE_USER: default + CLICKHOUSE_PASSWORD: "" + ports: + - "9009:9000" # native protocol + - "8124:8123" # HTTP + deploy: + resources: + limits: + cpus: "2" + memory: 8g + ulimits: + nofile: + soft: 262144 + hard: 262144 + volumes: + - ./config/users.xml:/etc/clickhouse-server/users.d/bench.xml:ro + healthcheck: + test: ["CMD-SHELL", "clickhouse-client --query 'SELECT 1'"] + interval: 2s + timeout: 5s + retries: 10 diff --git a/cmd/bench/metrics/ch/seed.sql b/cmd/bench/metrics/ch/seed.sql new file mode 100644 index 000000000..695bd0b72 --- /dev/null +++ b/cmd/bench/metrics/ch/seed.sql @@ -0,0 +1,245 @@ +-- seed.sql — Deterministic bulk seeding for CH metrics benchmarks. +-- +-- Usage: +-- clickhouse client --port 9009 --database bench --param_rows 1000000000 < cmd/bench/metrics/ch/seed.sql +-- +-- Default rows = 1000000000 (1B). Override with --param_rows N. +-- +-- Distribution (same as PG bench): +-- 2 tenants: tenant_0 gets 90%, tenant_1 gets 10% +-- Time: evenly spread across January 2000 (2000-01-01 to 2000-02-01) +-- +-- Attempt chain (1 event -> 1-4 attempts): +-- attempt 0: all events. Failed if n%5=0 (20%) +-- attempt 1: failed attempt 0. Failed if n%20=0 (25% of retries) +-- attempt 2: failed attempt 1. Failed if n%100=0 (20% of retries) +-- attempt 3: failed attempt 2. Failed if n%200=0 (50% of retries) +-- +-- For 1B events -> ~1.26B attempts. 0.5% events permanently failed. + +SELECT concat('Seeding ', toString({rows:UInt64}), ' events + chained attempts...') AS message; + +-- ============================================================================ +-- 1. Bulk INSERT into events +-- ============================================================================ +-- +-- Tenants: n%10 == 0 -> tenant_1 (10%), else tenant_0 (90%) +-- Destinations: dest_(n%500) [500 destinations] +-- Topics: n%3 -> order.created / order.updated / payment.received +-- Time: Even spread across 2000-01-01 to 2000-02-01 +-- eligible_for_retry: n%3 != 2 + +SELECT '[1/7] Inserting events...' AS message; + +INSERT INTO events (event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data) +SELECT + concat('evt_', toString(number)) AS event_id, + if(number % 10 = 0, 'tenant_1', 'tenant_0') AS tenant_id, + concat('dest_', toString(number % 500)) AS destination_id, + multiIf( + number % 3 = 0, 'order.created', + number % 3 = 1, 'order.updated', + 'payment.received' + ) AS topic, + number % 3 != 2 AS eligible_for_retry, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) AS event_time, + '{}' AS metadata, + '{}' AS data +FROM numbers({rows:UInt64}); + +-- ============================================================================ +-- 2. Bulk INSERT into attempts (chained retries) +-- ============================================================================ +-- +-- Each attempt's time = event_time + (attempt_number * 1 second). +-- manual: only attempt_number >= 2 AND n%10=9 (10% of late retries). +-- Code: success->200/201, failed->500/422 (alternating on n%2). + +SELECT '[2/7] Inserting attempt 0 (all events)...' AS message; + +INSERT INTO attempts ( + event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data, + attempt_id, status, attempt_time, code, response_data, manual, attempt_number +) +SELECT + concat('evt_', toString(number)) AS event_id, + if(number % 10 = 0, 'tenant_1', 'tenant_0') AS tenant_id, + concat('dest_', toString(number % 500)) AS destination_id, + multiIf( + number % 3 = 0, 'order.created', + number % 3 = 1, 'order.updated', + 'payment.received' + ) AS topic, + number % 3 != 2 AS eligible_for_retry, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) AS event_time, + '{}' AS metadata, + '{}' AS data, + concat('att_', toString(number), '_0') AS attempt_id, + if(number % 5 = 0, 'failed', 'success') AS status, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) + + toIntervalSecond(1) AS attempt_time, + multiIf( + number % 5 != 0 AND number % 2 = 0, '200', + number % 5 != 0, '201', + number % 2 = 0, '500', + '422' + ) AS code, + '' AS response_data, + false AS manual, + toUInt32(0) AS attempt_number +FROM numbers({rows:UInt64}); + +SELECT '[3/7] Inserting attempt 1 (20% of events)...' AS message; + +INSERT INTO attempts ( + event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data, + attempt_id, status, attempt_time, code, response_data, manual, attempt_number +) +SELECT + concat('evt_', toString(number)) AS event_id, + if(number % 10 = 0, 'tenant_1', 'tenant_0') AS tenant_id, + concat('dest_', toString(number % 500)) AS destination_id, + multiIf( + number % 3 = 0, 'order.created', + number % 3 = 1, 'order.updated', + 'payment.received' + ) AS topic, + number % 3 != 2 AS eligible_for_retry, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) AS event_time, + '{}' AS metadata, + '{}' AS data, + concat('att_', toString(number), '_1') AS attempt_id, + if(number % 20 = 0, 'failed', 'success') AS status, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) + + toIntervalSecond(2) AS attempt_time, + multiIf( + number % 20 != 0 AND number % 2 = 0, '200', + number % 20 != 0, '201', + number % 2 = 0, '500', + '422' + ) AS code, + '' AS response_data, + false AS manual, + toUInt32(1) AS attempt_number +FROM numbers({rows:UInt64}) +WHERE number % 5 = 0; + +SELECT '[4/7] Inserting attempt 2 (5% of events)...' AS message; + +INSERT INTO attempts ( + event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data, + attempt_id, status, attempt_time, code, response_data, manual, attempt_number +) +SELECT + concat('evt_', toString(number)) AS event_id, + if(number % 10 = 0, 'tenant_1', 'tenant_0') AS tenant_id, + concat('dest_', toString(number % 500)) AS destination_id, + multiIf( + number % 3 = 0, 'order.created', + number % 3 = 1, 'order.updated', + 'payment.received' + ) AS topic, + number % 3 != 2 AS eligible_for_retry, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) AS event_time, + '{}' AS metadata, + '{}' AS data, + concat('att_', toString(number), '_2') AS attempt_id, + if(number % 100 = 0, 'failed', 'success') AS status, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) + + toIntervalSecond(3) AS attempt_time, + multiIf( + number % 100 != 0 AND number % 2 = 0, '200', + number % 100 != 0, '201', + number % 2 = 0, '500', + '422' + ) AS code, + '' AS response_data, + number % 10 = 9 AS manual, + toUInt32(2) AS attempt_number +FROM numbers({rows:UInt64}) +WHERE number % 20 = 0; + +SELECT '[5/7] Inserting attempt 3 (1% of events)...' AS message; + +INSERT INTO attempts ( + event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data, + attempt_id, status, attempt_time, code, response_data, manual, attempt_number +) +SELECT + concat('evt_', toString(number)) AS event_id, + if(number % 10 = 0, 'tenant_1', 'tenant_0') AS tenant_id, + concat('dest_', toString(number % 500)) AS destination_id, + multiIf( + number % 3 = 0, 'order.created', + number % 3 = 1, 'order.updated', + 'payment.received' + ) AS topic, + number % 3 != 2 AS eligible_for_retry, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) AS event_time, + '{}' AS metadata, + '{}' AS data, + concat('att_', toString(number), '_3') AS attempt_id, + if(number % 200 = 0, 'failed', 'success') AS status, + toDateTime64('2000-01-01', 3) + + toIntervalMillisecond( + toUInt64(number * 2678400000 / {rows:UInt64}) + ) + + toIntervalSecond(4) AS attempt_time, + multiIf( + number % 200 != 0 AND number % 2 = 0, '200', + number % 200 != 0, '201', + number % 2 = 0, '500', + '422' + ) AS code, + '' AS response_data, + number % 10 = 9 AS manual, + toUInt32(3) AS attempt_number +FROM numbers({rows:UInt64}) +WHERE number % 100 = 0; + +-- ============================================================================ +-- 3. OPTIMIZE (force ReplacingMergeTree merge) +-- ============================================================================ + +SELECT '[6/7] Optimizing (forcing merge)...' AS message; + +OPTIMIZE TABLE events FINAL; +OPTIMIZE TABLE attempts FINAL; + +-- ============================================================================ +-- 4. Report +-- ============================================================================ + +SELECT '[7/7] Done. Row counts:' AS message; + +SELECT 'events' AS table_name, count() AS row_count FROM events +UNION ALL +SELECT 'attempts' AS table_name, count() AS row_count FROM attempts; + +SELECT attempt_number, status, count() AS cnt +FROM attempts GROUP BY attempt_number, status +ORDER BY attempt_number, status; diff --git a/cmd/bench/metrics/ch_test.go b/cmd/bench/metrics/ch_test.go new file mode 100644 index 000000000..6b54c61ed --- /dev/null +++ b/cmd/bench/metrics/ch_test.go @@ -0,0 +1,73 @@ +package metrics + +import ( + "context" + "os" + "testing" + + "github.com/hookdeck/outpost/internal/clickhouse" + "github.com/hookdeck/outpost/internal/logstore/chlogstore" + "github.com/hookdeck/outpost/internal/logstore/driver" + "github.com/hookdeck/outpost/internal/migrator" +) + +func newCHBench(tb testing.TB) driver.Metrics { + tb.Helper() + + chAddr := os.Getenv("BENCH_CH_ADDR") + if chAddr == "" { + tb.Skip("BENCH_CH_ADDR not set — skipping CH metrics benchmarks") + } + + chDB := os.Getenv("BENCH_CH_DB") + if chDB == "" { + chDB = "bench" + } + + ctx := context.Background() + + // Run migrations. + m, err := migrator.New(migrator.MigrationOpts{ + CH: migrator.MigrationOptsCH{ + Addr: chAddr, + Database: chDB, + Username: "default", + }, + }) + if err != nil { + tb.Fatalf("migrator: %v", err) + } + _, _, err = m.Up(ctx, -1) + if err != nil { + tb.Fatalf("migrate up: %v", err) + } + srcErr, dbErr := m.Close(ctx) + if srcErr != nil { + tb.Fatalf("migrator close src: %v", srcErr) + } + if dbErr != nil { + tb.Fatalf("migrator close db: %v", dbErr) + } + + conn, err := clickhouse.New(&clickhouse.ClickHouseConfig{ + Addr: chAddr, + Database: chDB, + Username: "default", + }) + if err != nil { + tb.Fatalf("clickhouse: %v", err) + } + tb.Cleanup(func() { conn.Close() }) + + return chlogstore.NewLogStore(conn, "") +} + +func BenchmarkCHEventMetrics(b *testing.B) { + store := newCHBench(b) + benchmarkEventMetrics(b, store) +} + +func BenchmarkCHAttemptMetrics(b *testing.B) { + store := newCHBench(b) + benchmarkAttemptMetrics(b, store) +} diff --git a/cmd/bench/metrics/pg/README.md b/cmd/bench/metrics/pg/README.md new file mode 100644 index 000000000..e35f42204 --- /dev/null +++ b/cmd/bench/metrics/pg/README.md @@ -0,0 +1,66 @@ +# PG Metrics Benchmark + +Benchmarks `QueryEventMetrics` / `QueryAttemptMetrics` against PostgreSQL (2 CPU, 8GB). + +## Prerequisites + +- Docker (Compose v2) +- Go 1.24+ +- `psql` + +## Quick Start + +```bash +cd outpost + +# 1. Start PG +docker compose -f cmd/bench/metrics/pg/docker-compose.yml up -d + +# 2. Run migrations +BENCH_PG_URL="postgres://outpost:outpost@localhost:5488/bench?sslmode=disable" \ + go test -run='^$' -bench=BenchmarkPG -benchtime=1x ./cmd/bench/metrics/ + +# 3. Seed (default 10M — adjust -v ROWS=N) +psql "postgres://outpost:outpost@localhost:5488/bench?sslmode=disable" \ + -v ROWS=10000000 -f cmd/bench/metrics/pg/seed.sql + +# 4a. Single iteration +BENCH_PG_URL="postgres://outpost:outpost@localhost:5488/bench?sslmode=disable" \ + go test -bench=BenchmarkPG -benchtime=1x -count=1 -timeout=30m ./cmd/bench/metrics/ + +# 4b. Sustained (10s x 3 runs) +BENCH_PG_URL="postgres://outpost:outpost@localhost:5488/bench?sslmode=disable" \ + go test -bench=BenchmarkPG -benchtime=10s -count=3 -timeout=30m ./cmd/bench/metrics/ + +# 5. Cleanup +docker compose -f cmd/bench/metrics/pg/docker-compose.yml down -v +``` + +## Re-seeding + +```bash +docker compose -f cmd/bench/metrics/pg/docker-compose.yml down -v +docker compose -f cmd/bench/metrics/pg/docker-compose.yml up -d +# Repeat steps 2-4 +``` + +## Data Distribution + +Deterministic via modulo arithmetic (shared with CH bench): + +- **2 tenants** — `tenant_0` (90%), `tenant_1` (10%) +- **500 destinations** — `dest_0` through `dest_499` +- **3 topics** — `order.created`, `order.updated`, `payment.received` +- **Time** — evenly spread across January 2000 +- **Attempts** — chained retries (1 event -> 1-4 attempts), 0.5% permanently failed +- 10M events -> ~12.6M attempts + +## Resource Tuning + +| Setting | Default | Purpose | +|---------|---------|---------| +| CPUs | 2 | Parallel query workers | +| Memory | 4GB | Container limit | +| shared_buffers | 1GB | PG buffer pool | +| work_mem | 256MB | Per-sort/hash memory | +| effective_cache_size | 3GB | Planner hint for OS cache | diff --git a/cmd/bench/metrics/pg/docker-compose.yml b/cmd/bench/metrics/pg/docker-compose.yml new file mode 100644 index 000000000..bd0db31e1 --- /dev/null +++ b/cmd/bench/metrics/pg/docker-compose.yml @@ -0,0 +1,27 @@ +services: + postgres: + image: postgres:16-alpine + environment: + POSTGRES_USER: outpost + POSTGRES_PASSWORD: outpost + POSTGRES_DB: bench + ports: + - "5488:5432" + deploy: + resources: + limits: + cpus: "2" + memory: 8g + command: + - postgres + - -c + - shared_buffers=2GB + - -c + - work_mem=512MB + - -c + - effective_cache_size=6GB + healthcheck: + test: ["CMD-SHELL", "pg_isready -U outpost -d bench"] + interval: 2s + timeout: 5s + retries: 10 diff --git a/cmd/bench/metrics/pg/seed.sql b/cmd/bench/metrics/pg/seed.sql new file mode 100644 index 000000000..7b77feeb2 --- /dev/null +++ b/cmd/bench/metrics/pg/seed.sql @@ -0,0 +1,256 @@ +-- seed.sql — Deterministic bulk seeding for PG metrics benchmarks. +-- +-- Usage: +-- psql "$POSTGRES_URL" -v ROWS=10000000 -f cmd/bench/metrics/pg/seed.sql +-- +-- Default :ROWS = 10000000 (10M). Override with -v ROWS=N. +-- +-- Distribution: +-- 2 tenants: tenant_0 gets 90%, tenant_1 gets 10% +-- Time: evenly spread across January 2000 (2000-01-01 to 2000-02-01) +-- No explicit partitions — data lands in the default partition. +-- +-- Attempt chain (1 event → 1-4 attempts): +-- attempt 0: all events. Failed if n%5=0 (20%) +-- attempt 1: failed attempt 0. Failed if n%20=0 (25% of retries) +-- attempt 2: failed attempt 1. Failed if n%100=0 (20% of retries) +-- attempt 3: failed attempt 2. Failed if n%200=0 (50% of retries) +-- +-- For 10M events → ~12.6M attempts. 0.5% events permanently failed. + +\set ON_ERROR_STOP on +\timing on + +-- Default if not supplied via -v +SELECT COALESCE(:'ROWS', '10000000') AS rows_count \gset + +\echo Seeding :rows_count events + chained attempts... + +-- ============================================================================ +-- 1. Bulk INSERT into events +-- ============================================================================ +-- +-- Tenants: n%10 == 0 → tenant_1 (10%), else tenant_0 (90%) +-- Destinations: dest_(n%500) [500 destinations] +-- Topics: n%3 → order.created / order.updated / payment.received +-- Time: Even spread across 2000-01-01 to 2000-02-01 +-- eligible_for_retry: n%3 != 2 + +\echo [1/7] Inserting events... + +INSERT INTO events (id, tenant_id, destination_id, time, topic, eligible_for_retry, data, metadata) +SELECT + 'evt_' || n AS id, + CASE WHEN n % 10 = 0 + THEN 'tenant_1' + ELSE 'tenant_0' + END AS tenant_id, + 'dest_' || (n % 500) AS destination_id, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + AS time, + CASE n % 3 + WHEN 0 THEN 'order.created' + WHEN 1 THEN 'order.updated' + ELSE 'payment.received' + END AS topic, + (n % 3 != 2) AS eligible_for_retry, + '{}' AS data, + '{}'::jsonb AS metadata +FROM generate_series(0, :'rows_count'::int - 1) AS n; + +-- ============================================================================ +-- 2. Bulk INSERT into attempts (chained retries) +-- ============================================================================ +-- +-- Shared columns reuse the same expressions as events. +-- Each attempt's time = event_time + (attempt_number * 1 second). +-- manual: only attempt_number >= 2 AND n%10=9 (10% of late retries). +-- Code: success→200/201, failed→500/422 (alternating on n%2). + +\echo [2/7] Inserting attempt 0 (all events)... + +INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +) +SELECT + 'att_' || n || '_0' AS id, + 'evt_' || n AS event_id, + CASE WHEN n % 10 = 0 THEN 'tenant_1' ELSE 'tenant_0' END + AS tenant_id, + 'dest_' || (n % 500) AS destination_id, + CASE n % 3 + WHEN 0 THEN 'order.created' + WHEN 1 THEN 'order.updated' + ELSE 'payment.received' + END AS topic, + CASE WHEN n % 5 = 0 THEN 'failed' ELSE 'success' END AS status, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + + interval '1 second' + AS time, + 0 AS attempt_number, + false AS manual, + CASE + WHEN n % 5 != 0 THEN CASE WHEN n%2=0 THEN '200' ELSE '201' END + ELSE CASE WHEN n%2=0 THEN '500' ELSE '422' END + END AS code, + NULL AS response_data, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + AS event_time, + (n % 3 != 2) AS eligible_for_retry, + '{}' AS event_data, + '{}'::jsonb AS event_metadata +FROM generate_series(0, :'rows_count'::int - 1) AS n; + +\echo [3/7] Inserting attempt 1 (20% of events)... + +INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +) +SELECT + 'att_' || n || '_1' AS id, + 'evt_' || n AS event_id, + CASE WHEN n % 10 = 0 THEN 'tenant_1' ELSE 'tenant_0' END + AS tenant_id, + 'dest_' || (n % 500) AS destination_id, + CASE n % 3 + WHEN 0 THEN 'order.created' + WHEN 1 THEN 'order.updated' + ELSE 'payment.received' + END AS topic, + CASE WHEN n % 20 = 0 THEN 'failed' ELSE 'success' END AS status, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + + interval '2 seconds' + AS time, + 1 AS attempt_number, + false AS manual, + CASE + WHEN n % 20 != 0 THEN CASE WHEN n%2=0 THEN '200' ELSE '201' END + ELSE CASE WHEN n%2=0 THEN '500' ELSE '422' END + END AS code, + NULL AS response_data, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + AS event_time, + (n % 3 != 2) AS eligible_for_retry, + '{}' AS event_data, + '{}'::jsonb AS event_metadata +FROM generate_series(0, :'rows_count'::int - 1) AS n +WHERE n % 5 = 0; + +\echo [4/7] Inserting attempt 2 (5% of events)... + +INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +) +SELECT + 'att_' || n || '_2' AS id, + 'evt_' || n AS event_id, + CASE WHEN n % 10 = 0 THEN 'tenant_1' ELSE 'tenant_0' END + AS tenant_id, + 'dest_' || (n % 500) AS destination_id, + CASE n % 3 + WHEN 0 THEN 'order.created' + WHEN 1 THEN 'order.updated' + ELSE 'payment.received' + END AS topic, + CASE WHEN n % 100 = 0 THEN 'failed' ELSE 'success' END AS status, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + + interval '3 seconds' + AS time, + 2 AS attempt_number, + (n % 10 = 9) AS manual, + CASE + WHEN n % 100 != 0 THEN CASE WHEN n%2=0 THEN '200' ELSE '201' END + ELSE CASE WHEN n%2=0 THEN '500' ELSE '422' END + END AS code, + NULL AS response_data, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + AS event_time, + (n % 3 != 2) AS eligible_for_retry, + '{}' AS event_data, + '{}'::jsonb AS event_metadata +FROM generate_series(0, :'rows_count'::int - 1) AS n +WHERE n % 20 = 0; + +\echo [5/7] Inserting attempt 3 (1% of events)... + +INSERT INTO attempts ( + id, event_id, tenant_id, destination_id, topic, status, time, + attempt_number, manual, code, response_data, + event_time, eligible_for_retry, event_data, event_metadata +) +SELECT + 'att_' || n || '_3' AS id, + 'evt_' || n AS event_id, + CASE WHEN n % 10 = 0 THEN 'tenant_1' ELSE 'tenant_0' END + AS tenant_id, + 'dest_' || (n % 500) AS destination_id, + CASE n % 3 + WHEN 0 THEN 'order.created' + WHEN 1 THEN 'order.updated' + ELSE 'payment.received' + END AS topic, + CASE WHEN n % 200 = 0 THEN 'failed' ELSE 'success' END AS status, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + + interval '4 seconds' + AS time, + 3 AS attempt_number, + (n % 10 = 9) AS manual, + CASE + WHEN n % 200 != 0 THEN CASE WHEN n%2=0 THEN '200' ELSE '201' END + ELSE CASE WHEN n%2=0 THEN '500' ELSE '422' END + END AS code, + NULL AS response_data, + '2000-01-01'::timestamptz + + (n::double precision / :'rows_count'::double precision) + * ('2000-02-01'::timestamptz - '2000-01-01'::timestamptz) + AS event_time, + (n % 3 != 2) AS eligible_for_retry, + '{}' AS event_data, + '{}'::jsonb AS event_metadata +FROM generate_series(0, :'rows_count'::int - 1) AS n +WHERE n % 100 = 0; + +-- ============================================================================ +-- 3. ANALYZE +-- ============================================================================ + +\echo [6/7] Analyzing... + +ANALYZE events; +ANALYZE attempts; + +-- ============================================================================ +-- 4. Report +-- ============================================================================ + +\echo [7/7] Done. Row counts: + +SELECT 'events' AS table_name, count(*) AS row_count FROM events +UNION ALL +SELECT 'attempts' AS table_name, count(*) AS row_count FROM attempts; + +SELECT attempt_number, status, count(*) AS cnt +FROM attempts GROUP BY attempt_number, status +ORDER BY attempt_number, status; diff --git a/cmd/bench/metrics/pg_test.go b/cmd/bench/metrics/pg_test.go new file mode 100644 index 000000000..deceef4c5 --- /dev/null +++ b/cmd/bench/metrics/pg_test.go @@ -0,0 +1,60 @@ +package metrics + +import ( + "context" + "os" + "testing" + + "github.com/hookdeck/outpost/internal/logstore/driver" + "github.com/hookdeck/outpost/internal/logstore/pglogstore" + "github.com/hookdeck/outpost/internal/migrator" + "github.com/jackc/pgx/v5/pgxpool" +) + +func newPGBench(tb testing.TB) driver.Metrics { + tb.Helper() + + pgURL := os.Getenv("BENCH_PG_URL") + if pgURL == "" { + tb.Skip("BENCH_PG_URL not set — skipping PG metrics benchmarks") + } + + ctx := context.Background() + + // Run migrations. + m, err := migrator.New(migrator.MigrationOpts{ + PG: migrator.MigrationOptsPG{URL: pgURL}, + }) + if err != nil { + tb.Fatalf("migrator: %v", err) + } + _, _, err = m.Up(ctx, -1) + if err != nil { + tb.Fatalf("migrate up: %v", err) + } + srcErr, dbErr := m.Close(ctx) + if srcErr != nil { + tb.Fatalf("migrator close src: %v", srcErr) + } + if dbErr != nil { + tb.Fatalf("migrator close db: %v", dbErr) + } + + db, err := pgxpool.New(ctx, pgURL) + if err != nil { + tb.Fatalf("pgxpool: %v", err) + } + tb.Cleanup(db.Close) + + return pglogstore.NewLogStore(db) +} + +func BenchmarkPGEventMetrics(b *testing.B) { + store := newPGBench(b) + benchmarkEventMetrics(b, store) +} + +func BenchmarkPGAttemptMetrics(b *testing.B) { + store := newPGBench(b) + benchmarkAttemptMetrics(b, store) +}