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/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/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..c1b66118 --- /dev/null +++ b/turbostats/wire/database.go @@ -0,0 +1,128 @@ +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. +// +// 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": {