From f39c90016f34572a61ec548a841a06b7607fbb7d Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Mon, 5 Oct 2026 05:19:19 -0400 Subject: [PATCH 1/2] turbostats: the database section, for dbhealth wire.Database and its sub-types, as the spec (#441) defines them: probe, resources, tables, replication and the collection cost. A failed probe marshals probe alone; the field is optional on the bundle beside Pipeline and Serve. bundle.schema.json regenerated. Tests: the spec's example round-trips with the spec's field names; a failed probe carries no resources, tables or replication; a bundle with only a database section marshals it and omits it when nil. --- CHANGELOG.md | 7 + turbostats/wire/bundle.go | 3 + turbostats/wire/database.go | 112 +++++++++++ turbostats/wire/database_test.go | 72 +++++++ turbostats/wire/schema/bundle.schema.json | 221 ++++++++++++++++++++++ 5 files changed, 415 insertions(+) create mode 100644 turbostats/wire/database.go create mode 100644 turbostats/wire/database_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 668f7577..145b2c08 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ ### Added +- **TurboStats `database` section.** The section a `dbhealth` reporter + sends: the probe (serving, and how fast), resources (connections against + the maximum, size on disk, the oldest open transaction), per-table + freshness and row counts, and replication, for one database endpoint. + Facts only; the receiver judges. `wire.Database`, optional on the bundle + beside `Pipeline` and `Serve`. A failed probe marshals `probe` alone. The + spec is `docs/superpowers/specs/2026-10-05-turbostats-database-section-design.md`. (#441) - **Webhook source `ack: after_flush`.** A 200 means the body reached the sink: the answer waits for the batch holding it to flush and commit, and a failed flush answers 503. The default, `on_receive`, is unchanged. Refused diff --git a/turbostats/wire/bundle.go b/turbostats/wire/bundle.go index d7fc0fc2..b398c3dd 100644 --- a/turbostats/wire/bundle.go +++ b/turbostats/wire/bundle.go @@ -59,6 +59,9 @@ type Bundle struct { Pipeline *Pipeline `json:"pipeline,omitempty"` // Serve is present when the process answers dataset requests. Serve *Serve `json:"serve,omitempty"` + // Database is present when the process watches a database endpoint, the + // dbhealth reporter. A bundle carries it and no Pipeline. + Database *Database `json:"database,omitempty"` // Freshness is present when the process measures a store's tables: the // rollup daemon's leader. Freshness *Freshness `json:"freshness,omitempty"` diff --git a/turbostats/wire/database.go b/turbostats/wire/database.go new file mode 100644 index 00000000..785a5bfa --- /dev/null +++ b/turbostats/wire/database.go @@ -0,0 +1,112 @@ +package wire + +import "time" + +// Database is the section a dbhealth reporter sends: one database endpoint, +// probed and measured. Facts only: timestamps, counts and limits. The +// receiver judges them. Fields a kind cannot provide are absent, not zero. +// +// Spec: docs/superpowers/specs/2026-10-05-turbostats-database-section-design.md. +type Database struct { + // Kind is postgres, mysql, mongo, snowflake or redshift. + Kind string `json:"kind"` + // Target is host:port/database. Never a DSN: it carries no credential. + Target string `json:"target"` + // Cluster groups a primary with its replicas. Absent when not configured. + Cluster string `json:"cluster,omitempty"` + ServerVersion string `json:"server_version,omitempty"` + + // Probe is always present. When it failed, nothing below it is. + Probe DatabaseProbe `json:"probe"` + + Resources *DatabaseResources `json:"resources,omitempty"` + Tables []DatabaseTable `json:"tables,omitempty"` + Replication *DatabaseReplication `json:"replication,omitempty"` + Collection DatabaseCollection `json:"collection"` +} + +// DatabaseProbe is one round trip per interval: is it serving, and how fast. +type DatabaseProbe struct { + OK bool `json:"ok"` + LatencyMs int64 `json:"latency_ms"` + LastOKAt *time.Time `json:"last_ok_at,omitempty"` + // ConsecutiveFailures counts probes failed in a row, 0 when OK. + ConsecutiveFailures int `json:"consecutive_failures"` + // Error is the class of the latest failure: refused, timeout, auth or + // other. Empty when OK. + Error string `json:"error,omitempty"` +} + +// DatabaseResources is how close the database is to its limits. Each field +// is absent when the kind, or the role, cannot read it. +type DatabaseResources struct { + Connections *DatabaseConnections `json:"connections,omitempty"` + // SizeBytes is this database on disk. + SizeBytes *int64 `json:"size_bytes,omitempty"` + // OldestTransactionSeconds is the age of the oldest open transaction; + // one idle in transaction holds vacuum back. + OldestTransactionSeconds *int64 `json:"oldest_transaction_seconds,omitempty"` + Memory *DatabaseMemory `json:"memory,omitempty"` +} + +// DatabaseConnections is connections in use against the configured maximum. +type DatabaseConnections struct { + Used int `json:"used"` + Max int `json:"max"` + Waiting int `json:"waiting"` +} + +// DatabaseMemory carries settings, not usage: a database does not know the +// host's memory from SQL. +type DatabaseMemory struct { + SharedBuffersBytes int64 `json:"shared_buffers_bytes"` +} + +// DatabaseTable is one watched table's freshness, size and row count. +type DatabaseTable struct { + Name string `json:"name"` + FreshnessColumn string `json:"freshness_column,omitempty"` + // NewestAt is max(FreshnessColumn). Absent when the table has none. + NewestAt *time.Time `json:"newest_at,omitempty"` + // Rows is an estimate unless RowsExact, which says count(*) ran. + Rows *int64 `json:"rows,omitempty"` + RowsExact bool `json:"rows_exact"` + SizeBytes int64 `json:"size_bytes"` + LastVacuumAt *time.Time `json:"last_vacuum_at,omitempty"` + CheckedAt time.Time `json:"checked_at"` +} + +// DatabaseReplication is measured from this endpoint's side. A replica +// reports how far behind what it serves is; a primary reports what it knows +// about each replica. +type DatabaseReplication struct { + // Role is primary or replica. + Role string `json:"role"` + LagSeconds *float64 `json:"lag_seconds,omitempty"` + LastReplayedAt *time.Time `json:"last_replayed_at,omitempty"` + Upstream string `json:"upstream,omitempty"` + // Replicas is a primary's view of each replica. + Replicas []DatabaseReplica `json:"replicas,omitempty"` +} + +// DatabaseReplica is one replica as its primary sees it. +type DatabaseReplica struct { + Name string `json:"name"` + LagSeconds *float64 `json:"lag_seconds,omitempty"` + State string `json:"state,omitempty"` +} + +// DatabaseCollection is what the interval cost, so the receiver can show +// the price of watching, and every per-table failure, so one table the +// role cannot read does not hide the others. +type DatabaseCollection struct { + Queries int `json:"queries"` + DurationMs int64 `json:"duration_ms"` + Errors []DatabaseError `json:"errors"` +} + +// DatabaseError is one query that failed this interval. +type DatabaseError struct { + Table string `json:"table,omitempty"` + Error string `json:"error"` +} diff --git a/turbostats/wire/database_test.go b/turbostats/wire/database_test.go new file mode 100644 index 00000000..4efe0745 --- /dev/null +++ b/turbostats/wire/database_test.go @@ -0,0 +1,72 @@ +package wire + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/turbolytics/sql-flow/internal/coverage" + "github.com/zeebo/assert" +) + +// The spec's example, byte for byte after a round trip: every field the +// section defines is here, and nothing a kind cannot provide is forced to +// zero. +func TestDatabase_SpecExampleRoundTrips(t *testing.T) { + coverage.Covers(t, "observability.turbostats") + at := time.Date(2026, 10, 5, 11, 59, 59, 0, time.UTC) + d := Database{ + Kind: "postgres", Target: "pg-2.internal:5432/billing", Cluster: "billing", ServerVersion: "18.0", + Probe: DatabaseProbe{OK: true, LatencyMs: 3, LastOKAt: &at}, + Resources: &DatabaseResources{ + Connections: &DatabaseConnections{Used: 18, Max: 100, Waiting: 0}, + SizeBytes: dbPtr(int64(14104255)), + OldestTransactionSeconds: dbPtr(int64(2)), + Memory: &DatabaseMemory{SharedBuffersBytes: 134217728}, + }, + Tables: []DatabaseTable{{ + Name: "public.usage_per_minute", FreshnessColumn: "minute", NewestAt: &at, + Rows: dbPtr(int64(7832)), RowsExact: false, SizeBytes: 1826816, LastVacuumAt: &at, CheckedAt: at, + }}, + Replication: &DatabaseReplication{Role: "replica", LagSeconds: dbPtr(0.4), LastReplayedAt: &at, Upstream: "pg-1.internal:5432"}, + Collection: DatabaseCollection{Queries: 9, DurationMs: 41, Errors: []DatabaseError{}}, + } + b, err := json.Marshal(d) + assert.NoError(t, err) + var back Database + assert.NoError(t, json.Unmarshal(b, &back)) + assert.Equal(t, d, back) + // Field names are the spec's. + for _, want := range []string{`"kind":"postgres"`, `"latency_ms":3`, `"rows_exact":false`, `"lag_seconds":0.4`, `"duration_ms":41`, `"size_bytes":14104255`} { + assert.That(t, strings.Contains(string(b), want)) + } +} + +// A failed probe sends probe alone: nothing else could run. +func TestDatabase_FailedProbeCarriesProbeAlone(t *testing.T) { + coverage.Covers(t, "observability.turbostats") + d := Database{Kind: "postgres", Target: "pg:5432/x", Probe: DatabaseProbe{OK: false, ConsecutiveFailures: 3, Error: "refused"}} + b, err := json.Marshal(d) + assert.NoError(t, err) + for _, absent := range []string{`"resources"`, `"tables"`, `"replication"`} { + assert.That(t, !strings.Contains(string(b), absent)) + } + assert.That(t, strings.Contains(string(b), `"error":"refused"`)) + assert.That(t, strings.Contains(string(b), `"consecutive_failures":3`)) +} + +// A bundle with a database section and no pipeline is the shape dbhealth +// sends; the field sits beside Pipeline and Serve, omitted when nil. +func TestBundle_DatabaseSectionIsOptional(t *testing.T) { + coverage.Covers(t, "observability.turbostats") + b, err := json.Marshal(Bundle{V: 1, Database: &Database{Kind: "postgres", Target: "pg:5432/x", Probe: DatabaseProbe{OK: true}}}) + assert.NoError(t, err) + assert.That(t, strings.Contains(string(b), `"database":{`)) + assert.That(t, !strings.Contains(string(b), `"pipeline"`)) + b, err = json.Marshal(Bundle{V: 1}) + assert.NoError(t, err) + assert.That(t, !strings.Contains(string(b), `"database"`)) +} + +func dbPtr[T any](v T) *T { return &v } diff --git a/turbostats/wire/schema/bundle.schema.json b/turbostats/wire/schema/bundle.schema.json index c3403a68..8445e705 100644 --- a/turbostats/wire/schema/bundle.schema.json +++ b/turbostats/wire/schema/bundle.schema.json @@ -529,6 +529,227 @@ ], "description": "Serve is present when the process answers dataset requests." }, + "database": { + "properties": { + "kind": { + "type": "string", + "description": "Kind is postgres, mysql, mongo, snowflake or redshift." + }, + "target": { + "type": "string", + "description": "Target is host:port/database. Never a DSN: it carries no credential." + }, + "cluster": { + "type": "string", + "description": "Cluster groups a primary with its replicas. Absent when not configured." + }, + "server_version": { + "type": "string" + }, + "probe": { + "properties": { + "ok": { + "type": "boolean" + }, + "latency_ms": { + "type": "integer" + }, + "last_ok_at": { + "type": "string", + "format": "date-time" + }, + "consecutive_failures": { + "type": "integer", + "description": "ConsecutiveFailures counts probes failed in a row, 0 when OK." + }, + "error": { + "type": "string", + "description": "Error is the class of the latest failure: refused, timeout, auth or\nother. Empty when OK." + } + }, + "type": "object", + "required": [ + "ok", + "latency_ms", + "consecutive_failures" + ], + "description": "Probe is always present. When it failed, nothing below it is." + }, + "resources": { + "properties": { + "connections": { + "properties": { + "used": { + "type": "integer" + }, + "max": { + "type": "integer" + }, + "waiting": { + "type": "integer" + } + }, + "type": "object", + "required": [ + "used", + "max", + "waiting" + ] + }, + "size_bytes": { + "type": "integer", + "description": "SizeBytes is this database on disk." + }, + "oldest_transaction_seconds": { + "type": "integer", + "description": "OldestTransactionSeconds is the age of the oldest open transaction;\none idle in transaction holds vacuum back." + }, + "memory": { + "properties": { + "shared_buffers_bytes": { + "type": "integer" + } + }, + "type": "object", + "required": [ + "shared_buffers_bytes" + ] + } + }, + "type": "object" + }, + "tables": { + "items": { + "properties": { + "name": { + "type": "string" + }, + "freshness_column": { + "type": "string" + }, + "newest_at": { + "type": "string", + "format": "date-time", + "description": "NewestAt is max(FreshnessColumn). Absent when the table has none." + }, + "rows": { + "type": "integer", + "description": "Rows is an estimate unless RowsExact, which says count(*) ran." + }, + "rows_exact": { + "type": "boolean" + }, + "size_bytes": { + "type": "integer" + }, + "last_vacuum_at": { + "type": "string", + "format": "date-time" + }, + "checked_at": { + "type": "string", + "format": "date-time" + } + }, + "type": "object", + "required": [ + "name", + "rows_exact", + "size_bytes", + "checked_at" + ], + "description": "DatabaseTable is one watched table's freshness, size and row count." + }, + "type": "array" + }, + "replication": { + "properties": { + "role": { + "type": "string", + "description": "Role is primary or replica." + }, + "lag_seconds": { + "type": "number" + }, + "last_replayed_at": { + "type": "string", + "format": "date-time" + }, + "upstream": { + "type": "string" + }, + "replicas": { + "items": { + "properties": { + "name": { + "type": "string" + }, + "lag_seconds": { + "type": "number" + }, + "state": { + "type": "string" + } + }, + "type": "object", + "required": [ + "name" + ], + "description": "DatabaseReplica is one replica as its primary sees it." + }, + "type": "array", + "description": "Replicas is a primary's view of each replica." + } + }, + "type": "object", + "required": [ + "role" + ] + }, + "collection": { + "properties": { + "queries": { + "type": "integer" + }, + "duration_ms": { + "type": "integer" + }, + "errors": { + "items": { + "properties": { + "table": { + "type": "string" + }, + "error": { + "type": "string" + } + }, + "type": "object", + "required": [ + "error" + ], + "description": "DatabaseError is one query that failed this interval." + }, + "type": "array" + } + }, + "type": "object", + "required": [ + "queries", + "duration_ms", + "errors" + ] + } + }, + "type": "object", + "required": [ + "kind", + "target", + "probe", + "collection" + ], + "description": "Database is present when the process watches a database endpoint, the\ndbhealth reporter. A bundle carries it and no Pipeline." + }, "freshness": { "properties": { "store_id": { From 43398bf7c9f9922b1839cf65b957e445f6aeefc3 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Mon, 5 Oct 2026 05:40:57 -0400 Subject: [PATCH 2/2] turbostats: bound the database section's three lists, and price the widest bundle TestWire_NoFieldScalesWithCardinality failed in CI: Database.Tables, Replication.Replicas and Collection.Errors are slices with no stated bound. None grows with the data in the database: tables are the operator's list, replicas are the cluster's topology, errors are one per table and one per catalog query. The contract now carries the caps, wire.MaxDatabaseTables 50, MaxDatabaseReplicas 16, MaxDatabaseErrors 60, and the shape guard exempts the three by name. TestCollect_AFullDatabaseBundleStaysUnderTheCeiling prices the widest bundle at those caps: 47725 bytes, under a 48 KiB ceiling. That is three times the 16 KiB a pipeline bundle's receiver allows, so the receiver that accepts database bundles needs 64 KiB; that lands with control's kind change. --- internal/turbostats/dimensional_test.go | 82 +++++++++++++++++++++++++ turbostats/wire/database.go | 16 +++++ 2 files changed, 98 insertions(+) diff --git a/internal/turbostats/dimensional_test.go b/internal/turbostats/dimensional_test.go index e4d4f5cf..1670301d 100644 --- a/internal/turbostats/dimensional_test.go +++ b/internal/turbostats/dimensional_test.go @@ -403,6 +403,16 @@ func TestWire_NoFieldScalesWithCardinality(t *testing.T) { if at == ".Bundle.Freshness.Tables" || at == ".Bundle.Rollup.Rollups" { continue } + // A database bundle's tables are the operator's list, capped + // by wire.MaxDatabaseTables; its replicas are the cluster's + // topology, capped by wire.MaxDatabaseReplicas; its errors + // are at most one per table and one per catalog query, + // wire.MaxDatabaseErrors. None grows with the data in the + // database. TestCollect_AFullDatabaseBundleStaysUnderTheCeiling + // prices all three at their caps. + if at == ".Bundle.Database.Tables" || at == ".Bundle.Database.Replication.Replicas" || at == ".Bundle.Database.Collection.Errors" { + continue + } // A duration's buckets are a fixed-length array whose length // the contract sets: len(wire.DurationBounds)+1, the same // for every process forever. It is the one shape a receiver @@ -565,6 +575,10 @@ func widestBundle(t *testing.T) wire.Bundle { }, Duration: &wire.ServeDurations{Request: widestDuration()}, }, + // One of each repeated field: enough for the schema to validate + // the shape. The full width is priced by + // TestCollect_AFullDatabaseBundleStaysUnderTheCeiling. + Database: widestDatabase(1, 1, 1), Exit: &wire.Exit{ Reason: "system.internal.unexpected", Code: 255, @@ -572,6 +586,74 @@ func widestBundle(t *testing.T) wire.Bundle { } } +// widestDatabase is a database section with every field at its widest +// value: the longest identifiers Postgres allows, the longest host name DNS +// allows, and 2^62 in every count, with the given number of tables, +// replicas and errors. +func widestDatabase(tables, replicas, errors int) *wire.Database { + big := int64(1) << 62 + n := 1 << 30 + at := time.Now().UTC() + secs := 1e9 + ident := strings.Repeat("i", 63) + table := strings.Repeat("s", 63) + "." + strings.Repeat("t", 63) + host := strings.Repeat("h", 253) + d := &wire.Database{ + Kind: "postgres", + Target: host + ":65535/" + ident, + Cluster: strings.Repeat("c", 64), + ServerVersion: "PostgreSQL 18.1 (Debian 18.1-1.pgdg13+1) on aarch64-unknown-linux-gnu, compiled by gcc (Debian 14.2.0-19) 14.2.0, 64-bit", + Probe: wire.DatabaseProbe{OK: true, LatencyMs: big, LastOKAt: &at, ConsecutiveFailures: n, Error: "timeout"}, + Resources: &wire.DatabaseResources{ + Connections: &wire.DatabaseConnections{Used: n, Max: n, Waiting: n}, + SizeBytes: &big, + OldestTransactionSeconds: &big, + Memory: &wire.DatabaseMemory{SharedBuffersBytes: big}, + }, + Replication: &wire.DatabaseReplication{Role: "primary", LagSeconds: &secs, LastReplayedAt: &at, Upstream: host + ":65535"}, + Collection: wire.DatabaseCollection{Queries: n, DurationMs: big}, + } + for i := 0; i < tables; i++ { + d.Tables = append(d.Tables, wire.DatabaseTable{ + Name: table, FreshnessColumn: ident, NewestAt: &at, Rows: &big, RowsExact: true, + SizeBytes: big, LastVacuumAt: &at, CheckedAt: at, + }) + } + for i := 0; i < replicas; i++ { + d.Replication.Replicas = append(d.Replication.Replicas, wire.DatabaseReplica{Name: ident, LagSeconds: &secs, State: "streaming"}) + } + for i := 0; i < errors; i++ { + d.Collection.Errors = append(d.Collection.Errors, wire.DatabaseError{Table: table, Error: strings.Repeat("e", 200)}) + } + return d +} + +// The widest database bundle the contract allows stays under 48 KiB: +// wire.MaxDatabaseTables tables, each with the longest schema and table +// name Postgres allows, wire.MaxDatabaseReplicas replicas, an error for +// every table and every catalog query, and every count at 2^62. It +// measured 47725 bytes on 2026-10-05. The 16 KiB limit that holds every +// pipeline bundle cannot hold it: a receiver that accepts database bundles +// allows 64 KiB, so 48 leaves the alarm a margin without letting the shape +// double quietly. +func TestCollect_AFullDatabaseBundleStaysUnderTheCeiling(t *testing.T) { + coverage.Covers(t, "observability.turbostats") + at := time.Now().UTC() + b := wire.Bundle{ + V: wire.Version, SentAt: at, IntervalSeconds: 86400, LastActivityAt: &at, + Instance: wire.Instance{ + ID: strings.Repeat("i", 64), Name: strings.Repeat("n", 64), Version: "v2026.10.05", Commit: strings.Repeat("c", 40), + Arch: "linux/arm64", ConfigHash: "sha256:" + strings.Repeat("f", 64), Labels: widestLabels(t), + }, + Process: wire.Process{StartedAt: at, RSSBytes: 1 << 62, Goroutines: 1 << 30}, + Database: widestDatabase(wire.MaxDatabaseTables, wire.MaxDatabaseReplicas, wire.MaxDatabaseErrors), + } + raw, err := json.Marshal(b) + assert.NoError(t, err) + t.Logf("the widest database bundle is %d bytes", len(raw)) + assert.That(t, len(raw) < 48<<10) +} + // widestLabels is the largest label set the config accepts: the most keys, // each at its length limit, each with a value at its length limit. // diff --git a/turbostats/wire/database.go b/turbostats/wire/database.go index 785a5bfa..c1b66118 100644 --- a/turbostats/wire/database.go +++ b/turbostats/wire/database.go @@ -2,6 +2,22 @@ package wire import "time" +// MaxDatabaseTables bounds Database.Tables. A reporter watches at most this +// many tables per endpoint; its config refuses more. It is the one number +// that sets a database bundle's width, so it is in the contract: the widest +// bundle of MaxDatabaseTables tables with the longest names Postgres allows +// is priced by TestCollect_AFullDatabaseBundleStaysUnderTheCeiling, and the +// receiver's body limit has to hold it. +const MaxDatabaseTables = 50 + +// MaxDatabaseReplicas bounds DatabaseReplication.Replicas. A primary with +// more sends the first MaxDatabaseReplicas by name. +const MaxDatabaseReplicas = 16 + +// MaxDatabaseErrors bounds DatabaseCollection.Errors: one per watched table +// and one per catalog query, which is fewer than ten. +const MaxDatabaseErrors = MaxDatabaseTables + 10 + // Database is the section a dbhealth reporter sends: one database endpoint, // probed and measured. Facts only: timestamps, counts and limits. The // receiver judges them. Fields a kind cannot provide are absent, not zero.