diff --git a/.github/workflows/render-template.yml b/.github/workflows/render-template.yml new file mode 100644 index 00000000..ba9513fe --- /dev/null +++ b/.github/workflows/render-template.yml @@ -0,0 +1,35 @@ +name: render-template + +# The Deploy to Render template under render/. Runs when it, the Blueprint or +# this file changes. +on: + push: + branches: [main] + paths: ["render/**", "render.yaml", ".github/workflows/render-template.yml"] + pull_request: + paths: ["render/**", "render.yaml", ".github/workflows/render-template.yml"] + +jobs: + validate: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/setup-uv@v5 + - name: Configs, the generated migration, and render.yaml against Render's schema + run: make -C render validate + - name: A render.yaml that Render would refuse is refused + run: | + sed 's/plan: 0.5c-512mb/plan: not-a-plan/' render.yaml > "$RUNNER_TEMP/bad.yaml" + status=0 + render/bin/validate-render.py "$RUNNER_TEMP/bad.yaml" || status=$? + if [ "$status" != 1 ]; then + echo "expected status 1 for an invalid plan, got $status" >&2 + exit 1 + fi + + e2e: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Post metrics and read them back at every grain + run: make -C render test diff --git a/.gitignore b/.gitignore index 1562722e..a561d29c 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,8 @@ __pycache__/ .venv/ dev/volume/ bin/ +# The Deploy to Render template's scripts are source, not build output. +!/render/bin/ dist/ .claude/ diff --git a/CHANGELOG.md b/CHANGELOG.md index e7e9a846..dc26c3b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,9 @@ ### Added +- A Deploy to Render button. `render.yaml` and `render/` deploy a Postgres, a + webhook pipeline that aggregates a generic metric by minute, and a + `sqlflow serve` API over a six-grain rollup ladder. See `render/README.md`. - `GET /healthz` on the webhook source's address. It answers 200 while the source admits deliveries and 503 once it is closing, needs no signature, and is not counted in `webhook_requests_total`. A platform that routes one diff --git a/README.md b/README.md index eec421dd..bc8624d5 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,12 @@ SQLFlow is a stream processing engine that lets you define pipelines with just S - [Coverage and invariant matrix](docs/coverage/matrix.md): what is tested, and separately, what is *proven* — regenerated from the suites on every push. +[![Deploy to Render](https://render.com/images/deploy-to-render-button.svg)](https://render.com/deploy?repo=https://github.com/turbolytics/sql-flow) + +One click runs sqlflow on Render: a webhook that accepts metrics, a one-minute +aggregation into Postgres, and an HTTP API that reads them back. It asks for +one secret and is a paid deploy. See [`render/`](render/README.md). + # Quick Start (Getting Started in 5 Minutes) 1. Build a binary (see [Installation](#installation) for prebuilt binaries and Docker): diff --git a/docs/coverage/features.yml b/docs/coverage/features.yml index e7e2e168..05960e4a 100644 --- a/docs/coverage/features.yml +++ b/docs/coverage/features.yml @@ -202,6 +202,12 @@ features: description: Generates rollup tables, the triggers that keep them current, and the serve datasets that read them from one declaration, and checks the generated files have not drifted. requires: [unit, integration] + # The template is YAML, SQL and shell, so what it promises under concurrency + # is tested from Go against its migrations, in internal/rendertemplate. + - id: template.render + description: The Deploy to Render template's schema keeps the minute table and every rollup exact when several pipeline processes write to one database. + requires: [integration] + # Release only. The defect this catches -- build args that never reached the # -ldflags symbol path, so the binary reports "dev" -- exists only in a built # image, and no unit test can see it. diff --git a/docs/coverage/invariants.yml b/docs/coverage/invariants.yml index 08944ea7..fc178699 100644 --- a/docs/coverage/invariants.yml +++ b/docs/coverage/invariants.yml @@ -538,6 +538,36 @@ invariants: requires: [] enforced: true + # --- Resilience: more than one process writing one destination ------------- + # The engine gives each pipeline process its own window, and a windowed + # pipeline publishes a closed minute by replacing the row for its key. That + # is right for one process and wrong for two: instances behind a load + # balancer each hold part of a minute, and keyed on the series alone the + # second to publish replaces the first. Two real instances, sent 7 and 5 of + # one minute, stored 7 (#331). Two instances also run during every deploy, + # the old beside the new. + # + # Nothing in the engine prevents it, so the claim is about what a destination + # ends up holding, and it is proven where a destination's schema makes it + # true: the Deploy to Render template keys its write on the process and has + # Postgres merge the processes. A pipeline configuration with no such schema + # has no evidence here, and that empty cell is accurate. + - id: pipeline.writers.merge_exactly + family: resilience + class: safety + applies_to: pipeline + claim: > + When several pipeline processes publish the same window of the same key + to one destination, the destination holds the merge of what every + process last published: no process's rows replace another's, and a + process that republishes replaces only its own. It holds at every grain + derived from that window, under concurrent publishes, without deadlock. + Proven in internal/rendertemplate against the Deploy to Render + template's migrations and the Postgres sink, not by the conformance + harness: the property belongs to the destination's schema. + verified_by: harness + requires: [] + # --- Cache: serve's response cache ---------------------------------------- # Serve is never told that its backend changed, so the cache invalidates # nothing and bounds everything. These say what the bounds are. Each is diff --git a/docs/coverage/matrix.md b/docs/coverage/matrix.md index 9d61a548..5aef8f7d 100644 --- a/docs/coverage/matrix.md +++ b/docs/coverage/matrix.md @@ -60,11 +60,12 @@ added, and this page changes only when a status does. | `cli.dev_invoke` | Runs a pipeline against a fixture file, without a source. | ✅ | — | ✅ | | `cli.serve` | Serves a config's named SQL datasets over HTTP, with client ids, typed params, grains and limits. | ✅ | ✅ | ✅ | | `cli.rollup` | Generates rollup tables, the triggers that keep them current, and the serve datasets that read them from one declaration, and checks the generated files have not drifted. | ✅ | ✅ | — | +| `template.render` | The Deploy to Render template's schema keeps the minute table and every rollup exact when several pipeline processes write to one database. | ⚠️ skipped | ✅ | — | | `cli.version` | The shipped binary reports the version it was built from. | — | — | ✅ | | `tooling.conformance` | The harness proves the declared invariants for any integration. | ✅ | — | — | | `tooling.coverage` | Tests attribute to features and invariants, and the registries match the code. | ✅ | — | — | -**41 features declared. 41 have at least one passing test attributed at every level they require, so 0 gap(s).** +**42 features declared. 41 have at least one passing test attributed at every level they require, so 0 gap(s).** That sentence counts attribution, not proof. A feature is green here when a test named for it ran and passed; it says nothing about whether the @@ -72,7 +73,7 @@ integration behind it keeps a batch it could not deliver, or commits offsets only after a flush. Those are invariants, they are counted separately below, and the two numbers are not interchangeable. -**50 invariants declared: 42 safety and 8 liveness. Of 178 (invariant, integration) cells: 94 proven, 50 missing, 0 skipped, 0 failing, 34 exempt. 0 gap(s).** +**51 invariants declared: 43 safety and 8 liveness. Of 180 (invariant, integration) cells: 95 proven, 51 missing, 0 skipped, 0 failing, 34 exempt. 0 gap(s).** Safety says nothing bad happens. Liveness says something good eventually does, and the two are not interchangeable: a sink that @@ -116,6 +117,17 @@ until an invariant's `requires` is filled in, and none is yet. | `sink.probe.fails_start` | A Prober whose destination is unreachable fails the start once, without retrying. | ✅ i | — exempt | — exempt | ✅ i | — exempt | ✅ i | — exempt | | `sink.flush.idempotent_on_key` | Delivering the same batch twice leaves the destination holding it once. The engine promises at-least-once; a sink that identifies rows by key is what turns that into exactly-once for the reader. A sink that declares no key is exempt, with a proof that it declares none. | — exempt | — exempt | — exempt | — exempt | — exempt | ✅ i | — exempt | +These resilience invariants are properties of the consume loop +rather than of anything a config file names. The columns are +its configurations, and `internal/conformance` runs each one +through every path that reaches a batch: the batch filling, the +flush interval elapsing, the source closing, and a cancel that +drains. An invariant holds only if it holds on all four. + +| Invariant | Claim | `pipeline.stateful` | `pipeline.stateless` | +| --- | --- | --- | --- | +| `pipeline.writers.merge_exactly` | When several pipeline processes publish the same window of the same key to one destination, the destination holds the merge of what every process last published: no process's rows replace another's, and a process that republishes replaces only its own. It holds at every grain derived from that window, under concurrent publishes, without deadlock. Proven in internal/rendertemplate against the Deploy to Render template's migrations and the Postgres sink, not by the conformance harness: the property belongs to the destination's schema. | ❌ missing | ✅ i | + ## Safety invariants: checkpoint | Invariant | Claim | `source.kafka` | `source.webhook` | `source.websocket` | `manager.watermark` | diff --git a/docs/coverage/status/features.yml b/docs/coverage/status/features.yml index 81a7dcc0..0f71b834 100644 --- a/docs/coverage/status/features.yml +++ b/docs/coverage/status/features.yml @@ -37,6 +37,7 @@ source.websocket: {unit: covered, integration: not_required, release: covered} state.corruption: {unit: covered, integration: not_required, release: covered} state.durability: {unit: covered, integration: not_required, release: covered} state.offsets: {unit: covered, integration: not_required, release: covered} +template.render: {unit: skipped, integration: covered, release: not_required} tooling.conformance: {unit: covered, integration: not_required, release: not_required} tooling.coverage: {unit: covered, integration: not_required, release: not_required} validate.schema: {unit: covered, integration: not_required, release: not_required} diff --git a/docs/coverage/status/pipeline.stateful.yml b/docs/coverage/status/pipeline.stateful.yml index a049ebf0..f122242a 100644 --- a/docs/coverage/status/pipeline.stateful.yml +++ b/docs/coverage/status/pipeline.stateful.yml @@ -12,3 +12,4 @@ pipeline.flush.eventually: {unit: covered, integration: missing, release: missin pipeline.progress.no_silent_stall: {unit: missing, integration: missing, release: missing} pipeline.shutdown.commits_only_delivered: {unit: covered, integration: missing, release: missing} pipeline.state.with_offsets: {unit: covered, integration: missing, release: missing} +pipeline.writers.merge_exactly: {unit: missing, integration: missing, release: missing} diff --git a/docs/coverage/status/pipeline.stateless.yml b/docs/coverage/status/pipeline.stateless.yml index 5f8fb18f..08a11367 100644 --- a/docs/coverage/status/pipeline.stateless.yml +++ b/docs/coverage/status/pipeline.stateless.yml @@ -12,3 +12,4 @@ pipeline.flush.eventually: {unit: covered, integration: missing, release: missin pipeline.progress.no_silent_stall: {unit: missing, integration: missing, release: missing} pipeline.shutdown.commits_only_delivered: {unit: covered, integration: missing, release: missing} pipeline.state.with_offsets: {unit: exempt, integration: exempt, release: exempt} +pipeline.writers.merge_exactly: {unit: skipped, integration: covered, release: missing} diff --git a/internal/rendertemplate/writers_integration_test.go b/internal/rendertemplate/writers_integration_test.go new file mode 100644 index 00000000..fa0b1c2d --- /dev/null +++ b/internal/rendertemplate/writers_integration_test.go @@ -0,0 +1,516 @@ +// Package rendertemplate holds the Deploy to Render template's SQL, under +// render/migrations, to what it promises when several pipeline processes +// write to one database. The template is YAML, SQL and shell; this package is +// tests only, because the promise is only as good as the concurrency it was +// tried under, and that needs goroutines and open transactions. +package rendertemplate + +import ( + "context" + "fmt" + "io" + "math/rand" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/apache/arrow-go/v18/arrow/array" + "github.com/apache/arrow-go/v18/arrow/memory" + "github.com/jackc/pgx/v5" + tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres" + "github.com/turbolytics/sql-flow/internal/config" + "github.com/turbolytics/sql-flow/internal/coverage" + "github.com/turbolytics/sql-flow/internal/sinks" + "github.com/zeebo/assert" +) + +const ( + migrationsDir = "../../render/migrations" + // The template's pipeline keeps its window in memory and has no state + // file, which is the stateless configuration of the consume loop. + templatePipeline = "pipeline.stateless" + mergeInvariant = "pipeline.writers.merge_exactly" +) + +var writerKey = []string{"bucket", "name", "type", "dimensions_key", "writer"} + +// migrations is every file the template's runner applies, in its order. +func migrations(t *testing.T) []string { + t.Helper() + paths, err := filepath.Glob(filepath.Join(migrationsDir, "*.sql")) + assert.NoError(t, err) + sort.Strings(paths) + if len(paths) == 0 { + t.Fatalf("no migrations under %s", migrationsDir) + } + out := make([]string, len(paths)) + for i, p := range paths { + b, err := os.ReadFile(p) + assert.NoError(t, err) + out[i] = string(b) + } + return out +} + +type server struct { + dsn string + conn *pgx.Conn +} + +// start runs Postgres 18, the template's version, and applies the migrations +// after edit has had its way with them. Each case gets its own container: the +// migrations create functions and triggers a second apply would collide with. +func start(t *testing.T, edit func(string) string) *server { + t.Helper() + ctx := context.Background() + pg, err := tcpostgres.Run(ctx, "postgres:18", + tcpostgres.WithDatabase("metrics"), + tcpostgres.WithUsername("metrics"), + tcpostgres.WithPassword("metrics"), + tcpostgres.BasicWaitStrategies(), + ) + if err != nil { + t.Fatalf("start postgres: %v", err) + } + t.Cleanup(func() { + // A deadlock reaches the test as one line. The server's log says which + // locks the two processes held and wanted, which is the diagnosis. + if t.Failed() { + if logs, err := pg.Logs(context.Background()); err == nil { + raw, _ := io.ReadAll(logs) + for _, line := range strings.Split(string(raw), "\n") { + if strings.Contains(line, "deadlock") || strings.Contains(line, "waits for") || + strings.Contains(line, "blocked by") || strings.Contains(line, "CONTEXT") { + t.Log(line) + } + } + } + } + _ = pg.Terminate(context.Background()) + }) + dsn, err := pg.ConnectionString(ctx, "sslmode=disable") + assert.NoError(t, err) + + srv := &server{dsn: dsn, conn: connect(t, dsn)} + for _, script := range migrations(t) { + if edit != nil { + script = edit(script) + } + // One transaction per file, as bin/migrate.sh applies them. + exec(t, srv.conn, "BEGIN") + exec(t, srv.conn, script) + exec(t, srv.conn, "COMMIT") + } + return srv +} + +func connect(t *testing.T, dsn string) *pgx.Conn { + t.Helper() + conn, err := pgx.Connect(context.Background(), dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(func() { _ = conn.Close(context.Background()) }) + exec(t, conn, "SET TIME ZONE 'UTC'") + return conn +} + +func exec(t *testing.T, conn *pgx.Conn, sql string, args ...any) { + t.Helper() + _, err := conn.Exec(context.Background(), sql, args...) + assert.NoError(t, err) +} + +var mergeLock = regexp.MustCompile(` PERFORM pg_advisory_xact_lock\(hashtextextended\('metrics_1m_merge', 0\)\);\n`) + +// withoutTheLock removes the merge trigger's advisory lock and nothing else. +func withoutTheLock(script string) string { + if !strings.Contains(script, "metrics_1m_merge") { + return script + } + out := mergeLock.ReplaceAllString(script, "") + if out == script { + panic("no lock in the merge trigger; migrations/0004_writers.sql changed") + } + return out +} + +const publish = ` +INSERT INTO metrics_1m_writers (bucket, name, type, dimensions_key, writer, + value_sum, value_count, value_min, value_max, value_last, last_at) +VALUES ('2026-09-15 10:01:00+00', 'checkout', 'count', '{}', $1, $2::double precision, 1, + $2::double precision, $2::double precision, $2::double precision, $3::timestamptz) +ON CONFLICT (bucket, name, type, dimensions_key, writer) DO UPDATE SET + value_sum = excluded.value_sum, value_count = excluded.value_count, value_min = excluded.value_min, + value_max = excluded.value_max, value_last = excluded.value_last, last_at = excluded.last_at` + +// Two pipeline processes hold parts of one minute of one series: two +// instances behind a load balancer, or the old and the new instance during a +// deploy. The first has published and not yet committed when the second +// publishes. What the minute holds afterwards must be both. +func TestIntegrationTemplateRender_TwoWritersOfOneMinuteAreSummed(t *testing.T) { + coverage.Covers(t, "template.render") + if testing.Short() { + t.Skip("integration: starts a Postgres container") + } + + for _, tt := range []struct { + name string + secondWriter string + edit func(string) string + want string + proves bool + }{ + {"each process has its own writer id", "b", nil, "12|2|5|7|5", true}, + // The test must be able to fail, and these are the two ways the + // template could. Without the lock the second writer re-merges from a + // snapshot that lacks the first's uncommitted row, waits on the + // merged row, and overwrites it with its own half. + {"without the lock", "b", withoutTheLock, "5|1|5|5|5", false}, + // The defect the writer column exists to prevent: processes that + // share an id share a key, and the second replaces the first. It is + // why bin/entrypoint.sh makes the id and never reads one. + {"both processes share a writer id", "a", nil, "5|1|5|5|5", false}, + } { + t.Run(tt.name, func(t *testing.T) { + if tt.proves { + coverage.Invariant(t, mergeInvariant, templatePipeline) + } + srv := start(t, tt.edit) + first, second := connect(t, srv.dsn), connect(t, srv.dsn) + + exec(t, first, "BEGIN") + exec(t, first, publish, "a", 7.0, "2026-09-15 10:01:10+00") + + done := make(chan error, 1) + go func() { + _, err := second.Exec(context.Background(), publish, tt.secondWriter, 5.0, "2026-09-15 10:01:40+00") + done <- err + }() + // Long enough for the second writer to reach the first's lock or row. + time.Sleep(500 * time.Millisecond) + exec(t, first, "COMMIT") + assert.NoError(t, <-done) + + for _, table := range []string{"metrics_1m", "metrics_5m", "metrics_1d"} { + var got string + assert.NoError(t, srv.conn.QueryRow(context.Background(), + "SELECT value_sum||'|'||value_count||'|'||value_min||'|'||value_max||'|'||value_last FROM "+table).Scan(&got)) + if got != tt.want { + t.Fatalf("%s holds %s, want %s (sum|count|min|max|last)", table, got, tt.want) + } + } + }) + } +} + +// The lock waits, then reads a new snapshot. Under REPEATABLE READ the merge +// would read the old one and lose the writer it waited for, so it refuses. +func TestIntegrationTemplateRender_MergeRefusesAnotherIsolationLevel(t *testing.T) { + coverage.Covers(t, "template.render") + if testing.Short() { + t.Skip("integration: starts a Postgres container") + } + srv := start(t, nil) + + exec(t, srv.conn, "BEGIN ISOLATION LEVEL REPEATABLE READ") + _, err := srv.conn.Exec(context.Background(), publish, "a", 7.0, "2026-09-15 10:01:10+00") + assert.Error(t, err) + assert.That(t, strings.Contains(err.Error(), "READ COMMITTED")) + exec(t, srv.conn, "ROLLBACK") +} + +type key struct { + minute int + series int +} + +type row struct { + sum, min, max, last float64 + count int64 + lastAt time.Time +} + +var ( + firstMinute = time.Date(2026, 9, 15, 10, 3, 0, 0, time.UTC) + seriesKeys = []string{`{}`, `{"plan":"free"}`, `{"plan":"pro"}`} +) + +// Twelve minutes from 10:03, so they cross 5m and 15m boundaries. +const minutes = 12 + +func bucketOf(k key) time.Time { return firstMinute.Add(time.Duration(k.minute) * time.Minute) } + +// flush publishes rows through the Postgres sink, the pipeline's own write +// path: a COPY into a temp table, then one INSERT ... ON CONFLICT. +func flush(s sinksWriter, writer string, rows map[key]row) error { + schema := arrow.NewSchema([]arrow.Field{ + {Name: "bucket", Type: &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: "UTC"}}, + {Name: "name", Type: arrow.BinaryTypes.String}, + {Name: "type", Type: arrow.BinaryTypes.String}, + {Name: "dimensions_key", Type: arrow.BinaryTypes.String}, + {Name: "writer", Type: arrow.BinaryTypes.String}, + {Name: "value_sum", Type: arrow.PrimitiveTypes.Float64}, + {Name: "value_count", Type: arrow.PrimitiveTypes.Int64}, + {Name: "value_min", Type: arrow.PrimitiveTypes.Float64}, + {Name: "value_max", Type: arrow.PrimitiveTypes.Float64}, + {Name: "value_last", Type: arrow.PrimitiveTypes.Float64}, + {Name: "last_at", Type: &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: "UTC"}}, + }, nil) + b := array.NewRecordBuilder(memory.NewGoAllocator(), schema) + defer b.Release() + for k, r := range rows { + b.Field(0).(*array.TimestampBuilder).Append(arrow.Timestamp(bucketOf(k).UnixMicro())) + b.Field(1).(*array.StringBuilder).Append("load") + b.Field(2).(*array.StringBuilder).Append("gauge") + b.Field(3).(*array.StringBuilder).Append(seriesKeys[k.series]) + b.Field(4).(*array.StringBuilder).Append(writer) + b.Field(5).(*array.Float64Builder).Append(r.sum) + b.Field(6).(*array.Int64Builder).Append(r.count) + b.Field(7).(*array.Float64Builder).Append(r.min) + b.Field(8).(*array.Float64Builder).Append(r.max) + b.Field(9).(*array.Float64Builder).Append(r.last) + b.Field(10).(*array.TimestampBuilder).Append(arrow.Timestamp(r.lastAt.UnixMicro())) + } + rec := b.NewRecord() + defer rec.Release() + tbl := array.NewTableFromRecords(schema, []arrow.Record{rec}) + defer tbl.Release() + if err := s.WriteTable(context.Background(), tbl); err != nil { + return err + } + return s.Flush(context.Background()) +} + +type sinksWriter interface { + WriteTable(context.Context, arrow.Table) error + Flush(context.Context) error +} + +// merge folds one writer's row into what a minute holds. +func merge(into *row, r row, first bool) { + if first { + *into = r + return + } + into.sum += r.sum + into.count += r.count + into.min = min(into.min, r.min) + into.max = max(into.max, r.max) + if r.lastAt.After(into.lastAt) { + into.last, into.lastAt = r.last, r.lastAt + } +} + +// Four processes publish and republish overlapping minutes of three series at +// once, through the sink, for forty rounds each. Every value is a multiple of +// 0.25, so sums are exact and equality is exact. +// +// What each process last published for a key is what it holds the minute to +// be. The merge of those, worked out here in Go from what the writers +// remember and never from the database, is what metrics_1m and every grain +// above it must hold: nothing lost to another writer, nothing counted twice +// by a republish. +func TestIntegrationTemplateRender_ConcurrentWritersLoseNothingAtAnyGrain(t *testing.T) { + coverage.Covers(t, "template.render") + coverage.Invariant(t, mergeInvariant, templatePipeline) + if testing.Short() { + t.Skip("integration: starts a Postgres container") + } + srv := start(t, nil) + + const writers, rounds = 4, 40 + published := make([]map[key]row, writers) + errs := make(chan error, writers) + var wg sync.WaitGroup + for w := range writers { + published[w] = map[key]row{} + wg.Add(1) + go func() { + defer wg.Done() + s, err := sinks.NewPostgresSink(config.PostgresSink{ + DSN: srv.dsn, Table: "metrics_1m_writers", Mode: sinks.PostgresModeUpsert, Key: writerKey, + }) + if err != nil { + errs <- err + return + } + defer s.Close() + + rng := rand.New(rand.NewSource(int64(w) + 1)) + for round := range rounds { + batch := map[key]row{} + for range 1 + rng.Intn(4) { + k := key{rng.Intn(minutes), rng.Intn(len(seriesKeys))} + lo, hi := float64(1+rng.Intn(4))*0.25, float64(5+rng.Intn(4))*0.25 + batch[k] = row{ + sum: float64(1+rng.Intn(40)) * 0.25, count: int64(1 + rng.Intn(9)), + min: lo, max: hi, last: float64(rng.Intn(9)) * 0.25, + // Unique across writers and rounds, so which reading is the + // latest is never a tie. + lastAt: bucketOf(k).Add(time.Duration(round*writers+w+1) * time.Millisecond), + } + } + if err := flush(s, fmt.Sprintf("w%d", w), batch); err != nil { + errs <- fmt.Errorf("writer %d round %d: %w", w, round, err) + return + } + for k, r := range batch { + published[w][k] = r + } + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatal(err) + } + + wantMinute := map[key]*row{} + for _, rows := range published { + for k, r := range rows { + if _, ok := wantMinute[k]; !ok { + wantMinute[k] = &row{} + merge(wantMinute[k], r, true) + continue + } + merge(wantMinute[k], r, false) + } + } + if len(wantMinute) < minutes*len(seriesKeys)/2 { + t.Fatalf("only %d keys were written; the writers did not overlap enough to prove anything", len(wantMinute)) + } + overlapped := 0 + for k := range wantMinute { + n := 0 + for _, rows := range published { + if _, ok := rows[k]; ok { + n++ + } + } + if n > 1 { + overlapped++ + } + } + if overlapped == 0 { + t.Fatal("no key had two writers; nothing was merged") + } + t.Logf("%d keys, %d of them published by more than one writer", len(wantMinute), overlapped) + + type got struct { + sum, min, max float64 + count int64 + last *float64 + } + read := func(sql string) map[string]got { + out := map[string]got{} + rows, err := srv.conn.Query(context.Background(), sql) + assert.NoError(t, err) + defer rows.Close() + for rows.Next() { + var bucket time.Time + var series string + var g got + assert.NoError(t, rows.Scan(&bucket, &series, &g.sum, &g.count, &g.min, &g.max, &g.last)) + out[bucket.UTC().Format(time.RFC3339)+" "+series] = g + } + assert.NoError(t, rows.Err()) + return out + } + + for _, grain := range []struct { + table string + width time.Duration + }{ + {"metrics_1m", time.Minute}, {"metrics_5m", 5 * time.Minute}, {"metrics_15m", 15 * time.Minute}, + {"metrics_1h", time.Hour}, {"metrics_6h", 6 * time.Hour}, {"metrics_1d", 24 * time.Hour}, + } { + // Every width divides a day, so truncating to it lands where Postgres's + // date_bin from a midnight origin does. + want := map[string]*row{} + latest := map[string]time.Time{} + wantTotal := map[string]*row{} + for k, r := range wantMinute { + b := bucketOf(k).Truncate(grain.width).Format(time.RFC3339) + for id, m := range map[string]map[string]*row{b + " " + seriesKeys[k.series]: want, b + " *": wantTotal} { + if _, ok := m[id]; !ok { + m[id] = &row{} + merge(m[id], *r, true) + } else { + merge(m[id], *r, false) + } + } + if id := b + " " + seriesKeys[k.series]; bucketOf(k).After(latest[id]) { + latest[id] = bucketOf(k) + } + } + // A grain's last is its latest minute's last, not the latest last_at + // within it: the rollups order by bucket. Set once every minute is in. + for k, r := range wantMinute { + id := bucketOf(k).Truncate(grain.width).Format(time.RFC3339) + " " + seriesKeys[k.series] + if bucketOf(k).Equal(latest[id]) { + want[id].last = r.last + } + } + + have := read("SELECT bucket, dimensions_key, value_sum, value_count, value_min, value_max, value_last FROM " + grain.table) + if len(have) != len(want) { + t.Fatalf("%s holds %d rows, want %d", grain.table, len(have), len(want)) + } + for id, w := range want { + h, ok := have[id] + if !ok { + t.Fatalf("%s has no row for %s", grain.table, id) + } + if h.sum != w.sum || h.count != w.count || h.min != w.min || h.max != w.max || h.last == nil || *h.last != w.last { + t.Fatalf("%s %s holds sum=%v count=%d min=%v max=%v last=%v, want sum=%v count=%d min=%v max=%v last=%v", + grain.table, id, h.sum, h.count, h.min, h.max, h.last, w.sum, w.count, w.min, w.max, w.last) + } + } + + if grain.table == "metrics_1m" { + continue + } + total := strings.Replace(grain.table, "metrics_", "metrics_total_", 1) + haveTotal := read("SELECT bucket, '*', value_sum, value_count, value_min, value_max, NULL::double precision FROM " + total) + if len(haveTotal) != len(wantTotal) { + t.Fatalf("%s holds %d rows, want %d", total, len(haveTotal), len(wantTotal)) + } + for id, w := range wantTotal { + h := haveTotal[id] + if h.sum != w.sum || h.count != w.count || h.min != w.min || h.max != w.max { + t.Fatalf("%s %s holds sum=%v count=%d min=%v max=%v, want sum=%v count=%d min=%v max=%v", + total, id, h.sum, h.count, h.min, h.max, w.sum, w.count, w.min, w.max) + } + } + } + + // The writers' rows are scaffolding once their minutes are merged. The + // template's README says deleting them changes no other table, so that a + // deployer can trim the one table that grows per process. + before := read("SELECT bucket, dimensions_key, value_sum, value_count, value_min, value_max, value_last FROM metrics_1d") + exec(t, srv.conn, "DELETE FROM metrics_1m_writers") + var minuteRows int64 + assert.NoError(t, srv.conn.QueryRow(context.Background(), "SELECT count(*) FROM metrics_1m").Scan(&minuteRows)) + assert.Equal(t, int64(len(wantMinute)), minuteRows) + after := read("SELECT bucket, dimensions_key, value_sum, value_count, value_min, value_max, value_last FROM metrics_1d") + assert.Equal(t, len(before), len(after)) + for id, b := range before { + a := after[id] + if a.sum != b.sum || a.count != b.count || a.min != b.min || a.max != b.max || *a.last != *b.last { + t.Fatalf("metrics_1d %s changed when the writers' rows were deleted", id) + } + } + + // A series is recorded once however many writers published it. + var series int64 + assert.NoError(t, srv.conn.QueryRow(context.Background(), "SELECT count(*) FROM series").Scan(&series)) + assert.Equal(t, int64(len(seriesKeys)), series) +} diff --git a/render.yaml b/render.yaml new file mode 100644 index 00000000..035043dc --- /dev/null +++ b/render.yaml @@ -0,0 +1,110 @@ +# Render Blueprint for the Deploy to Render button in the README: a Postgres, +# a web service that accepts metrics over a signed webhook and aggregates them +# by minute, and a web service that reads them back over HTTP. +# +# This file is at the repository root because the button reads it from there. +# Everything it deploys is under render/. +# +# The deploy asks for three values. The first two are yours to choose, so you +# have them when the deploy finishes. `openssl rand -hex 32` and `openssl rand -hex 16` +# make good ones. +# SQLFLOW_WEBHOOK_HMAC_SECRET signs what you send. A blank one fails the +# ingest service: the pipeline will not start +# unsigned unless SQLFLOW_WEBHOOK_AUTH is none. +# SQLFLOW_SERVE_CLIENT_ID reads what you sent, as ?client_id=. A blank +# one fails the API service. Letters, digits, +# '.', '_', '~' and '-' only. +# SQLFLOW_TELEMETRY leave it blank, or type off. Blank sends the +# sqlflow maintainers two events, ever: that +# this was deployed, and that it received its +# first metric. render/README.md, "What this +# sends", prints both in full. +databases: + - name: sqlflow-metrics-db + plan: 0.1c-256mb + # Same region as both services: connectionString is the private-network + # URL, which only resolves within one region. + region: virginia + postgresMajorVersion: "18" + diskSizeGB: 5 + # Off, so storage is watched rather than grown silently. Nothing deletes + # from the minute table yet. + storageAutoscalingEnabled: false + +services: + # The pipeline. A web service, not a worker, because only a web service + # gets a public URL, and senders need one. + - type: web + name: sqlflow-metrics-ingest + runtime: docker + dockerfilePath: ./render/Dockerfile + dockerContext: ./render + plan: 0.5c-512mb + region: virginia + # The webhook source answers /healthz on the port it takes deliveries on: + # 200 while it admits them, 503 once it is closing. Without this Render + # calls a service live when its port opens, cannot tell an instance that + # is shutting down from one that is serving, and cannot hold a deploy + # until the new instance is ready. It needs no signature. + healthCheckPath: /healthz + # Off: a copy of this file in someone else's workspace would otherwise + # redeploy whenever this repository's main moves. Quoted: unquoted, YAML + # reads off as false, and Render's schema wants a trigger name. + autoDeployTrigger: "off" + envVars: + - key: SQLFLOW_POSTGRES_URI + fromDatabase: + name: sqlflow-metrics-db + property: connectionString + # Prompted for during the deploy. + - key: SQLFLOW_WEBHOOK_HMAC_SECRET + sync: false + # Prompted for during the deploy, so that it is seen before anything is + # sent rather than found afterwards. Blank is on. off sends nothing. + # sync: false also means a later sync never touches what was typed. + - key: SQLFLOW_TELEMETRY + sync: false + # SQLFLOW_WEBHOOK_AUTH is not here, on purpose. Unset, the pipeline + # requires a signature. To accept unsigned requests, add it to this + # service in the dashboard with the value none. Render keeps a variable + # this file does not name and rewrites one it does, so a value here + # would put a pipeline set to none back to hmac on every sync, and it + # would start refusing its senders without anyone having changed it. + # SQLFLOW_METRIC_NAME_PREFIX and SQLFLOW_WEBHOOK_MAX_BODY_BYTES are left + # out for the same reason. + # pipeline.yml listens on SQLFLOW_WEBHOOK_PORT. PORT tells Render the + # same port, so it routes there rather than scanning for one. + - key: SQLFLOW_WEBHOOK_PORT + value: "10000" + - key: PORT + value: "10000" + - key: SQLFLOW_LOG_LEVEL + value: INFO + + # The API, from the same image. + - type: web + name: sqlflow-metrics-api + runtime: docker + dockerfilePath: ./render/Dockerfile + dockerContext: ./render + dockerCommand: /app/bin/serve.sh + plan: 0.5c-512mb + region: virginia + healthCheckPath: /healthz + autoDeployTrigger: "off" + envVars: + - key: SQLFLOW_POSTGRES_URI + fromDatabase: + name: sqlflow-metrics-db + property: connectionString + # Prompted for during the deploy, not generated. Render's generated + # values hold '+', '/' and '=', and one pasted into a URL as it is + # answers 401: '+' reads as a space. + - key: SQLFLOW_SERVE_CLIENT_ID + sync: false + - key: SQLFLOW_SERVE_PORT + value: "8080" + - key: PORT + value: "8080" + - key: SQLFLOW_LOG_LEVEL + value: INFO diff --git a/render/Dockerfile b/render/Dockerfile new file mode 100644 index 00000000..7f662f8b --- /dev/null +++ b/render/Dockerfile @@ -0,0 +1,29 @@ +# The published sqlflow image carries the matching libduckdb and sets +# SQLFLOW_DUCKDB_LIB. Pinned: a later release could change the config schema +# under this template, and a copy deployed from the button never rebuilds on +# its own. Must match the Makefile and docker-compose.yml. +# +# An argument so an unreleased build can be tried locally: +# docker build --build-arg SQLFLOW_IMAGE=turbolytics/sql-flow: render +ARG SQLFLOW_IMAGE=turbolytics/sql-flow:v2026.09.18.1 +FROM ${SQLFLOW_IMAGE} + +# psql and pg_isready apply the migrations. curl sends the install events. +RUN apt-get update \ + && apt-get install -y --no-install-recommends postgresql-client curl \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +COPY pipeline.yml serve.yml /app/ +COPY migrations /app/migrations +COPY bin /app/bin + +RUN chmod +x /app/bin/*.sh + +# Logs and window timestamps are UTC regardless of the host. +ENV TZ=UTC + +# The ingest service's entrypoint. The API service overrides it with +# bin/serve.sh. +ENTRYPOINT ["/app/bin/entrypoint.sh"] diff --git a/render/Makefile b/render/Makefile new file mode 100644 index 00000000..7bdbedfb --- /dev/null +++ b/render/Makefile @@ -0,0 +1,40 @@ +# Must match the Dockerfile and docker-compose.yml. +SQLFLOW_IMAGE ?= turbolytics/sql-flow:v2026.09.18.1 + +.PHONY: validate rollups up test psql clean + +# render.yaml is checked against Render's published schema, fetched each run: +# Render renames plans and deprecates keys, and a vendored copy would pass a +# file Render had started to refuse. Needs uv: https://docs.astral.sh/uv/ +validate: + ./bin/validate-render.py ../render.yaml + docker run --rm -v $(CURDIR):/w -w /w \ + -e SQLFLOW_POSTGRES_URI=x -e SQLFLOW_WEBHOOK_HMAC_SECRET=x -e SQLFLOW_WRITER_ID=x \ + $(SQLFLOW_IMAGE) validate pipeline.yml + docker run --rm -v $(CURDIR):/w -w /w \ + -e SQLFLOW_POSTGRES_URI=x -e SQLFLOW_SERVE_CLIENT_ID=x \ + $(SQLFLOW_IMAGE) validate serve.yml + docker run --rm -v $(CURDIR):/w -w /w $(SQLFLOW_IMAGE) validate rollups.yml + @# rollup check requires --serve. rollups.yml generates no dataset, so the + @# serve file is only loaded, and the migration is what is compared. + docker run --rm -v $(CURDIR):/w -w /w \ + -e SQLFLOW_POSTGRES_URI=x -e SQLFLOW_SERVE_CLIENT_ID=x \ + $(SQLFLOW_IMAGE) rollup check \ + -c rollups.yml --migration migrations/0003_rollups.sql --serve serve.yml + +rollups: + docker run --rm -v $(CURDIR):/w -w /w $(SQLFLOW_IMAGE) rollup ddl \ + -c rollups.yml > migrations/0003_rollups.sql + @echo "# regenerated migrations/0003_rollups.sql" + +up: + docker compose up --build + +test: + SQLFLOW_IMAGE=$(SQLFLOW_IMAGE) test/e2e.sh + +psql: + docker compose exec postgres psql -U metrics -d metrics + +clean: + docker compose down -v diff --git a/render/README.md b/render/README.md new file mode 100644 index 00000000..d0b04a9e --- /dev/null +++ b/render/README.md @@ -0,0 +1,287 @@ +# Deploy to Render: a metrics pipeline + +[![Deploy to Render](https://render.com/images/deploy-to-render-button.svg)](https://render.com/deploy?repo=https://github.com/turbolytics/sql-flow) + +One click creates a Postgres, a sqlflow pipeline that accepts metrics over a +signed webhook and aggregates them by minute, and an HTTP API that reads them +back. Everything it deploys is in this directory, and +[`render.yaml`](../render.yaml) at the repository root declares it. + +It is a paid deploy: a `0.1c-256mb` Postgres and two `0.5c-512mb` web +services, in `virginia`. Render shows the price before you confirm. + +## Deploy + +Render asks for three values. Make the first two before you click, and keep +them. You choose both, so you have +them when the deploy finishes: + +```sh +openssl rand -hex 32 # SQLFLOW_WEBHOOK_HMAC_SECRET +openssl rand -hex 16 # SQLFLOW_SERVE_CLIENT_ID +``` + +| Prompt | What it does | +|---|---| +| `SQLFLOW_WEBHOOK_HMAC_SECRET` | Signs what you send. Any long random string. | +| `SQLFLOW_SERVE_CLIENT_ID` | Reads what you sent, as `?client_id=`. Letters, digits, `.`, `_`, `~` and `-` only, so it can be pasted into a URL. | +| `SQLFLOW_TELEMETRY` | Leave it blank, or type `off`. See [What this sends](#what-this-sends). | + +A blank value fails its service, on purpose. The pipeline writes to your +database and will not start unsigned unless you set +`SQLFLOW_WEBHOOK_AUTH=none` yourself. The client id is the only thing between +a reader and your metrics, so it has no default. It travels in the URL: treat +it as you would a link to a private document, and change it on the API +service if it leaks. + +When the deploy finishes, copy the URLs of `sqlflow-metrics-ingest` and +`sqlflow-metrics-api` from the Render dashboard. + +## What this sends + +Unless you type `off` at the `SQLFLOW_TELEMETRY` prompt, this deploy sends the +sqlflow maintainers two events, once each, for as long as it exists. They +tell us that the template was deployed and that it worked. This is the whole +of both: + +```json +{"name": "install.deployed", "type": "count", "value": 1, + "dimensions": {"install_id": "3f0c1b7e-…", "source": "render", + "template": "render-metrics", "sqlflow_version": "v2026.09.18.1"}} +``` + +```json +{"name": "install.first_request", "type": "count", "value": 1, + "dimensions": {"install_id": "3f0c1b7e-…", "source": "render", + "template": "render-metrics", "sqlflow_version": "v2026.09.18.1"}} +``` + +| Field | What it is | +|---|---| +| `install_id` | A random uuid your database made for itself in `migrations/0005_install.sql`. It is derived from nothing: not your account, a hostname, an address or a Render id. It lets us count a deploy once however often it restarts. | +| `source`, `template` | Constants: that this is the Render template. | +| `sqlflow_version` | The image tag. | + +The first is sent when the pipeline first starts. The second is sent when your +database holds its first metric, from anyone. **Your metrics are never read +for it and never sent:** not a name, a value, a dimension or a count. +`bin/telemetry.sh` asks Postgres one question, whether `metrics_1m` has a row. + +They go to `https://telemetry.turbolytics.io`, which is this same template +with signatures off. As with any HTTP request, the receiving end sees the +address it came from. The pipeline neither reads nor stores it. Render's +request log for that service holds it for a while, as any host's does. + +The ingest service logs each event in full when it sends it. `bin/telemetry.sh` +is 91 lines and is the only code that sends anything. The sqlflow +binary itself never phones home. + +**To turn it off**, type `off` at the prompt, or set `SQLFLOW_TELEMETRY=off` on +`sqlflow-metrics-ingest` later. `false`, `0` and `no` also work. Nothing is +sent, and the log says so. + +A collector that is down, slow or gone never affects your deploy. A send is +bounded at five seconds, runs in the background, and is retried later. + +## Send a metric + +```sh +export URL=https://sqlflow-metrics-ingest-xxxx.onrender.com +export SECRET= +bin/send.sh '{"name":"hello","type":"count","dimensions":{"from":"readme"}}' +``` + +``` +{"status":"received"} +``` + +`bin/send.sh` is short. The signature is an HMAC-SHA256 of the exact +bytes of the body, hex-encoded, in `X-Signature-256` with a `sha256=` prefix: + +```sh +sig=$(printf '%s' "$body" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //') +curl -X POST "$URL/events" -H "X-Signature-256: sha256=$sig" --data-binary "$body" +``` + +## Read it back + +A metric is readable once its minute closes, one to two minutes after you +send it. Ask which series exist: + +```sh +export API=https://sqlflow-metrics-api-xxxx.onrender.com +export CLIENT_ID= +curl -sG "$API/v1/datasets/series" --data-urlencode "client_id=$CLIENT_ID" +``` + +Then read one: + +```sh +curl -sG "$API/v1/datasets/metric" --data-urlencode "client_id=$CLIENT_ID" \ + --data-urlencode 'name=hello' +``` + +**The pipeline answers 200 to anything signed.** It acknowledges a request +before it reads it, so a metric with no `name`, a body that is not JSON, and +a bare JSON array all answer `{"status":"received"}` and store nothing. If a +metric does not appear in `series` after two minutes, check it against the +rules below. + +## The metric + +One metric: + +```json +{ + "name": "checkout.completed", + "type": "count", + "value": 1, + "dimensions": {"region": "us-east", "plan": "pro"}, + "timestamp": "2026-09-18T14:03:22Z" +} +``` + +Several, in one signed request: + +```json +{"metrics": [{"name": "a", "type": "count"}, {"name": "b", "type": "gauge", "value": 0.73}]} +``` + +| Field | Required | Default | Rule | +|---|---|---|---| +| `name` | yes | | 1 to 200 bytes. | +| `type` | yes | | `count` or `gauge`. | +| `value` | no | `1` | A finite number. | +| `dimensions` | no | `{}` | A JSON object of at most 16 keys. Every value is stored as a string: `1` and `"1"` are the same. | +| `timestamp` | no | arrival | RFC 3339. More than a minute in the future is dropped. Older than the open minutes, about two, is dropped: there is no backfill. | + +A series is a name, a type and a set of dimensions. Key order does not +matter. + +Send `timestamp` when order matters. Metrics without one that arrive within +the same five seconds share an arrival time, and which of them is a gauge's +`value_last` is not defined. + +## What is stored + +Every minute of every series stores five values: `value_sum`, `value_count`, +`value_min`, `value_max` and `value_last`. `type` does not change what is +stored. It says which value to read: `value_sum` for a count, `value_last` +for a gauge. An average is `value_sum / value_count`, at any grain. + +### More than one pipeline instance + +You can run several instances of `sqlflow-metrics-ingest`, and Render runs two +for a moment during every deploy, the old beside the new. Each instance holds +its own minute in memory, so two of them can each hold part of the same +minute of the same series. + +The pipeline therefore writes to `metrics_1m_writers`, keyed on the series +and on a `writer` id that `bin/entrypoint.sh` makes new on every start. An +instance replaces only what it published itself. A trigger merges every +writer's row into `metrics_1m` inside the same transaction: sums add, min and +max nest, and `value_last` is the reading with the latest event time across +instances. Everything else reads `metrics_1m` and never sees a writer. + +Keyed on the series alone, the second instance to publish replaced the first: +two instances sent 7 and 5 of one minute stored 7. `make test` runs two +instances and sends each a share of one minute, and +`internal/rendertemplate` runs four concurrent writers through forty rounds +and holds every grain to what they published. + +Never set `SQLFLOW_WRITER_ID` yourself. Two instances that share one replace +each other's minutes, which is the defect it exists to prevent. + +`metrics_1m_writers` holds one row per instance per minute per series, so with +one instance it is as large as `metrics_1m`. Nothing reads a row of it once +its minute is merged and no instance can publish that minute again, a few +minutes later. Deleting older rows is safe and changes no other table. + +### Coarser grains + +Postgres keeps the same five values at 5m, 15m, 1h, 6h and 1d, by triggers +that run inside the pipeline's own write. [`rollups.yml`](rollups.yml) +declares them, `migrations/0003_rollups.sql` is generated from it with +`make rollups`, and `make validate` fails when it has drifted. + +## The API + +Every request except `/healthz` and `/metrics` sends `?client_id=`. It names +the caller in the log, and a request without it answers 401. + +An answer is held for 30 seconds. Ask twice within that and the second answer +is the first, with `"cache": "hit"` and its age in `age_ms`. + +| Route | Returns | +|---|---| +| `GET /v1/datasets/series` | Every series: `name`, `type`, `dimensions`, `first_bucket`, `last_bucket`. | +| `GET /v1/datasets/metric?name=…` | Per bucket per series: the five values, `type`, and `dimensions`. | +| `GET /v1/datasets/metric_total?name=…` | Per bucket, across every series of the name: `value_sum`, `value_count`, `value_min`, `value_max`. | + +`dimensions` in a response is a JSON object encoded as a string. Parse it. + +`metric` and `metric_total` take: + +| Param | Meaning | +|---|---| +| `name` | Required. Without it the answer is empty. | +| `since`, `until` | RFC 3339 with an offset. Encode `+` as `%2B`. Default: the last hour. | +| `grain` | Optional. Without it the API picks the finest grain that covers the range, which is what you want. Pin one only with a range at least as wide as its bucket. | +| `dimensions` | `metric` only. A JSON object. The answer holds the series whose dimensions contain every pair in it. | + +`dimensions={"region":"us-east"}` returns every plan in us-east, each its own +series. `dimensions={"region":"us-east","plan":"pro"}` returns one series. A +`dimensions` that is not JSON matches nothing. + +The API snaps a range to the grain's bucket boundaries, and a bucket is in +the range when its start is. The response's `range` says what was resolved. +So `grain=1d` with the default range, the last hour, resolves to an empty +range and answers no rows: no day starts inside the last hour. Send `since` +at or before the start of the first bucket you want. + +Use `metric_total` for a name with many series, such as one per user. It +reads one row per bucket. `metric` reads a row per bucket per series, and a +response stops at 10,000 rows and says `"truncated": true`. There is no +filter-then-sum: sum the series of a `metric` response yourself. Every value +but `value_last` adds. + +| Grain | Widest range | +|---|---| +| `1m` | 6 hours | +| `5m` | 1 day | +| `15m` | 3 days | +| `1h` | 14 days | +| `6h` | 90 days | +| `1d` | 365 days | + +## Health + +Both services answer `GET` and `HEAD /healthz` without a signature or a client +id, and Render checks both. Point an uptime monitor at either. The pipeline's +answers 200 while it admits metrics and 503 once it is shutting down. A +health check stores nothing and is not counted as a request. + +## Settings + +Set on the `sqlflow-metrics-ingest` service, in the dashboard. None of these +is in `render.yaml`, on purpose: Render keeps a variable the Blueprint does +not name and rewrites one it does, so a setting made here survives a sync. + +| Variable | Default | Meaning | +|---|---|---| +| `SQLFLOW_WEBHOOK_AUTH` | `hmac` | `none` accepts unsigned requests from anyone who has the URL. Anything but exactly `none` is signed. | +| `SQLFLOW_WEBHOOK_HMAC_SECRET` | | Required under `hmac`. | +| `SQLFLOW_METRIC_NAME_PREFIX` | empty | When set, a metric whose name does not start with it is dropped. Letters, digits, `.`, `_` and `-`. | +| `SQLFLOW_WEBHOOK_MAX_BODY_BYTES` | 26214400 | A larger body is refused with 413. | +| `SQLFLOW_TELEMETRY` | on | `off` sends nothing. See [What this sends](#what-this-sends). | + +## Run it locally + +```sh +make up # Postgres, two pipeline instances on :10000 and :10001, the API on :8080 +make test # two pipeline instances: posts metrics, reads them back at every grain +make validate # the configs, the generated migration, and render.yaml +``` + +Locally the secret is `local-secret`, the client id is `local-dev`, and a +minute closes in seconds. diff --git a/render/bin/entrypoint.sh b/render/bin/entrypoint.sh new file mode 100755 index 00000000..69ddddf8 --- /dev/null +++ b/render/bin/entrypoint.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +# The ingest service's entrypoint: check the environment, provision the +# schema, then run the pipeline. +set -euo pipefail + +# Checked here rather than left to sqlflow. An unset template variable renders +# as an empty string, and the error for an empty DSN does not name the cause. +if [ -z "${SQLFLOW_POSTGRES_URI:-}" ]; then + echo "SQLFLOW_POSTGRES_URI is not set" >&2 + exit 2 +fi + +# Unsigned mode is chosen, never reached by omission. Render lets a deploy +# leave the secret's prompt blank, and a blank secret under hmac would be a +# public write endpoint into this database. +# +# This script is the one place the mode is decided. It exports the mode it +# settled on, so pipeline.yml never reads a value this script did not check. +# The two once decided separately and disagreed about an empty string: this +# script read it as hmac and was satisfied by the secret, the template read it +# as not-hmac and rendered no signature block, and an unsigned request was +# answered 200 by a pipeline whose owner believed it was signed. +case "${SQLFLOW_WEBHOOK_AUTH:-hmac}" in + hmac) + SQLFLOW_WEBHOOK_AUTH=hmac + if [ -z "${SQLFLOW_WEBHOOK_HMAC_SECRET:-}" ]; then + echo "SQLFLOW_WEBHOOK_HMAC_SECRET is not set. Set it, or set SQLFLOW_WEBHOOK_AUTH=none to accept unsigned requests." >&2 + exit 2 + fi + ;; + none) + SQLFLOW_WEBHOOK_AUTH=none + echo "webhook accepts unsigned requests: SQLFLOW_WEBHOOK_AUTH=none" >&2 + ;; + *) + echo "SQLFLOW_WEBHOOK_AUTH is '${SQLFLOW_WEBHOOK_AUTH}'. Use hmac or none." >&2 + exit 2 + ;; +esac +export SQLFLOW_WEBHOOK_AUTH + +# The prefix is rendered into the handler's SQL inside single quotes. +case "${SQLFLOW_METRIC_NAME_PREFIX:-}" in + *[!A-Za-z0-9._-]*) + echo "SQLFLOW_METRIC_NAME_PREFIX may hold only letters, digits, '.', '_' and '-'" >&2 + exit 2 + ;; +esac + +# Names this process in every row it publishes, so that another instance +# publishing the same minute adds to it rather than replacing it. Made here +# and never read from the environment: a value someone set once would be +# shared by every instance, which is the defect it exists to prevent. New on +# every start, because a restarted process has lost its window and must add +# to what the last one published, not replace it. +SQLFLOW_WRITER_ID="$(od -An -N8 -tx1 /dev/urandom | tr -d ' \n')" +if [ "${#SQLFLOW_WRITER_ID}" != 16 ]; then + echo "could not make a writer id from /dev/urandom" >&2 + exit 2 +fi +export SQLFLOW_WRITER_ID + +/app/bin/migrate.sh + +# Two events per install, ever, to the sqlflow maintainers: that this was +# deployed, and that it received its first metric. SQLFLOW_TELEMETRY=off stops +# both. In the background and detached from this script's fate: it logs and +# swallows its own failures, and nothing it does can stop the pipeline below. +/app/bin/telemetry.sh & + +# exec keeps sqlflow as PID 1, so the platform's SIGTERM reaches it and the +# graceful drain runs. Without exec, bash holds PID 1 and forwards nothing. +exec sqlflow run -c /app/pipeline.yml "$@" diff --git a/render/bin/migrate.sh b/render/bin/migrate.sh new file mode 100755 index 00000000..ffbf48b1 --- /dev/null +++ b/render/bin/migrate.sh @@ -0,0 +1,85 @@ +#!/usr/bin/env bash +# Applies every migrations/*.sql exactly once, in filename order. +# +# Safe to run on every process start, which is how both services use it. Two +# instances starting during the same deploy cannot race: the applied check and +# the migration run in one transaction behind an advisory lock, so the second +# instance sees the first's commit and skips. +set -euo pipefail + +if [ -z "${SQLFLOW_POSTGRES_URI:-}" ]; then + echo "SQLFLOW_POSTGRES_URI is not set" >&2 + exit 2 +fi + +for tool in psql pg_isready; do + if ! command -v "$tool" >/dev/null; then + echo "$tool not found on PATH" >&2 + exit 2 + fi +done + +# Wait for the database to accept connections before the first psql. +# +# Render resolves fromDatabase as soon as the database record exists, but a +# freshly provisioned instance is not listening yet. psql exits 2 on a refused +# connection and set -e turns that into a failed deploy, which is what a +# Blueprint's first deploy hit: "connection to server ... failed: Connection +# refused", exit 2, before a single migration ran. +# +# Nothing caught it earlier because every other way into this script gates on +# readiness first -- compose on service_healthy, CI on pg_isready -- and on a +# database that already exists there is nothing to wait for. +wait_seconds="${SQLFLOW_MIGRATE_WAIT_SECONDS:-180}" +if ! pg_isready -q -d "$SQLFLOW_POSTGRES_URI"; then + echo "waiting up to ${wait_seconds}s for the database to accept connections" + deadline=$((SECONDS + wait_seconds)) + until pg_isready -q -d "$SQLFLOW_POSTGRES_URI"; do + if [ "$SECONDS" -ge "$deadline" ]; then + echo "database did not accept connections within ${wait_seconds}s" >&2 + exit 2 + fi + sleep 2 + done + echo "database is accepting connections" +fi + +cd "$(dirname "$0")/.." + +# Under the same lock as the migrations: CREATE TABLE IF NOT EXISTS is not +# safe against itself, and two concurrent runs can fail on the system catalog's +# unique index. client_min_messages silences the "already exists, skipping" +# notice that every run after the first would otherwise print. +psql "$SQLFLOW_POSTGRES_URI" -v ON_ERROR_STOP=1 -q <<'SQL' +SET client_min_messages = warning; +BEGIN; +SELECT pg_advisory_xact_lock(hashtext('sqlflow_render_metrics_migrations')) AS locked \gset +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +COMMIT; +SQL + +for file in migrations/*.sql; do + version="$(basename "$file" .sql)" + + # Postgres DDL is transactional, so a migration that fails rolls back whole + # and records no version. The next run retries it from a clean state. + # + # "AS locked \gset" consumes the lock's result row; without it psql prints a + # table on every run. \echo interpolates :version only outside quotes. + psql "$SQLFLOW_POSTGRES_URI" -v ON_ERROR_STOP=1 -q -v version="$version" <.onrender.com SECRET= bin/send.sh '{"name":"hello","type":"count"}' +# +# Leave SECRET unset for a pipeline running with SQLFLOW_WEBHOOK_AUTH=none. +set -euo pipefail + +body="${1:?usage: URL=... [SECRET=...] send.sh ''}" +url="${URL:?URL is not set}" + +args=(-sS -X POST "$url/events" -H 'Content-Type: application/json' --data-binary "$body") +if [ -n "${SECRET:-}" ]; then + # The signature covers the exact bytes sent. printf '%s' adds no newline, + # and --data-binary sends the body as is, where -d would strip newlines. + sig="$(printf '%s' "$body" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')" + args+=(-H "X-Signature-256: sha256=$sig") +fi +curl "${args[@]}" +echo diff --git a/render/bin/serve.sh b/render/bin/serve.sh new file mode 100755 index 00000000..afab3d0f --- /dev/null +++ b/render/bin/serve.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# The API service's entrypoint. +set -euo pipefail + +if [ -z "${SQLFLOW_POSTGRES_URI:-}" ]; then + echo "SQLFLOW_POSTGRES_URI is not set" >&2 + exit 2 +fi + +# The client id is the only thing between a reader and this database's +# metrics, so there is no default. It travels in a URL as ?client_id=, where +# '+' reads as a space and '/' and '=' need encoding: an id with any of them +# answers 401 to a caller who pasted it as it is. +if [ -z "${SQLFLOW_SERVE_CLIENT_ID:-}" ]; then + echo "SQLFLOW_SERVE_CLIENT_ID is not set. Set it to a long random string, such as the output of: openssl rand -hex 16" >&2 + exit 2 +fi +case "$SQLFLOW_SERVE_CLIENT_ID" in + *[!A-Za-z0-9._~-]*) + echo "SQLFLOW_SERVE_CLIENT_ID may hold only letters, digits, '.', '_', '~' and '-', so it can be pasted into a URL. Try: openssl rand -hex 16" >&2 + exit 2 + ;; +esac + +# Both services migrate. Whichever starts first applies the schema, and the +# advisory lock in migrate.sh makes the other skip. Without it the API would +# fail its first deploy whenever it won the race: serve prepares every +# dataset at startup, against tables that do not exist yet. +/app/bin/migrate.sh + +exec sqlflow serve -c /app/serve.yml "$@" diff --git a/render/bin/telemetry.sh b/render/bin/telemetry.sh new file mode 100755 index 00000000..dc757f91 --- /dev/null +++ b/render/bin/telemetry.sh @@ -0,0 +1,94 @@ +#!/usr/bin/env bash +# Tells the sqlflow maintainers that this template was deployed, and that it +# received its first metric. Two events, each sent once per install, and +# nothing else, ever. render/README.md, "What this sends", prints both. +# +# SQLFLOW_TELEMETRY=off sends nothing. Also: false, 0, no. +# +# bin/entrypoint.sh starts this in the background after the migrations. It must +# never be the reason a pipeline does not start or stops: every failure here is +# logged and swallowed, a send is bounded at five seconds, and a collector that +# is down costs a retry on a later tick, not a deploy. +set -uo pipefail + +log() { echo "telemetry: $*" >&2; } + +case "$(printf '%s' "${SQLFLOW_TELEMETRY:-on}" | tr '[:upper:]' '[:lower:]')" in + off|false|0|no) + log "off (SQLFLOW_TELEMETRY=${SQLFLOW_TELEMETRY}). Nothing is sent." + exit 0 + ;; +esac + +# A hostname the maintainers own, not a platform's URL: a deployed copy never +# updates itself, so this is the address it calls for as long as it runs. +URL="${SQLFLOW_TELEMETRY_URL:-https://telemetry.turbolytics.io}" +POLL="${SQLFLOW_TELEMETRY_POLL_SECONDS:-30}" +# Where this template runs. Other templates report to the same collector, each +# with its own constant. Not a setting: a template knows where it runs. +SOURCE=render +TEMPLATE=render-metrics + +sql() { psql "$SQLFLOW_POSTGRES_URI" -X -Atq -v ON_ERROR_STOP=1 -c "$1" 2>/dev/null; } + +# The first line of `sqlflow --version` is "sqlflow v2026.09.18.1". +VERSION="$(sqlflow --version 2>/dev/null | awk 'NR==1 {print $2}')" +case "$VERSION" in ''|*[!A-Za-z0-9._-]*) VERSION=unknown ;; esac + +# try +# Returns 0 once the event is sent, now or earlier. Returns 1 to try again. +try() { + local name="$1" col="$2" id body + + [ "$(sql "SELECT ${col}_sent_at IS NOT NULL FROM install")" = t ] && return 0 + + # The lease. One process gets the row back; the others get nothing. + id="$(sql "UPDATE install SET ${col}_claimed_at = now() + WHERE ${col}_sent_at IS NULL + AND (${col}_claimed_at IS NULL OR ${col}_claimed_at < now() - interval '2 minutes') + RETURNING install_id")" + case "$id" in ''|*[!0-9a-f-]*) return 1 ;; esac + + body="{\"name\":\"$name\",\"type\":\"count\",\"value\":1,\"dimensions\":{\"install_id\":\"$id\",\"source\":\"$SOURCE\",\"template\":\"$TEMPLATE\",\"sqlflow_version\":\"$VERSION\"}}" + + # --fail: only a 2xx counts. A custom domain that is still propagating + # answers 404, and an install recorded as sent on a 404 is never counted. + if curl --fail --silent --show-error --max-time 5 -o /dev/null \ + -X POST "$URL/events" -H 'Content-Type: application/json' --data-binary "$body" 2>/dev/null; then + sql "UPDATE install SET ${col}_sent_at = now() WHERE ${col}_sent_at IS NULL" >/dev/null + log "sent $name to $URL: $body" + log "this is one of two events this install ever sends. SQLFLOW_TELEMETRY=off stops them. See render/README.md." + return 0 + fi + + sql "UPDATE install SET ${col}_claimed_at = NULL WHERE ${col}_sent_at IS NULL" >/dev/null + return 1 +} + +tries=0 +while :; do + deployed=1; first=1 + try install.deployed deployed && deployed=0 + + # The first metric, from anyone. Nothing about it is read: the question is + # whether a row exists. install.* names do not count. They are telemetry, not + # a deployer's metric, and a database that is itself a collector holds + # nothing else. + if [ "$(sql "SELECT EXISTS (SELECT 1 FROM metrics_1m WHERE name NOT LIKE 'install.%')")" = t ]; then + try install.first_request first_request && first=0 + fi + + if [ "$deployed" = 0 ] && [ "$first" = 0 ]; then + exit 0 + fi + + # Every POLL seconds for the first hour, then every five minutes: a deploy + # nobody ever sends a metric to should not query its database twice a minute + # for as long as it lives. + tries=$((tries + 1)) + if [ "$tries" -gt $((3600 / POLL)) ] && [ "$POLL" -lt 300 ]; then + sleep 300 + else + sleep "$POLL" + fi +done diff --git a/render/bin/validate-render.py b/render/bin/validate-render.py new file mode 100755 index 00000000..f84159f8 --- /dev/null +++ b/render/bin/validate-render.py @@ -0,0 +1,142 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.11" +# dependencies = ["jsonschema", "pyyaml"] +# /// +"""Check render.yaml against Render's published Blueprint schema. + +The schema is fetched, not vendored. Render renames plans and deprecates keys +between releases -- `autoDeploy` became `autoDeployTrigger`, and the Postgres +plan `basic-256mb` became `0.1c-256mb` -- and a vendored copy would keep +passing a file Render had already started to reject. Fetching it means CI is +what notices, rather than a failed sync against the live database. + +Usage: + bin/validate-render.py [render.yaml] + +Set RENDER_SCHEMA to a local path or another URL to check against that +instead, which is also how this runs without a network. + +Exit status: 0 valid, 1 invalid, 2 the schema could not be read. A caller +checking that a bad file is rejected has to require 1, because an unreachable +schema fails everything it is handed. +""" + +import json +import os +import sys +import time +import urllib.request + +import yaml +from jsonschema import Draft202012Validator + +SCHEMA_URL = "https://render.com/schema/render.yaml.json" + +# Identifies the caller in Render's logs, rather than leaving it Python-urllib. +USER_AGENT = "sql-flow render.yaml validator" + +# The schema is served chunked, and a proxy between here and Render can hand +# back a short body. That reads as corrupt JSON, not as a network error, so it +# is worth several tries before believing it. +FETCH_ATTEMPTS = 5 + + +def load_schema(source): + """Read the schema from a URL, or from a path when given one.""" + if not source.startswith(("http://", "https://")): + with open(source) as f: + return json.load(f) + + request = urllib.request.Request(source, headers={"User-Agent": USER_AGENT}) + for attempt in range(1, FETCH_ATTEMPTS + 1): + try: + with urllib.request.urlopen(request, timeout=30) as response: + # Read to completion here, so a body that arrives short is a + # retry rather than a parse error. + return json.loads(response.read()) + except Exception as err: + if attempt == FETCH_ATTEMPTS: + raise + print( + f"fetching {source} failed ({type(err).__name__}), " + f"retrying {attempt}/{FETCH_ATTEMPTS - 1}", + file=sys.stderr, + ) + time.sleep(attempt) + + +def leaves(error): + """Flatten an error to the ones that name a field. + + A service is a union of per-type schemas, so a bad `plan` surfaces as the + whole service dict being "not valid under any of the given schemas", with + every branch's complaints underneath. Reporting the wrong branch is worse + than reporting nothing: the union's `keyValue` branch wants an + `ipAllowList` this file will never have. + + Services are discriminated by `type`, so the branch that did not object to + `type` is the one the author meant. Among the rest, the branch that + objected least. + """ + if not error.context: + return [error] + + branches = {} + for sub in error.context: + branches.setdefault(sub.schema_path[0], []).append(sub) + + matched = [ + errors + for errors in branches.values() + if not any(list(e.path)[-1:] == ["type"] for e in errors) + ] + chosen = min(matched or list(branches.values()), key=len) + + return [leaf for sub in chosen for leaf in leaves(sub)] + + +def report(errors): + """Format errors as `path: message`, dropping cascade noise.""" + # `unevaluatedProperties: false` at the root re-reports every top-level key + # whenever anything beneath it fails. Only useful when it is the sole error. + if len(errors) > 1: + errors = [e for e in errors if e.validator != "unevaluatedProperties" or e.path] + + lines = [] + for error in errors: + for leaf in leaves(error): + where = ".".join(str(p) for p in leaf.absolute_path) or "(root)" + line = f"{where}: {leaf.message}" + if line not in lines: + lines.append(line) + return lines + + +def main(argv): + path = argv[1] if len(argv) > 1 else "render.yaml" + source = os.environ.get("RENDER_SCHEMA", SCHEMA_URL) + + try: + schema = load_schema(source) + except Exception as err: + print(f"could not read {source}: {type(err).__name__}: {err}", file=sys.stderr) + return 2 + + with open(path) as f: + document = yaml.safe_load(f) + + validator = Draft202012Validator(schema) + errors = sorted(validator.iter_errors(document), key=lambda e: list(e.path)) + if errors: + for line in report(errors): + print(f"{path}: {line}", file=sys.stderr) + print(f"{path}: invalid against {source}", file=sys.stderr) + return 1 + + print(f"{path}: valid against {source}") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/render/docker-compose.yml b/render/docker-compose.yml new file mode 100644 index 00000000..5af60d45 --- /dev/null +++ b/render/docker-compose.yml @@ -0,0 +1,106 @@ +# The template on one machine: the local run, and what CI tests. +# +# The window and flush settings close a minute in seconds, so a test does not +# wait two minutes for its metric. A deploy leaves them at their defaults. +# +# No service here ever calls the maintainers' collector. Every pipeline is +# pointed at `collector`, which is this same template with signatures off and +# the install. prefix on: what the real one is. It shares the database, so the +# test reads install events back through the same API. +# +# An idle bound below the window's 60 seconds has a cost the test works +# around: a minute can close while it is still the current minute, and a +# metric of that minute that arrives afterwards is dropped as late. A deploy's +# bound is 60, where a closed minute has always ended. So the test sends each +# series in one burst, and starts its burst in a fresh minute, after the one +# install.deployed was sent in. +# +# Two pipeline instances, on purpose. Each holds its own window, and the test +# sends each a share of one minute: what Postgres holds must be the sum. Two +# instances is also what every deploy runs for a moment, the old beside the +# new. +services: + postgres: + image: postgres:18 + environment: + POSTGRES_USER: metrics + POSTGRES_PASSWORD: metrics + POSTGRES_DB: metrics + ports: + - "127.0.0.1:${POSTGRES_HOST_PORT:-5442}:5432" + healthcheck: + test: ["CMD-SHELL", "pg_isready -U metrics -d metrics"] + interval: 2s + timeout: 3s + retries: 15 + + ingest: &ingest + build: + context: . + args: + SQLFLOW_IMAGE: ${SQLFLOW_IMAGE:-turbolytics/sql-flow:v2026.09.18.1} + depends_on: + postgres: + condition: service_healthy + environment: + SQLFLOW_POSTGRES_URI: postgresql://metrics:metrics@postgres:5432/metrics + # No value: passed through when the caller sets it and absent when not, + # which is how Render leaves it. The default is the entrypoint's. + SQLFLOW_WEBHOOK_AUTH: + SQLFLOW_WEBHOOK_HMAC_SECRET: ${SQLFLOW_WEBHOOK_HMAC_SECRET-local-secret} + SQLFLOW_WEBHOOK_PORT: "10000" + SQLFLOW_METRIC_NAME_PREFIX: ${SQLFLOW_METRIC_NAME_PREFIX:-} + # No value: on unless the caller says off, as on Render. + SQLFLOW_TELEMETRY: + SQLFLOW_TELEMETRY_URL: ${SQLFLOW_TELEMETRY_URL:-http://collector:10000} + SQLFLOW_TELEMETRY_POLL_SECONDS: "1" + SQLFLOW_WINDOW_GRACE_SECONDS: "3" + SQLFLOW_WINDOW_IDLE_SECONDS: "5" + SQLFLOW_WINDOW_POLL_SECONDS: "1" + SQLFLOW_FLUSH_INTERVAL_SECONDS: "1" + SQLFLOW_LOG_LEVEL: INFO + ports: + - "127.0.0.1:${INGEST_HOST_PORT:-10000}:10000" + + ingest2: + <<: *ingest + ports: + - "127.0.0.1:${INGEST2_HOST_PORT:-10001}:10000" + + collector: + build: + context: . + args: + SQLFLOW_IMAGE: ${SQLFLOW_IMAGE:-turbolytics/sql-flow:v2026.09.18.1} + depends_on: + postgres: + condition: service_healthy + environment: + SQLFLOW_POSTGRES_URI: postgresql://metrics:metrics@postgres:5432/metrics + SQLFLOW_WEBHOOK_AUTH: none + SQLFLOW_METRIC_NAME_PREFIX: install. + SQLFLOW_WEBHOOK_MAX_BODY_BYTES: "4096" + SQLFLOW_WEBHOOK_PORT: "10000" + # The collector reports on nobody, itself included. + SQLFLOW_TELEMETRY: "off" + SQLFLOW_WINDOW_GRACE_SECONDS: "3" + SQLFLOW_WINDOW_IDLE_SECONDS: "5" + SQLFLOW_WINDOW_POLL_SECONDS: "1" + SQLFLOW_FLUSH_INTERVAL_SECONDS: "1" + SQLFLOW_LOG_LEVEL: INFO + + api: + build: + context: . + args: + SQLFLOW_IMAGE: ${SQLFLOW_IMAGE:-turbolytics/sql-flow:v2026.09.18.1} + entrypoint: /app/bin/serve.sh + depends_on: + postgres: + condition: service_healthy + environment: + SQLFLOW_POSTGRES_URI: postgresql://metrics:metrics@postgres:5432/metrics + SQLFLOW_SERVE_CLIENT_ID: local-dev + SQLFLOW_LOG_LEVEL: INFO + ports: + - "127.0.0.1:${API_HOST_PORT:-8080}:8080" diff --git a/render/migrations/0001_metrics_1m.sql b/render/migrations/0001_metrics_1m.sql new file mode 100644 index 00000000..18293d07 --- /dev/null +++ b/render/migrations/0001_metrics_1m.sql @@ -0,0 +1,27 @@ +-- One row per minute per series. A series is a name, a type and a set of +-- dimensions, and dimensions_key is that set in canonical form: keys sorted, +-- every value a string, compact JSON. The pipeline builds it. +-- +-- The five values are what every coarser grain can be merged from exactly: +-- sums add, min and max nest, and last is the latest bucket's. An average is +-- value_sum / value_count at any grain. An average stored per minute could +-- not be merged: a minute of one sample would weigh as much as one of 1,000. +CREATE TABLE metrics_1m ( + bucket TIMESTAMPTZ NOT NULL, + name TEXT NOT NULL, + type TEXT NOT NULL CHECK (type IN ('count', 'gauge')), + dimensions_key TEXT NOT NULL, + value_sum DOUBLE PRECISION NOT NULL, + value_count BIGINT NOT NULL, + value_min DOUBLE PRECISION NOT NULL, + value_max DOUBLE PRECISION NOT NULL, + value_last DOUBLE PRECISION NOT NULL, + -- The event time of value_last. + last_at TIMESTAMPTZ NOT NULL, + -- The Postgres clock at the first write. A republished minute keeps it. + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (bucket, name, type, dimensions_key) +); + +-- The API reads one name over a range. +CREATE INDEX metrics_1m_name_bucket ON metrics_1m (name, bucket); diff --git a/render/migrations/0002_series.sql b/render/migrations/0002_series.sql new file mode 100644 index 00000000..744d90f8 --- /dev/null +++ b/render/migrations/0002_series.sql @@ -0,0 +1,37 @@ +-- Every series seen, and where a key turns back into dimensions. The rollup +-- tables carry dimensions_key and never jsonb, so the API joins here to +-- filter by dimension and to return dimensions. +CREATE TABLE series ( + name TEXT NOT NULL, + type TEXT NOT NULL, + dimensions_key TEXT NOT NULL, + -- dimensions_key is JSON text, so the cast is the whole conversion. + dimensions JSONB NOT NULL, + first_bucket TIMESTAMPTZ NOT NULL, + last_bucket TIMESTAMPTZ NOT NULL, + PRIMARY KEY (name, type, dimensions_key) +); + +-- Kept in the writer's transaction, as the rollups are. least and greatest +-- because a republished minute may be older than what is recorded. +CREATE FUNCTION metrics_series() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + INSERT INTO series (name, type, dimensions_key, dimensions, first_bucket, last_bucket) + SELECT name, type, dimensions_key, dimensions_key::jsonb, min(bucket), max(bucket) + FROM changed + GROUP BY name, type, dimensions_key + ON CONFLICT (name, type, dimensions_key) DO UPDATE SET + first_bucket = least(series.first_bucket, excluded.first_bucket), + last_bucket = greatest(series.last_bucket, excluded.last_bucket); + RETURN NULL; +END $fn$; + +-- Two triggers: Postgres refuses a transition table on a trigger with more +-- than one event. An upsert fires both, each with its own rows. +CREATE TRIGGER metrics_series_ins AFTER INSERT ON metrics_1m + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION metrics_series(); +CREATE TRIGGER metrics_series_upd AFTER UPDATE ON metrics_1m + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION metrics_series(); diff --git a/render/migrations/0003_rollups.sql b/render/migrations/0003_rollups.sql new file mode 100644 index 00000000..b4c9410b --- /dev/null +++ b/render/migrations/0003_rollups.sql @@ -0,0 +1,482 @@ +-- Generated by sqlflow rollup ddl. Do not edit: change the rollups file and +-- regenerate. Run it inside one transaction, which a migration runner opens. + +-- Rollup metrics, kept from metrics_1m. +-- Blocks the source's writers until commit, so no write lands between the +-- triggers existing and the backfill reading. +LOCK TABLE "metrics_1m" IN SHARE ROW EXCLUSIVE MODE; + +CREATE TABLE IF NOT EXISTS "metrics_5m" AS +SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + f."dimensions_key" AS "dimensions_key", + sum(f."value_count")::bigint AS "value_count", + (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1] AS "value_last", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_5m_key" + ON "metrics_5m" ("bucket", "name", "type", "dimensions_key") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_15m" AS +SELECT date_bin(INTERVAL '15 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + f."dimensions_key" AS "dimensions_key", + sum(f."value_count")::bigint AS "value_count", + (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1] AS "value_last", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_15m_key" + ON "metrics_15m" ("bucket", "name", "type", "dimensions_key") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_1h" AS +SELECT date_bin(INTERVAL '1 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + f."dimensions_key" AS "dimensions_key", + sum(f."value_count")::bigint AS "value_count", + (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1] AS "value_last", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_1h_key" + ON "metrics_1h" ("bucket", "name", "type", "dimensions_key") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_6h" AS +SELECT date_bin(INTERVAL '6 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + f."dimensions_key" AS "dimensions_key", + sum(f."value_count")::bigint AS "value_count", + (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1] AS "value_last", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_6h_key" + ON "metrics_6h" ("bucket", "name", "type", "dimensions_key") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_1d" AS +SELECT date_bin(INTERVAL '24 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + f."dimensions_key" AS "dimensions_key", + sum(f."value_count")::bigint AS "value_count", + (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1] AS "value_last", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_1d_key" + ON "metrics_1d" ("bucket", "name", "type", "dimensions_key") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_total_5m" AS +SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + sum(f."value_count")::bigint AS "value_count", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_total_5m_key" + ON "metrics_total_5m" ("bucket", "name", "type") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_total_15m" AS +SELECT date_bin(INTERVAL '15 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + sum(f."value_count")::bigint AS "value_count", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_total_15m_key" + ON "metrics_total_15m" ("bucket", "name", "type") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_total_1h" AS +SELECT date_bin(INTERVAL '1 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + sum(f."value_count")::bigint AS "value_count", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_total_1h_key" + ON "metrics_total_1h" ("bucket", "name", "type") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_total_6h" AS +SELECT date_bin(INTERVAL '6 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + sum(f."value_count")::bigint AS "value_count", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_total_6h_key" + ON "metrics_total_6h" ("bucket", "name", "type") NULLS NOT DISTINCT; + +CREATE TABLE IF NOT EXISTS "metrics_total_1d" AS +SELECT date_bin(INTERVAL '24 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS "bucket", + f."name" AS "name", + f."type" AS "type", + sum(f."value_count")::bigint AS "value_count", + max(f."value_max") AS "value_max", + min(f."value_min") AS "value_min", + sum(f."value_sum")::double precision AS "value_sum" +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +WITH NO DATA; +CREATE UNIQUE INDEX IF NOT EXISTS "metrics_total_1d_key" + ON "metrics_total_1d" ("bucket", "name", "type") NULLS NOT DISTINCT; + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_5m"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_5m needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_5m:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '5 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_5m" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_1m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '5 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '5 minutes' + GROUP BY 1, 2, 3, 4 + ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_5m_ins" + AFTER INSERT ON "metrics_1m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_5m"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_5m_upd" + AFTER UPDATE ON "metrics_1m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_5m"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_15m"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_15m needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_15m:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '15 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_15m" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '15 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_5m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '15 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '15 minutes' + GROUP BY 1, 2, 3, 4 + ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_15m_ins" + AFTER INSERT ON "metrics_5m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_15m"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_15m_upd" + AFTER UPDATE ON "metrics_5m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_15m"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_1h"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_1h needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_1h:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '1 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_1h" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '1 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_15m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '1 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '1 hours' + GROUP BY 1, 2, 3, 4 + ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_1h_ins" + AFTER INSERT ON "metrics_15m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_1h"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_1h_upd" + AFTER UPDATE ON "metrics_15m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_1h"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_6h"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_6h needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_6h:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '6 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_6h" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '6 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_1h" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '6 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '6 hours' + GROUP BY 1, 2, 3, 4 + ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_6h_ins" + AFTER INSERT ON "metrics_1h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_6h"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_6h_upd" + AFTER UPDATE ON "metrics_1h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_6h"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_1d"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_1d needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_1d:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '24 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_1d" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '24 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_6h" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '24 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '24 hours' + GROUP BY 1, 2, 3, 4 + ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_1d_ins" + AFTER INSERT ON "metrics_6h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_1d"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_1d_upd" + AFTER UPDATE ON "metrics_6h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_1d"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_total_5m"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_total_5m needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_total_5m:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '5 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_total_5m" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_1m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '5 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '5 minutes' + GROUP BY 1, 2, 3 + ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_5m_ins" + AFTER INSERT ON "metrics_1m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_5m"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_5m_upd" + AFTER UPDATE ON "metrics_1m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_5m"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_total_15m"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_total_15m needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_total_15m:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '15 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_total_15m" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '15 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_total_5m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '15 minutes', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '15 minutes' + GROUP BY 1, 2, 3 + ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_15m_ins" + AFTER INSERT ON "metrics_total_5m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_15m"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_15m_upd" + AFTER UPDATE ON "metrics_total_5m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_15m"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_total_1h"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_total_1h needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_total_1h:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '1 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_total_1h" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '1 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_total_15m" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '1 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '1 hours' + GROUP BY 1, 2, 3 + ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_1h_ins" + AFTER INSERT ON "metrics_total_15m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_1h"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_1h_upd" + AFTER UPDATE ON "metrics_total_15m" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_1h"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_total_6h"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_total_6h needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_total_6h:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '6 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_total_6h" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '6 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_total_1h" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '6 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '6 hours' + GROUP BY 1, 2, 3 + ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_6h_ins" + AFTER INSERT ON "metrics_total_1h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_6h"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_6h_upd" + AFTER UPDATE ON "metrics_total_1h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_6h"(); + +CREATE OR REPLACE FUNCTION "sqlflow_rollup_metrics_total_1d"() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock serializes writers only because each statement in READ COMMITTED + -- takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'sqlflow rollup metrics_total_1d needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + IF current_setting('sqlflow.rollup_backfill', true) IS DISTINCT FROM 'on' THEN + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_total_1d:' || extract(epoch FROM touched.b)::bigint, 0)) + FROM (SELECT DISTINCT date_bin(INTERVAL '24 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed ORDER BY 1) AS touched; + END IF; + + INSERT INTO "metrics_total_1d" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") + SELECT date_bin(INTERVAL '24 hours', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision + FROM "metrics_total_6h" AS f + JOIN (SELECT DISTINCT date_bin(INTERVAL '24 hours', "bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00') AS b FROM changed) AS touched + ON f."bucket" >= touched.b AND f."bucket" < touched.b + INTERVAL '24 hours' + GROUP BY 1, 2, 3 + ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; + RETURN NULL; +END $fn$; +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_1d_ins" + AFTER INSERT ON "metrics_total_6h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_1d"(); +CREATE OR REPLACE TRIGGER "sqlflow_rollup_metrics_total_1d_upd" + AFTER UPDATE ON "metrics_total_6h" + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION "sqlflow_rollup_metrics_total_1d"(); + +-- Backfill the grains built from the source. Their inserts fire the triggers, +-- which fill every coarser grain once. +SET LOCAL sqlflow.rollup_backfill = on; +INSERT INTO "metrics_5m" ("bucket", "name", "type", "dimensions_key", "value_count", "value_last", "value_max", "value_min", "value_sum") +SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", f."dimensions_key", sum(f."value_count")::bigint, (array_agg(f."value_last" ORDER BY (f."value_last" IS NULL), f."bucket" DESC))[1], max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3, 4 +ON CONFLICT ("bucket", "name", "type", "dimensions_key") DO UPDATE SET "value_count" = excluded."value_count", "value_last" = excluded."value_last", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; +INSERT INTO "metrics_total_5m" ("bucket", "name", "type", "value_count", "value_max", "value_min", "value_sum") +SELECT date_bin(INTERVAL '5 minutes', f."bucket", TIMESTAMPTZ '2000-01-01 00:00:00+00'), f."name", f."type", sum(f."value_count")::bigint, max(f."value_max"), min(f."value_min"), sum(f."value_sum")::double precision +FROM "metrics_1m" AS f +GROUP BY 1, 2, 3 +ON CONFLICT ("bucket", "name", "type") DO UPDATE SET "value_count" = excluded."value_count", "value_max" = excluded."value_max", "value_min" = excluded."value_min", "value_sum" = excluded."value_sum"; +SET LOCAL sqlflow.rollup_backfill = off; diff --git a/render/migrations/0004_writers.sql b/render/migrations/0004_writers.sql new file mode 100644 index 00000000..2c67957e --- /dev/null +++ b/render/migrations/0004_writers.sql @@ -0,0 +1,102 @@ +-- What each pipeline process published, before the processes are merged. +-- +-- A pipeline process holds its own window in memory and publishes a closed +-- minute by replacing the row for its key. Replacing is right for one +-- process: a minute written twice, by a retry or a republish, is the same +-- minute. It is wrong across processes. Two instances behind a load balancer +-- each hold part of the 14:03 checkouts, and keyed on the series alone the +-- second to publish replaces the first: 7 and 5 stored as 5. Two instances +-- also run during every deploy, the old beside the new. +-- +-- So the key carries the writer. A process replaces only what it published +-- itself, and metrics_1m is the merge of every writer's row. The pipeline +-- writes here and never to metrics_1m. +-- +-- writer names one process from start to exit. bin/entrypoint.sh makes a new +-- one on every start, never a stable one: a restarted process has lost its +-- window, and what it publishes next for a minute must add to what the dead +-- process published for it, not replace it. +CREATE TABLE metrics_1m_writers ( + bucket TIMESTAMPTZ NOT NULL, + name TEXT NOT NULL, + type TEXT NOT NULL CHECK (type IN ('count', 'gauge')), + dimensions_key TEXT NOT NULL, + writer TEXT NOT NULL, + value_sum DOUBLE PRECISION NOT NULL, + value_count BIGINT NOT NULL, + value_min DOUBLE PRECISION NOT NULL, + value_max DOUBLE PRECISION NOT NULL, + value_last DOUBLE PRECISION NOT NULL, + last_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (bucket, name, type, dimensions_key, writer) +); + +-- Re-merges every minute of every series a statement touched, from all of its +-- writers' rows, in the writer's own transaction. The write into metrics_1m +-- fires the series trigger and the rollup triggers, so every grain follows. +-- +-- Sums add, min and max nest, and value_last is the value with the latest +-- last_at across writers. The writer breaks a tie so the answer is the same +-- on every re-merge. +CREATE FUNCTION metrics_1m_merge() RETURNS trigger +LANGUAGE plpgsql AS $fn$ +BEGIN + -- The lock below serializes writers only because each statement in READ + -- COMMITTED takes a new snapshot, which sees the writer the lock waited on. + IF current_setting('transaction_isolation') <> 'read committed' THEN + RAISE EXCEPTION 'metrics_1m_merge needs READ COMMITTED, not %', current_setting('transaction_isolation'); + END IF; + + -- One lock, taken before this transaction touches anything another writer + -- wants: metrics_1m, series, or a rollup table. + -- + -- It does two jobs. Without it two writers publishing one minute at once + -- each re-merge from a snapshot missing the other's uncommitted row, and the + -- later commit overwrites the merge with its own half. + -- + -- It is one lock and not a lock per minute because of what runs under it. + -- The write below fires the series trigger, which locks a series row, and + -- then the rollup triggers, which take a lock per bucket of each grain. An + -- upsert fires this function twice, once for the rows it inserted and once + -- for the rows it updated, so a transaction can hold a rollup lock from its + -- first pass and want a series row in its second, while another holds that + -- row and wants the rollup lock. Four concurrent writers deadlocked that + -- way within one round under a lock per minute. Behind one lock, writers + -- run the whole chain one at a time, and no cycle can form. + -- + -- It serializes publishes across processes. A publish is a few milliseconds + -- and a process publishes once every poll, ten seconds by default. + PERFORM pg_advisory_xact_lock(hashtextextended('metrics_1m_merge', 0)); + + INSERT INTO metrics_1m (bucket, name, type, dimensions_key, + value_sum, value_count, value_min, value_max, value_last, last_at) + SELECT f.bucket, f.name, f.type, f.dimensions_key, + sum(f.value_sum), + sum(f.value_count)::bigint, + min(f.value_min), + max(f.value_max), + (array_agg(f.value_last ORDER BY f.last_at DESC, f.writer))[1], + max(f.last_at) + FROM metrics_1m_writers AS f + JOIN (SELECT DISTINCT bucket, name, type, dimensions_key FROM changed) AS touched + USING (bucket, name, type, dimensions_key) + GROUP BY f.bucket, f.name, f.type, f.dimensions_key + ON CONFLICT (bucket, name, type, dimensions_key) DO UPDATE SET + value_sum = excluded.value_sum, + value_count = excluded.value_count, + value_min = excluded.value_min, + value_max = excluded.value_max, + value_last = excluded.value_last, + last_at = excluded.last_at; + RETURN NULL; +END $fn$; + +-- Two triggers: Postgres refuses a transition table on a trigger with more +-- than one event. An upsert fires both, each with its own rows. +CREATE TRIGGER metrics_1m_merge_ins AFTER INSERT ON metrics_1m_writers + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION metrics_1m_merge(); +CREATE TRIGGER metrics_1m_merge_upd AFTER UPDATE ON metrics_1m_writers + REFERENCING NEW TABLE AS changed + FOR EACH STATEMENT EXECUTE FUNCTION metrics_1m_merge(); diff --git a/render/migrations/0005_install.sql b/render/migrations/0005_install.sql new file mode 100644 index 00000000..ec76268d --- /dev/null +++ b/render/migrations/0005_install.sql @@ -0,0 +1,25 @@ +-- One row: which install this is, and which of its two telemetry events have +-- been sent. bin/telemetry.sh reads and writes it. Nothing else does. +-- +-- install_id names this database. It is made here, from nothing about the +-- deployer: not a hostname, not a Render id, not an address. It survives +-- restarts and redeploys, so an install is counted once however often it +-- starts. +-- +-- Each event has two columns. sent_at is set once our collector has answered +-- 2xx, and the event is never sent again. claimed_at is a lease: a process +-- takes it before it sends, so that two pipeline instances starting together +-- do not both report one install, and gives it back if the send fails. A +-- lease older than two minutes is up for grabs, so a process that died between +-- claiming and sending does not lose the event for good. +CREATE TABLE install ( + singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton), + install_id uuid NOT NULL DEFAULT gen_random_uuid(), + created_at timestamptz NOT NULL DEFAULT now(), + deployed_claimed_at timestamptz, + deployed_sent_at timestamptz, + first_request_claimed_at timestamptz, + first_request_sent_at timestamptz +); + +INSERT INTO install DEFAULT VALUES; diff --git a/render/pipeline.yml b/render/pipeline.yml new file mode 100644 index 00000000..a3ec7870 --- /dev/null +++ b/render/pipeline.yml @@ -0,0 +1,164 @@ +# Accepts metrics over a webhook, aggregates them in a 1-minute tumbling +# window, and upserts each closed minute into Postgres. +# +# SQLFLOW_POSTGRES_URI required +# SQLFLOW_WEBHOOK_AUTH hmac (default) or none. bin/entrypoint.sh +# refuses anything else, refuses hmac without +# a secret, and exports the mode it settled +# on. Anything but exactly none is signed. +# SQLFLOW_WEBHOOK_HMAC_SECRET required under hmac +# SQLFLOW_WEBHOOK_PORT default 8001 +# SQLFLOW_WEBHOOK_MAX_BODY_BYTES default 26214400, 25 MiB +# SQLFLOW_WRITER_ID required. bin/entrypoint.sh makes one per +# process start. Never set it by hand: two +# processes that share one replace each +# other's minutes. +# SQLFLOW_METRIC_NAME_PREFIX default empty. When set, a metric whose name +# does not start with it is dropped. +# +# The four window and flush variables exist so the test closes a minute in +# seconds. Leave them unset in a deploy. +commands: + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + # One request is one row. A request carries one metric in its top-level + # keys, or several under metrics. Every scalar is TEXT: StructuredBatch + # fails the whole batch on a value that does not fit its column, and one + # sender's "value": "abc" must not cost the others their metrics. The + # handler casts, and drops what does not cast. + - name: declare the request schema + sql: | + CREATE TABLE IF NOT EXISTS events ( + name TEXT, type TEXT, value TEXT, dimensions TEXT, timestamp TEXT, + metrics STRUCT(name TEXT, type TEXT, value TEXT, dimensions TEXT, timestamp TEXT)[] + ); + +tables: + sql: + - name: metrics_1m + # No unique index, on purpose: in-memory DuckDB never frees rows deleted + # from an indexed table, and the engine deletes every closed minute. Each + # batch appends its own rows and emit_sql merges them per minute. + # See turbolytics/sql-flow#268. + sql: | + CREATE TABLE IF NOT EXISTS metrics_1m ( + bucket TIMESTAMPTZ, name TEXT, type TEXT, dimensions_key TEXT, + value_sum DOUBLE, value_count BIGINT, value_min DOUBLE, value_max DOUBLE, + value_last DOUBLE, last_at TIMESTAMPTZ + ); + + # late_rows is drop, not reemit. The sink replaces a minute, it does not + # merge into one, so a late row republished alone would replace the + # minute's values with its own. sqlflow validate refuses reemit with an + # upserting sink. A metric whose timestamp is older than the open minutes + # is dropped and counted in window_late_rows_total. + window: + time_column: bucket + size_seconds: 60 + grace_seconds: {{ SQLFLOW_WINDOW_GRACE_SECONDS|default('60') }} + idle_close_seconds: {{ SQLFLOW_WINDOW_IDLE_SECONDS|default('60') }} + late_rows: drop + poll_interval_seconds: {{ SQLFLOW_WINDOW_POLL_SECONDS|default('10') }} + # value_last is the value with the latest last_at, not the last row + # appended: batches of one minute can arrive out of order. + emit_sql: | + SELECT bucket, name, type, dimensions_key, + '{{ SQLFLOW_WRITER_ID }}' AS writer, + sum(value_sum) AS value_sum, + sum(value_count)::BIGINT AS value_count, + min(value_min) AS value_min, + max(value_max) AS value_max, + arg_max(value_last, last_at) AS value_last, + max(last_at) AS last_at + FROM closed + GROUP BY bucket, name, type, dimensions_key + + sink: + # upsert on the primary key makes at-least-once delivery survivable: + # a minute this process writes twice replaces itself. updated_at is + # not in the batch, so a new row takes the column's DEFAULT now(). + # + # The key carries the writer, and the table is metrics_1m_writers, + # not metrics_1m. Another process holds its own part of the same + # minute, and keyed on the series alone the second to publish would + # replace the first. Postgres merges the writers into metrics_1m. + # See migrations/0004_writers.sql. + type: postgres + postgres: + dsn: "{{ SQLFLOW_POSTGRES_URI }}" + table: metrics_1m_writers + mode: upsert + key: [bucket, name, type, dimensions_key, writer] + +pipeline: + name: render-metrics + batch_size: 500 + # A lone curl is a partial batch. It waits this long, then the minute has to + # close, so a first metric is readable after one to two minutes. + flush_interval_seconds: {{ SQLFLOW_FLUSH_INTERVAL_SECONDS|default('5') }} + + source: + type: webhook + webhook: + addr: "0.0.0.0:{{ SQLFLOW_WEBHOOK_PORT|default('8001') }}" + max_body_bytes: {{ SQLFLOW_WEBHOOK_MAX_BODY_BYTES|default('26214400') }} +{# Signed unless the mode is exactly none. bin/entrypoint.sh exports hmac or + none and nothing else, and this test fails closed if it is ever run without + it: an unset, empty or misspelt mode gets the signature block. #} +{% if SQLFLOW_WEBHOOK_AUTH|default('hmac') != 'none' %} + signature_type: hmac + hmac: + header: X-Signature-256 + sig_key: sha256 + secret: "{{ SQLFLOW_WEBHOOK_HMAC_SECRET }}" +{% endif %} + + handler: + type: handlers.StructuredBatch + table: events + # A row is kept when every rule holds and dropped otherwise. The request + # was answered 200 before this ran, so nothing here can refuse one. + # + # dimensions_key is the series' identity: keys sorted, every value a + # string, compact JSON. This is its only author. Nothing reads a key back + # into this form. + # + # A timestamp more than a minute ahead is dropped. The window closes + # minutes against the newest event time it has seen, so one metric dated + # 2030 would close every minute until 2030 and drop all that follows as + # late. + # + # The column is ts, not at: AT is reserved in DuckDB. + sql: | + INSERT INTO metrics_1m + WITH raw AS ( + SELECT name, type, value, dimensions, timestamp FROM events WHERE metrics IS NULL + UNION ALL + SELECT m.name, m.type, m.value, m.dimensions, m.timestamp + FROM (SELECT unnest(metrics) AS m FROM events WHERE metrics IS NOT NULL) + ), + parsed AS ( + SELECT name, type, + CASE WHEN value IS NULL THEN 1.0 ELSE TRY_CAST(value AS DOUBLE) END AS value, + TRY(from_json(coalesce(dimensions, '{}'), '"MAP(VARCHAR, VARCHAR)"')) AS dims, + CASE WHEN timestamp IS NULL THEN now() ELSE TRY_CAST(timestamp AS TIMESTAMPTZ) END AS ts + FROM raw + ) + SELECT time_bucket(INTERVAL '1 minute', ts) AS bucket, name, type, + to_json(map_from_entries(list_sort(map_entries(dims))))::TEXT AS dimensions_key, + sum(value), count(*), min(value), max(value), arg_max(value, ts), max(ts) + FROM parsed + WHERE name IS NOT NULL AND name <> '' AND strlen(name) <= 200 + AND starts_with(name, '{{ SQLFLOW_METRIC_NAME_PREFIX|default('') }}') + AND type IN ('count', 'gauge') + AND value IS NOT NULL AND isfinite(value) + AND dims IS NOT NULL AND cardinality(dims) <= 16 + AND ts IS NOT NULL AND ts <= now() + INTERVAL '60 seconds' + GROUP BY ALL + + # The window's sink does the writing. A pipeline sink would write every + # batch, which is the opposite of aggregating. + sink: + type: noop diff --git a/render/rollups.yml b/render/rollups.yml new file mode 100644 index 00000000..6ef7e71f --- /dev/null +++ b/render/rollups.yml @@ -0,0 +1,49 @@ +# The rollups Postgres keeps from the pipeline's minute table. +# +# make rollups regenerates migrations/0003_rollups.sql +# +# CI runs `sqlflow rollup check`, which fails when the migration differs from +# what this file generates. +# +# There is no serve block. sqlflow rollup generates a dataset of at most one +# dimension, folded. A series here is three columns that cannot fold, so +# serve.yml's metric and metric_total grains are written by hand. +rollups: + - name: metrics + source: + table: metrics_1m + time_column: bucket + grain: 1m + dimensions: [name, type, dimensions_key] + + # Each grain a few times the one below it, so a re-merge reads few rows. + grains: + 5m: {from: 1m} + 15m: {from: 5m} + 1h: {from: 15m} + 6h: {from: 1h} + 1d: {from: 6h} + + dimension_sets: + - name: metrics + dimensions: [name, type, dimensions_key] + measures: + # numeric: double keeps the fraction. A sum is stored as a bigint by + # default, and Postgres rounds a fraction cast to one: a gauge + # reading of 0.73 would be stored as 1, and one of 0.25 as 0. + # value_count needs no double: a count of samples is a whole number. + value_sum: {type: sum, column: value_sum, numeric: double} + value_count: {type: sum, column: value_count} + value_min: {type: min, column: value_min} + value_max: {type: max, column: value_max} + # A gauge's headline value: queue depth now, not its sum. + value_last: {type: last, column: value_last} + # A name across every set of dimensions, for a name with a series per + # user. No last: two series share a bucket here, and neither is later. + - name: metrics_total + dimensions: [name, type] + measures: + value_sum: {type: sum, column: value_sum, numeric: double} + value_count: {type: sum, column: value_count} + value_min: {type: min, column: value_min} + value_max: {type: max, column: value_max} diff --git a/render/serve.yml b/render/serve.yml new file mode 100644 index 00000000..ba9aab26 --- /dev/null +++ b/render/serve.yml @@ -0,0 +1,241 @@ +# Serves what the pipeline writes, read-only, from the same Postgres. +# +# sqlflow serve -c serve.yml +# +# SQLFLOW_POSTGRES_URI required +# SQLFLOW_SERVE_CLIENT_ID required. Sent as ?client_id= on every request. The +# deploy asks for it. bin/serve.sh refuses one that +# would need encoding in a URL. +# SQLFLOW_SERVE_PORT default 8080 +# +# The metric and metric_total grains are written by hand. sqlflow rollup +# generates a dataset of at most one dimension, folded, and a series here is +# three columns that cannot fold. Every grain of a dataset is the same +# statement over another table. CI queries each one. +# +# Every grain holds an answer for the same 30 seconds. A coarse grain changes +# slowly and could be held for minutes, but the first thing a new deploy does +# is send one metric and ask for it, and an empty answer held for ten minutes +# reads as a pipeline that does not work. +commands: + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + - name: load postgres extension + sql: | + INSTALL postgres; + LOAD postgres; + + # One scan fans out across ctid ranges, each on its own Postgres connection. + # The default cap is 64. The Render plan allows far fewer. + - name: bound the postgres connections one scan may open + sql: | + SET pg_connection_limit = 4; + + # READ_ONLY is the guard. A write in a dataset's SQL fails at startup. + - name: attach postgres + sql: | + ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES, READ_ONLY); + +serve: + name: render-metrics-api + + http: + addr: "0.0.0.0:{{ SQLFLOW_SERVE_PORT|default('8080') }}" + + clients: + - name: default + id: "{{ SQLFLOW_SERVE_CLIENT_ID }}" + + pool: + size: 4 + + metrics: + enabled: true + + limits: + max_rows: 10000 + timeout_seconds: 10 + + datasets: + - name: series + description: Every series seen. A series is a name, a type and a set of dimensions. dimensions is a JSON object, as a string. + cache: {ttl_seconds: 15} + sql: | + SELECT name, type, dimensions, first_bucket, last_bucket + FROM pg.series + ORDER BY name, type, dimensions_key + + - name: metric + description: One metric per bucket per series. name is required; without it the answer is empty. dimensions is a JSON object, and the answer holds the series whose dimensions contain every pair in it; a dimensions that is not JSON matches nothing. since and until bound the range, default the last hour, and pick the grain. Read value_sum for a count and value_last for a gauge; value_sum / value_count is the average. + params: + - name: name + type: string + - name: dimensions + type: string + - name: since + type: timestamp + - name: until + type: timestamp + cache: + ttl_seconds: 30 + grains: + 1m: + bucket: 1m + max_range: 6h + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_1m AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + 5m: + bucket: 5m + max_range: 1d + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_5m AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + 15m: + bucket: 15m + max_range: 3d + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_15m AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + 1h: + bucket: 1h + max_range: 14d + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_1h AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + 6h: + bucket: 6h + max_range: 90d + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_6h AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + 1d: + bucket: 1d + max_range: 365d + sql: | + SELECT m.bucket, m.type, s.dimensions, + m.value_sum, m.value_count, m.value_min, m.value_max, m.value_last + FROM pg.metrics_1d AS m + JOIN pg.series AS s USING (name, type, dimensions_key) + WHERE m.name = $name AND s.name = $name + AND m.bucket >= $since AND m.bucket < $until + AND json_contains(s.dimensions::JSON, + CASE WHEN $dimensions IS NULL THEN '{}'::JSON + ELSE TRY_CAST($dimensions AS JSON) END) + ORDER BY m.bucket, m.type, m.dimensions_key + range: + since: since + until: until + default: 1h + + - name: metric_total + description: One metric per bucket, across every series of the name. The answer for a name with many series, such as one per user. name is required. It has no value_last, because the last value of many series is no series' value. + params: + - name: name + type: string + - name: since + type: timestamp + - name: until + type: timestamp + cache: + ttl_seconds: 30 + grains: + 1m: + bucket: 1m + max_range: 6h + sql: | + SELECT bucket, type, + sum(value_sum)::DOUBLE AS value_sum, + sum(value_count)::BIGINT AS value_count, + min(value_min) AS value_min, + max(value_max) AS value_max + FROM pg.metrics_1m + WHERE name = $name AND bucket >= $since AND bucket < $until + GROUP BY bucket, type + ORDER BY bucket, type + 5m: + bucket: 5m + max_range: 1d + sql: | + SELECT bucket, type, value_sum, value_count, value_min, value_max + FROM pg.metrics_total_5m + WHERE name = $name AND bucket >= $since AND bucket < $until + ORDER BY bucket, type + 15m: + bucket: 15m + max_range: 3d + sql: | + SELECT bucket, type, value_sum, value_count, value_min, value_max + FROM pg.metrics_total_15m + WHERE name = $name AND bucket >= $since AND bucket < $until + ORDER BY bucket, type + 1h: + bucket: 1h + max_range: 14d + sql: | + SELECT bucket, type, value_sum, value_count, value_min, value_max + FROM pg.metrics_total_1h + WHERE name = $name AND bucket >= $since AND bucket < $until + ORDER BY bucket, type + 6h: + bucket: 6h + max_range: 90d + sql: | + SELECT bucket, type, value_sum, value_count, value_min, value_max + FROM pg.metrics_total_6h + WHERE name = $name AND bucket >= $since AND bucket < $until + ORDER BY bucket, type + 1d: + bucket: 1d + max_range: 365d + sql: | + SELECT bucket, type, value_sum, value_count, value_min, value_max + FROM pg.metrics_total_1d + WHERE name = $name AND bucket >= $since AND bucket < $until + ORDER BY bucket, type + range: + since: since + until: until + default: 1h diff --git a/render/test/e2e.sh b/render/test/e2e.sh new file mode 100755 index 00000000..9df8159a --- /dev/null +++ b/render/test/e2e.sh @@ -0,0 +1,317 @@ +#!/usr/bin/env bash +# Brings the template up, posts metrics, and reads them back at every grain. +# Run from render/: test/e2e.sh. Needs docker, curl, jq and openssl. +set -euo pipefail +cd "$(dirname "$0")/.." + +INGEST="http://127.0.0.1:${INGEST_HOST_PORT:-10000}" +INGEST2="http://127.0.0.1:${INGEST2_HOST_PORT:-10001}" +API="http://127.0.0.1:${API_HOST_PORT:-8080}" +SECRET=local-secret +# The tag the image is built from, which an install reports as its version. +# make test passes it. Must match the Dockerfile's default. +: "${SQLFLOW_IMAGE:=turbolytics/sql-flow:v2026.09.18.1}" +export SQLFLOW_IMAGE +GRAINS="1m 5m 15m 1h 6h 1d" + +fail() { echo "FAIL: $*" >&2; docker compose logs --tail 40 ingest api >&2 || true; exit 1; } +cleanup() { docker compose down -v >/dev/null 2>&1 || true; } +trap cleanup EXIT + +# status [instance URL] +status() { + local want="$1" mode="$2" body="$3" url="${4:-$INGEST}" got sig + local args=(-s -o /dev/null -w '%{http_code}' -X POST "$url/events" --data-binary "$body") + if [ "$mode" = signed ]; then + sig="$(printf '%s' "$body" | openssl dgst -sha256 -hmac "$SECRET" -hex | sed 's/^.* //')" + args+=(-H "X-Signature-256: sha256=$sig") + fi + got="$(curl "${args[@]}")" + [ "$got" = "$want" ] || fail "POST ($mode) $body: status $got, want $want" +} + +# get [curl --data-urlencode args...] -> response JSON +get() { + local ds="$1"; shift + local args=(-s -G "$API/v1/datasets/$ds" --data-urlencode client_id=local-dev) + for kv in "$@"; do args+=(--data-urlencode "$kv"); done + curl "${args[@]}" +} + +# expect +expect() { + [ "$(jq -r "$2" <<<"$3")" = true ] || fail "$1: $(jq -c . <<<"$3")" + echo "ok $1" +} + +wait_for() { + local what="$1" url="$2" i + for i in $(seq 1 60); do + if curl -s -o /dev/null "$url"; then return 0; fi + sleep 1 + done + fail "$what did not come up" +} + +echo "== the entrypoint refuses a blank secret and an unknown auth mode" +cleanup +docker compose build -q +for env in "SQLFLOW_WEBHOOK_HMAC_SECRET=" "SQLFLOW_WEBHOOK_AUTH=open" "SQLFLOW_METRIC_NAME_PREFIX=a'b"; do + code=0 + docker compose run --rm -T --no-deps -e "$env" ingest >/dev/null 2>&1 || code=$? + [ "$code" = 2 ] || fail "ingest with $env exited $code, want 2" + echo "ok $env exits 2" +done + +echo "== an auth mode that is empty still requires a signature" +# The entrypoint and the template once disagreed about an empty mode, and an +# unsigned request was answered 200 by a pipeline with a secret set. +docker compose up -d --wait postgres +for env in "SQLFLOW_WEBHOOK_AUTH=" "SQLFLOW_WEBHOOK_AUTH=hmac"; do + # Telemetry off: these pipelines are killed as soon as they are probed. One + # killed between claiming install.deployed and giving the claim back leaves a + # two-minute lease behind, and the pipelines started below cannot send the + # event until it expires, which is longer than the test waits for it. + docker compose run -d --rm --no-deps -p 127.0.0.1:10077:10000 -e "$env" -e SQLFLOW_TELEMETRY=off ingest >/dev/null + wait_for "ingest with $env" "http://127.0.0.1:10077/events" + got="$(curl -s -o /dev/null -w '%{http_code}' -X POST http://127.0.0.1:10077/events --data-binary '{"name":"x","type":"count"}')" + docker ps -q --filter "publish=10077" | xargs -r docker rm -f >/dev/null + [ "$got" = 400 ] || fail "with $env an unsigned request answered $got, want 400" + echo "ok $env refuses an unsigned request" +done + +echo "== the API refuses a blank client id, and one a URL would mangle" +for env in "SQLFLOW_SERVE_CLIENT_ID=" "SQLFLOW_SERVE_CLIENT_ID=aA+SZt88/x="; do + code=0 + docker compose run --rm -T --no-deps -e "$env" api >/dev/null 2>&1 || code=$? + [ "$code" = 2 ] || fail "api with $env exited $code, want 2" + echo "ok $env exits 2" +done + +echo "== signed requests land" +docker compose up -d --wait postgres +# Whatever ran before this, the pipelines below start with no claim held. +docker compose exec -T postgres psql -U metrics -d metrics -X -Atqc "UPDATE install SET deployed_claimed_at = NULL, first_request_claimed_at = NULL" >/dev/null 2>&1 || true + +docker compose up -d collector ingest ingest2 api +wait_for ingest "$INGEST/events" +wait_for ingest2 "$INGEST2/events" +wait_for api "$API/healthz" + +# install.deployed went out when the pipelines started. install.first_request +# goes out once the metrics below land, and under this stack's five-second idle +# bound the collector may already have closed the minute the first was sent +# in. Start in the next one, so the second is not dropped as late. +sleep $(( 61 - 10#$(date -u +%S) )) + +# One timestamp for every metric. Without it the posts could straddle a +# minute, and the 1m assertions below would see two buckets. +TS="$(date -u +%Y-%m-%dT%H:%M:%SZ)" + +echo "== the pipeline answers a health check with no signature, and nothing else" +# render.yaml's healthCheckPath, and what an uptime monitor calls. Signatures +# are on here: the check must not need one, and must not open anything. +[ "$(curl -s "$INGEST/healthz")" = '{"status":"ok"}' ] || fail "GET /healthz did not answer {\"status\":\"ok\"}" +[ "$(curl -s -o /dev/null -w '%{http_code}' -I "$INGEST/healthz")" = 200 ] || fail "HEAD /healthz did not answer 200" +[ "$(curl -s -o /dev/null -w '%{http_code}' -X POST "$INGEST/healthz")" = 405 ] || fail "POST /healthz did not answer 405" +echo "ok GET and HEAD /healthz answer 200 unsigned, and POST is refused" + +status 400 unsigned '{"name":"unsigned","type":"count"}' +status 200 signed '{"name":"checkout","type":"count","timestamp":"'"$TS"'","dimensions":{"region":"us-east","plan":"pro"}}' +status 200 signed '{"name":"checkout","type":"count","value":2,"timestamp":"'"$TS"'","dimensions":{"plan":"pro","region":"us-east"}}' +status 200 signed '{"name":"checkout","type":"count","value":5,"timestamp":"'"$TS"'","dimensions":{"region":"us-east","plan":"free"}}' +status 200 signed '{"name":"checkout","type":"count","timestamp":"'"$TS"'","dimensions":{"region":"eu","plan":"pro"}}' +status 200 signed '{"name":"quote","type":"count","timestamp":"'"$TS"'","dimensions":{"q":"say \"hi\""}}' +status 200 signed '{"metrics":[{"name":"cpu","type":"gauge","value":0.25,"timestamp":"'"$TS"'"},{"name":"cpu","type":"gauge","value":0.5,"timestamp":"'"$TS"'"}]}' +# One minute of one series, split across two pipeline instances. Each holds +# its own window and publishes its own part. Keyed on the series alone, the +# second to publish replaced the first, and 7 and 5 were stored as one of +# them. The gauge's later reading goes to the second instance, so value_last +# must be chosen by event time across writers, not by who published last. +MIN="${TS%:*}" +status 200 signed '{"name":"split","type":"count","value":7,"timestamp":"'"$TS"'"}' "$INGEST" +status 200 signed '{"name":"split","type":"count","value":5,"timestamp":"'"$TS"'"}' "$INGEST2" +status 200 signed '{"name":"split.gauge","type":"gauge","value":9,"timestamp":"'"$MIN"':20Z"}' "$INGEST2" +status 200 signed '{"name":"split.gauge","type":"gauge","value":1,"timestamp":"'"$MIN"':10Z"}' "$INGEST" + +# Each is answered 200 and stores nothing. +status 200 signed '{"name":"bad-type","type":"histogram"}' +status 200 signed '{"name":"bad-value","type":"count","value":"abc"}' +status 200 signed '{"name":"bad-dims","type":"count","dimensions":[1,2]}' +status 200 signed '{"name":"future","type":"count","timestamp":"2030-01-01T00:00:00Z"}' +status 200 signed '{"name":"bad-ts","type":"count","timestamp":"yesterday"}' +status 200 signed '[{"name":"bare-array","type":"count"}]' +status 200 signed 'not json' + +echo "== waiting for the minute to close" +for i in $(seq 1 60); do + n="$(get series | jq '[.rows[] | select(.name | startswith("install.") | not)] | length')" + [ "$n" -ge 7 ] && break + sleep 1 +done +# Both instances must have published before the split minute is read. +for i in $(seq 1 60); do + w="$(docker compose exec -T postgres psql -U metrics -d metrics -Atc "SELECT count(DISTINCT writer) FROM metrics_1m_writers WHERE name = 'split'")" + [ "$w" = 2 ] && break + sleep 1 +done +[ "$w" = 2 ] || fail "the split minute has $w writers, want 2: both instances must publish it" +echo "ok two writers published the split minute" + +series="$(get series)" +expect "series holds the seven that were valid, and no bad one" \ + '[.rows[].name | select(startswith("install.") | not)] | sort == ["checkout","checkout","checkout","cpu","quote","split","split.gauge"]' "$series" +expect "a dimension value with a quote in it survives" \ + '[.rows[] | select(.name == "quote") | .dimensions | fromjson | .q] == ["say \"hi\""]' "$series" + +# A pinned grain needs a range at least as wide as its bucket. serve snaps the +# range to bucket boundaries, and a bucket is in the range only when its start +# is, so the default range of one hour holds no whole 6h or 1d bucket and +# answers empty. Each grain is asked for a range just inside its max_range. +since_for() { + local hours epoch + case "$1" in 1m) hours=5 ;; 5m) hours=23 ;; 15m) hours=71 ;; 1h) hours=335 ;; 6h) hours=2159 ;; 1d) hours=8735 ;; esac + epoch=$(( $(date -u +%s) - hours * 3600 )) + # GNU date reads @epoch, BSD date reads -r epoch. + date -u -d "@$epoch" +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || date -u -r "$epoch" +%Y-%m-%dT%H:%M:%SZ +} + +for g in $GRAINS; do + echo "== grain $g" + since="since=$(since_for "$g")" + all="$(get metric name=checkout grain="$g" "$since")" + expect "$g: three checkout series" '.grain == "'"$g"'" and (.rows | length) == 3' "$all" + expect "$g: two key orders were one series, summed" \ + '[.rows[] | select((.dimensions | fromjson) == {"plan":"pro","region":"us-east"}) | [.value_sum, .value_count, .value_min, .value_max]] == [[3,2,1,2]]' "$all" + + one="$(get metric name=checkout grain="$g" "$since" 'dimensions={"region":"us-east"}')" + expect "$g: one pair returns the two series that contain it" '(.rows | length) == 2' "$one" + both="$(get metric name=checkout grain="$g" "$since" 'dimensions={"region":"us-east","plan":"free"}')" + expect "$g: both pairs return one series" '[.rows[].value_sum] == [5]' "$both" + none="$(get metric name=checkout grain="$g" "$since" 'dimensions={"region":"us"}')" + expect "$g: a value that is a prefix of another matches nothing" '(.rows | type) == "array" and (.rows | length) == 0' "$none" + bad="$(get metric name=checkout grain="$g" "$since" 'dimensions={not json')" + expect "$g: a filter that is not JSON matches nothing, and is not an error" '(.rows | type) == "array" and (.rows | length) == 0' "$bad" + + cpu="$(get metric name=cpu grain="$g" "$since")" + expect "$g: a gauge keeps its fraction" \ + '[.rows[] | [.value_sum, .value_count, .value_min, .value_max]] == [[0.75,2,0.25,0.5]]' "$cpu" + + split="$(get metric name=split grain="$g" "$since")" + expect "$g: a minute split across two instances is their sum, not the last to publish" \ + '[.rows[] | [.value_sum, .value_count, .value_min, .value_max]] == [[12,2,5,7]]' "$split" + splitg="$(get metric name=split.gauge grain="$g" "$since")" + expect "$g: across two instances value_last is the later reading" \ + '[.rows[] | [.value_last, .value_min, .value_max]] == [[9,1,9]]' "$splitg" + + total="$(get metric_total name=checkout grain="$g" "$since")" + expect "$g: the total sums every series" \ + '[.rows[] | [.value_sum, .value_count, .value_min, .value_max]] == [[9,4,1,5]]' "$total" +done + +narrow="$(get metric name=checkout grain=1d)" +expect "a pinned 1d grain with the default hour is an empty range, not an error" \ + '(.range.since == .range.until) and (.rows | type) == "array" and (.rows | length) == 0' "$narrow" +auto="$(get metric name=checkout)" +expect "without a grain the API picks the finest that covers the range" '.grain == "1m" and (.rows | length) == 3' "$auto" + +# What a reader does with the id they chose: paste it into a URL as it is. +pasted="$(curl -s -o /dev/null -w '%{http_code}' "$API/v1/datasets/series?client_id=local-dev")" +[ "$pasted" = 200 ] || fail "a client id pasted into the URL answered $pasted" +echo "ok the client id works pasted into a URL, unencoded" +wrong="$(curl -s -o /dev/null -w '%{http_code}' "$API/v1/datasets/series?client_id=not-the-id")" +[ "$wrong" = 401 ] || fail "a wrong client id answered $wrong, want 401" +echo "ok a wrong client id is refused" + +noname="$(get metric)" +expect "no name answers empty" '(.rows | type) == "array" and (.rows | length) == 0' "$noname" + +echo "== telemetry: two events, once each, to the local collector and nowhere else" +psqlc() { docker compose exec -T postgres psql -U metrics -d metrics -X -Atc "$1"; } +install_id="$(psqlc 'SELECT install_id FROM install')" +for i in $(seq 1 60); do + n="$(get series | jq '[.rows[] | select(.name == "install.deployed" or .name == "install.first_request")] | length')" + [ "$n" -ge 2 ] && break + sleep 1 +done +series="$(get series)" +expect "one install.deployed and one install.first_request, from two pipeline instances" \ + '[.rows[] | select(.name | startswith("install.")) | .name] | sort == ["install.deployed","install.first_request"]' "$series" +expect "the event names this database and nothing about who deployed it" \ + '[.rows[] | select(.name == "install.deployed") | .dimensions | fromjson] == [{"install_id":"'"$install_id"'","source":"render","template":"render-metrics","sqlflow_version":"'"${SQLFLOW_IMAGE##*:}"'"}]' "$series" +for name in install.deployed install.first_request; do + expect "$name was sent once, though two instances started together" \ + '[.rows[].value_sum] == [1]' "$(get metric_total name="$name")" +done +[ "$(psqlc 'SELECT deployed_sent_at IS NOT NULL AND first_request_sent_at IS NOT NULL FROM install')" = t ] \ + || fail "install does not record both events as sent" +echo "ok both events are recorded as sent" + +docker compose restart ingest ingest2 >/dev/null +wait_for ingest "$INGEST/events" +sleep 8 +for name in install.deployed install.first_request; do + expect "$name is still one after both instances restarted" \ + '[.rows[].value_sum] | add == 1' "$(get metric_total name="$name" "since=$(since_for 1h)" grain=1h)" +done + +# What off and a dead collector do, each from a clean slate. +reset_install() { psqlc 'UPDATE install SET deployed_claimed_at = NULL, deployed_sent_at = NULL, first_request_claimed_at = NULL, first_request_sent_at = NULL' >/dev/null; } +run_probe() { # run_probe