diff --git a/CHANGELOG.md b/CHANGELOG.md index 061c899..499950d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,7 @@ All notable changes to pgbot are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/), and the project aims for [Semantic Versioning](https://semver.org/). The `--json` contract is versioned -separately by `model.SchemaVersion` (currently 1.2.0). +separately by `model.SchemaVersion` (currently 1.3.0). ## [Unreleased] @@ -21,6 +21,20 @@ separately by `model.SchemaVersion` (currently 1.2.0). `pgbot ask "why is it slow?"`. ### Added +- **`pgbot inspect --all-instances` — every Aurora writer and reader behind one + endpoint (experimental)** (#23). An Aurora cluster endpoint stands for several + instances; this discovers the members with `aurora_replica_status()`, derives + each instance endpoint from the cluster endpoint's DNS name (custom domains + are followed through their CNAME), verifies each derived endpoint reached the + member it names with `aurora_db_instance_identifier()` before collecting, and + inspects every one through the existing fan-out — writer first, then readers, + composing with `--all-databases`. SQL and DNS only: no AWS credentials, CLI, + SDK, or API. An RDS Proxy endpoint, a non-RDS name, or an unreachable member + fails loudly rather than guessing; missing members mean partial coverage and + exit 3. Text output banners each target, JSON carries `server.instance` and + `server.instance_role`, SARIF/JUnit objects are prefixed `instance:/`, and + Prometheus series gain `instance` and `role` labels. Needs validation on a + real cluster — please report the cluster endpoint shape if derivation fails. - **`$PGSERVICE` as a connection fallback** (#25). When no connection string is passed and neither `$DATABASE_URL` nor `$PGBOT_DATABASE_URL` is set, pgbot now checks `$PGSERVICE` too, so a @@ -42,6 +56,10 @@ separately by `model.SchemaVersion` (currently 1.2.0). ever go to the Mantle host for the configured region, and Bedrock requests never follow redirects. +### Changed +- `model.ServerInfo` gains `instance` and `instance_role` (additive). JSON + contract `SchemaVersion` → **1.3.0**; a 1.2.0 consumer still parses 1.3.0 + output unchanged. ### Fixed - **Connection-string redaction now covers `?password=` in URL form.** libpq accepts the password as a query parameter as well as in the userinfo; the diff --git a/README.md b/README.md index 81f9de0..5d8759c 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ Provider notes

-> **Status: beta.** The `--json` contract is versioned (currently `1.2.0`, JSON +> **Status: beta.** The `--json` contract is versioned (currently `1.3.0`, JSON > Schema published in [`schema/`](schema/)) and breaking changes to it are > treated as breaking changes to the tool. The human-readable report is **not** > a stable interface — parse `--json`, not the terminal output. @@ -256,6 +256,7 @@ Key `inspect` flags: | `--profile=full\|schema` | `schema` runs only catalog-derived findings — safe on an empty CI database | | `--fail-on-new ` | act only on findings not already in a base report (migration PRs) | | `--all-databases` | inspect every database in the cluster; cluster-wide findings reported once | +| `--all-instances` | Aurora (experimental): discover every writer and reader instance behind the cluster endpoint and inspect each; composes with `--all-databases` | | `--config ` | a `.pgbot.toml` for thresholds, severity remaps, and `[[ignore]]` rules | Exit codes are a scriptable contract: `0` clean · `1` warn · `2` critical · `3` @@ -823,7 +824,7 @@ rates; the rest are point-in-time reads trended against the baseline. ## The `--json` contract `--json` (and `--format=json`) is the interface to build on — a versioned, -PII-free document (`schema_version`, currently `1.2.0`) whose machine-checkable +PII-free document (`schema_version`, currently `1.3.0`) whose machine-checkable JSON Schema is published in [`schema/`](schema/). Every section carries an `exactness` label — `sampled`, `cumulative`, `scraped`, or `unavailable` — so a consumer never mistakes a cumulative total for a live rate. @@ -1094,7 +1095,8 @@ pgbot inspect "$DATABASE_URL" --format=prometheus > /var/lib/node_exporter/pgbot mv /var/lib/node_exporter/pgbot.prom.$$ /var/lib/node_exporter/pgbot.prom # atomic ``` -Under `--all-databases`, each database's series carry a `database="…"` label. +Under `--all-databases`, each database's series carry a `database="…"` label; +under `--all-instances`, `instance="…"` and `role="writer|reader"` as well. ## The findings catalogue diff --git a/cmd/pgbot/alldbs.go b/cmd/pgbot/alldbs.go index 115455a..0cb827c 100644 --- a/cmd/pgbot/alldbs.go +++ b/cmd/pgbot/alldbs.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "sort" + "strings" "sync" "github.com/pgrundev/pgbot/internal/collect" @@ -16,23 +17,43 @@ import ( "github.com/pgrundev/pgbot/internal/store" ) -// runInspectAll inspects every connectable, non-template database in the cluster -// (B3). Cluster-wide findings (settings, replication, archiving, wraparound, -// cluster activity) are computed on every connection but reported ONCE — marked -// on the first database and dropped from the rest — so they aren't repeated N -// times. Connections are serial by default; --parallel caps concurrency, because -// opening N connections to a cluster that already fired connection_saturation -// would be the wrong default. +// inspectTarget is one (cluster member, database) pair the fan-out inspects. +// Without --all-instances the member is nil and the target is just a database; +// without --all-databases the database is "" — the connString's own. +type inspectTarget struct { + instance *conn.AuroraInstance + database string +} + +// label names the target in diagnostics. +func (t inspectTarget) label() string { + db := t.database + if db == "" { + db = "the connection's database" + } + if t.instance == nil { + return db + } + return fmt.Sprintf("instance %s (%s), %s", t.instance.ID, t.instance.Role(), db) +} + +// runInspectAll inspects every target of the fan-out: every connectable, +// non-template database (--all-databases, B3), every Aurora cluster member +// (--all-instances), or their cross product. Cluster-wide findings (settings, +// replication, archiving, wraparound, cluster activity) are computed on every +// connection but reported ONCE per server — marked on the first database and +// dropped from the rest — so they aren't repeated N times; under --all-instances +// each member is its own server and keeps its own copy, since parameter groups +// and replication state differ per instance. Connections are serial by default; +// --parallel caps concurrency, because opening N connections to a cluster that +// already fired connection_saturation would be the wrong default. func runInspectAll(ctx context.Context, connString string, f inspectFlags) error { - dbs, err := listAllDatabases(ctx, connString) + targets, err := planTargets(ctx, connString, f) if err != nil { - return fmt.Errorf("list databases: %s", conn.RedactConnString(err.Error())) - } - if len(dbs) == 0 { - return fmt.Errorf("no connectable databases found") + return err } - contexts := make([]*model.Context, len(dbs)) + contexts := make([]*model.Context, len(targets)) workers := f.parallel if workers < 1 { workers = 1 @@ -41,16 +62,18 @@ func runInspectAll(ctx context.Context, connString string, f inspectFlags) error var wg sync.WaitGroup var mu sync.Mutex var firstErr error - for i, db := range dbs { + skipped := 0 + for i, t := range targets { wg.Add(1) - go func(i int, db string) { + go func(i int, t inspectTarget) { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() - c, err := inspectOne(ctx, connString, db, f) + c, err := inspectOne(ctx, connString, t, f) if err != nil { - fmt.Fprintf(os.Stderr, "pgbot: skipping %s: %s\n", db, conn.RedactConnString(err.Error())) + fmt.Fprintf(os.Stderr, "pgbot: skipping %s: %s\n", t.label(), conn.RedactConnString(err.Error())) mu.Lock() + skipped++ if firstErr == nil { firstErr = err } @@ -58,11 +81,11 @@ func runInspectAll(ctx context.Context, connString string, f inspectFlags) error return } contexts[i] = c - }(i, db) + }(i, t) } wg.Wait() - // Keep only the databases that inspected, in cluster order. + // Keep only the targets that inspected, in plan order. var out []*model.Context for _, c := range contexts { if c != nil { @@ -70,13 +93,13 @@ func runInspectAll(ctx context.Context, connString string, f inspectFlags) error } } if len(out) == 0 { - return fmt.Errorf("every database failed to inspect: %w", firstErr) + return fmt.Errorf("every target failed to inspect: %w", firstErr) } - // Fingerprints must be distinct per database (P0-1) — assert it, because this - // is the mode that actually exercises many databases on one server. + // Fingerprints must be distinct per target (P0-1) — assert it, because this + // is the mode that actually exercises many databases (and members) at once. if fp := duplicateFingerprint(out); fp != "" { - return fmt.Errorf("internal: two databases share fingerprint %s — baselines would collide", fp) + return fmt.Errorf("internal: two targets share fingerprint %s — baselines would collide", fp) } dedupeClusterWide(out) @@ -87,6 +110,13 @@ func runInspectAll(ctx context.Context, connString string, f inspectFlags) error worst = code } } + // --all-instances promises the whole cluster; anything less is partial + // coverage, reported after the partial output and failed in the exit code so + // a script cannot mistake "the members we could reach" for "the cluster". + if f.allInstances && skipped > 0 { + fmt.Fprintf(os.Stderr, "pgbot: partial cluster coverage — %d of %d targets not inspected (see above)\n", skipped, len(targets)) + worst = partialCoverageExit(worst) + } if err := renderAll(out, f); err != nil { return err @@ -95,14 +125,111 @@ func runInspectAll(ctx context.Context, connString string, f inspectFlags) error return nil } -// inspectOne runs the full read-only pipeline for one database and returns its -// Context (no rendering, no exit). -func inspectOne(ctx context.Context, connString, database string, f inspectFlags) (*model.Context, error) { - target, err := conn.ConnectDB(ctx, connString, database) +// partialCoverageExit is the exit code for a fan-out that missed targets: the +// findings' own code if that is already a failure, else exitFailure. +func partialCoverageExit(worst int) int { + if worst < exitFailure { + return exitFailure + } + return worst +} + +// planTargets lists what the fan-out inspects, in output order. +func planTargets(ctx context.Context, connString string, f inspectFlags) ([]inspectTarget, error) { + dbs := []string{""} + if f.allDatabases { + list, err := listAllDatabases(ctx, connString) + if err != nil { + return nil, fmt.Errorf("list databases: %s", conn.RedactConnString(err.Error())) + } + if len(list) == 0 { + return nil, fmt.Errorf("no connectable databases found") + } + dbs = list + } + var instances []conn.AuroraInstance + if f.allInstances { + var err error + if instances, err = discoverAuroraInstances(ctx, connString); err != nil { + return nil, fmt.Errorf("discover instances: %s", conn.RedactConnString(err.Error())) + } + } + return composeTargets(instances, dbs), nil +} + +// composeTargets crosses members with databases: writer first (instances arrive +// in that order), every database of one member before the next member, so the +// output reads as one server at a time. +func composeTargets(instances []conn.AuroraInstance, dbs []string) []inspectTarget { + var out []inspectTarget + if len(instances) == 0 { + for _, db := range dbs { + out = append(out, inspectTarget{database: db}) + } + return out + } + for i := range instances { + inst := instances[i] + for _, db := range dbs { + out = append(out, inspectTarget{instance: &inst, database: db}) + } + } + return out +} + +// discoverAuroraInstances asks the entry endpoint for the cluster's members and +// derives each one's instance endpoint from the entry's DNS name — SQL and DNS +// only. A custom domain in front of the cluster endpoint is followed through +// its CNAME; a proxy or a non-RDS name fails here instead of guessing. +func discoverAuroraInstances(ctx context.Context, connString string) ([]conn.AuroraInstance, error) { + target, err := conn.Connect(ctx, connString) + if err != nil { + return nil, err + } + defer target.Close() + instances, err := conn.AuroraInstances(ctx, target) + if err != nil { + return nil, err + } + host, _ := hostPort(target) + endpoint, err := conn.CanonicalRDSHost(host) + if err != nil { + return nil, err + } + for i := range instances { + if instances[i].Host, err = conn.AuroraInstanceHost(endpoint, instances[i].ID); err != nil { + return nil, err + } + } + fmt.Fprintf(os.Stderr, "pgbot: %d Aurora instance(s) behind %s\n", len(instances), endpoint) + return instances, nil +} + +// inspectOne runs the full read-only pipeline for one target and returns its +// Context (no rendering, no exit). Under --all-instances the member's identity +// is verified before anything is collected, so a derived endpoint that reached +// the wrong member is an error, never a mislabeled report. +func inspectOne(ctx context.Context, connString string, t inspectTarget, f inspectFlags) (*model.Context, error) { + var target *conn.Target + var err error + if t.instance != nil { + target, err = conn.ConnectDBAt(ctx, connString, t.database, t.instance.Host) + } else { + target, err = conn.ConnectDB(ctx, connString, t.database) + } if err != nil { return nil, err } defer target.Close() + if t.instance != nil { + got, err := conn.AuroraInstanceIdentifier(ctx, target) + if err != nil { + return nil, err + } + if !strings.EqualFold(got, t.instance.ID) { + return nil, fmt.Errorf("%s reached instance %q, not %q — endpoint derivation does not fit this cluster; please report the cluster endpoint's shape", t.instance.Host, got, t.instance.ID) + } + } c, err := collect.Run(ctx, target, collect.Options{ Interval: f.interval, RawQueryText: f.rawQueries, ASHHz: f.ashHz, ASHWindow: f.window, Deadline: f.timeout, @@ -112,6 +239,10 @@ func inspectOne(ctx context.Context, connString, database string, f inspectFlags return nil, err } c.Server.ViaPooler = target.Pooler.Detected + if t.instance != nil { + c.Server.Instance = t.instance.ID + c.Server.InstanceRole = t.instance.Role() + } host, port := hostPort(target) c.Fingerprint = store.Fingerprint(host, port, c.Server.Database, target.Caps.SystemIdentifier) if !f.noStore { @@ -146,12 +277,13 @@ func listAllDatabases(ctx context.Context, connString string) ([]string, error) return dbs, rows.Err() } -// dedupeClusterWide keeps the FIRST occurrence of each cluster-wide finding and -// removes the rest, so it's reported once (B3). First occurrence, not "the first -// database's copy": if database 0's collection missed the finding (permissions, -// a per-connection timeout), stripping it from every later context would erase -// a live report — for checksum_failures that's a corruption finding vanishing -// from the output and the exit code. +// dedupeClusterWide keeps the FIRST occurrence of each cluster-wide finding per +// server and removes the rest, so it's reported once (B3). First occurrence, not +// "the first database's copy": if database 0's collection missed the finding +// (permissions, a per-connection timeout), stripping it from every later context +// would erase a live report — for checksum_failures that's a corruption finding +// vanishing from the output and the exit code. Under --all-instances the key +// includes the member: each instance is its own server with its own settings. func dedupeClusterWide(contexts []*model.Context) { seen := map[string]bool{} for _, c := range contexts { @@ -159,10 +291,11 @@ func dedupeClusterWide(contexts []*model.Context) { for j := range c.Findings { fd := c.Findings[j] if findings.ClusterWide(fd.ID) { - if seen[fd.ID] { + key := c.Server.Instance + "\x00" + fd.ID + if seen[key] { continue } - seen[fd.ID] = true + seen[key] = true fd.ClusterScoped = true } kept = append(kept, fd) @@ -202,8 +335,8 @@ func renderAll(contexts []*model.Context, f inspectFlags) error { if i > 0 { fmt.Fprintln(os.Stdout) } - fmt.Fprintf(os.Stdout, "═══ database: %s ═══\n", c.Server.Database) - opts := render.Options{Color: useColor(f.noColor), Width: terminalWidth(), Full: f.full, Host: c.Server.Database} + fmt.Fprintf(os.Stdout, "═══ %s ═══\n", contextHeader(c)) + opts := render.Options{Color: useColor(f.noColor), Width: terminalWidth(), Full: f.full, Host: contextScope(c)} if err := render.Terminal(os.Stdout, c, opts); err != nil { return err } @@ -212,20 +345,44 @@ func renderAll(contexts []*model.Context, f inspectFlags) error { } } -// mergeContexts flattens per-database findings into one Context for the SARIF/JUnit -// aggregate, prefixing each finding's object with its database so entries from -// different databases stay distinct. +// contextHeader is the text-mode banner for one target. +func contextHeader(c *model.Context) string { + if c.Server.Instance != "" { + return fmt.Sprintf("instance: %s (%s) · database: %s", c.Server.Instance, c.Server.InstanceRole, c.Server.Database) + } + return "database: " + c.Server.Database +} + +// contextScope is the short target name: "db" or "instance/db". +func contextScope(c *model.Context) string { + if c.Server.Instance != "" { + return c.Server.Instance + "/" + c.Server.Database + } + return c.Server.Database +} + +// mergeContexts flattens per-target findings into one Context for the SARIF/JUnit +// aggregate, prefixing each finding's object with its database (and member, under +// --all-instances) so entries from different targets stay distinct. Cluster-scoped +// findings keep their object: they were already reduced to one per server. func mergeContexts(contexts []*model.Context) *model.Context { merged := &model.Context{SchemaVersion: model.SchemaVersion} for _, c := range contexts { - db := c.Server.Database + scope := "db:" + c.Server.Database + if c.Server.Instance != "" { + scope = "instance:" + c.Server.Instance + "/" + scope + } for _, fd := range c.Findings { if !fd.ClusterScoped { if fd.Object == "" { - fd.Object = "db:" + db + fd.Object = scope } else { - fd.Object = "db:" + db + "/" + fd.Object + fd.Object = scope + "/" + fd.Object } + } else if c.Server.Instance != "" { + // One copy per member: name the member so two instances' copies + // of the same cluster-wide finding stay distinct in the aggregate. + fd.Object = "instance:" + c.Server.Instance } merged.Findings = append(merged.Findings, fd) } diff --git a/cmd/pgbot/allinstances_test.go b/cmd/pgbot/allinstances_test.go new file mode 100644 index 0000000..2048c11 --- /dev/null +++ b/cmd/pgbot/allinstances_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "strings" + "testing" + + "github.com/pgrundev/pgbot/internal/conn" + "github.com/pgrundev/pgbot/internal/findings" + "github.com/pgrundev/pgbot/internal/model" +) + +func TestComposeTargets(t *testing.T) { + // No instances: one target per database, as --all-databases always did. + got := composeTargets(nil, []string{"app", "billing"}) + if len(got) != 2 || got[0].instance != nil || got[0].database != "app" || got[1].database != "billing" { + t.Fatalf("database-only plan = %+v", got) + } + // --all-instances alone: every member at the connection's own database. + inst := []conn.AuroraInstance{{ID: "w", Writer: true, Host: "w.x.us-east-1.rds.amazonaws.com"}, {ID: "r1", Host: "r1.x.us-east-1.rds.amazonaws.com"}} + got = composeTargets(inst, []string{""}) + if len(got) != 2 || got[0].instance.ID != "w" || got[0].database != "" || got[1].instance.ID != "r1" { + t.Fatalf("instance-only plan = %+v", got) + } + // Both: the cross product, one member at a time, writer first. + got = composeTargets(inst, []string{"app", "billing"}) + want := []string{"w/app", "w/billing", "r1/app", "r1/billing"} + if len(got) != len(want) { + t.Fatalf("cross plan has %d targets, want %d", len(got), len(want)) + } + for i, w := range want { + if key := got[i].instance.ID + "/" + got[i].database; key != w { + t.Errorf("target %d = %s, want %s", i, key, w) + } + } + // Each target keeps its own member: a shared loop variable would alias them. + if got[0].instance == got[2].instance { + t.Error("targets share one AuroraInstance pointer across members") + } + if !strings.Contains(got[0].label(), "instance w (writer)") || !strings.Contains(got[0].label(), "app") { + t.Errorf("label = %q", got[0].label()) + } +} + +func TestPartialCoverageExit(t *testing.T) { + if partialCoverageExit(exitClean) != exitFailure || partialCoverageExit(exitCritical) != exitFailure { + t.Error("partial coverage must fail the run even when the findings are clean") + } + if partialCoverageExit(exitUsage) != exitUsage { + t.Error("a higher code must not be lowered") + } +} + +func instanceContext(id, role, db string, findingIDs ...string) *model.Context { + c := &model.Context{} + c.Server.Instance, c.Server.InstanceRole, c.Server.Database = id, role, db + for _, f := range findingIDs { + c.Findings = append(c.Findings, model.Finding{ID: f}) + } + return c +} + +// Cluster-wide findings are reduced to one per server. Under --all-instances a +// member is a server: the same finding on two members survives on both, while a +// second database on the same member drops it. +func TestDedupeClusterWide_perInstance(t *testing.T) { + cw := clusterWideFindingID(t) + contexts := []*model.Context{ + instanceContext("w", "writer", "app", cw), + instanceContext("w", "writer", "billing", cw), + instanceContext("r1", "reader", "app", cw), + } + dedupeClusterWide(contexts) + if len(contexts[0].Findings) != 1 || len(contexts[1].Findings) != 0 || len(contexts[2].Findings) != 1 { + t.Fatalf("findings per context after dedupe = %d/%d/%d; want 1/0/1", len(contexts[0].Findings), len(contexts[1].Findings), len(contexts[2].Findings)) + } + if !contexts[0].Findings[0].ClusterScoped || !contexts[2].Findings[0].ClusterScoped { + t.Error("surviving copies must be marked cluster-scoped") + } +} + +func TestMergeContexts_tagsByInstance(t *testing.T) { + cw := clusterWideFindingID(t) + contexts := []*model.Context{ + instanceContext("w", "writer", "app", "table_bloat", cw), + instanceContext("r1", "reader", "app", "table_bloat", cw), + } + contexts[0].Findings[0].Object = "public.orders" + contexts[1].Findings[0].Object = "public.orders" + dedupeClusterWide(contexts) + merged := mergeContexts(contexts) + var objects []string + for _, f := range merged.Findings { + objects = append(objects, f.ID+"="+f.Object) + } + joined := strings.Join(objects, " ") + for _, want := range []string{"table_bloat=instance:w/db:app/public.orders", "table_bloat=instance:r1/db:app/public.orders", cw + "=instance:w", cw + "=instance:r1"} { + if !strings.Contains(joined, want) { + t.Errorf("merged objects %q lack %q", joined, want) + } + } +} + +func TestContextHeaderAndScope(t *testing.T) { + plain := &model.Context{} + plain.Server.Database = "app" + if contextHeader(plain) != "database: app" || contextScope(plain) != "app" { + t.Errorf("plain header/scope = %q / %q", contextHeader(plain), contextScope(plain)) + } + member := instanceContext("prod-1", "writer", "app") + if contextHeader(member) != "instance: prod-1 (writer) · database: app" || contextScope(member) != "prod-1/app" { + t.Errorf("member header/scope = %q / %q", contextHeader(member), contextScope(member)) + } +} + +// clusterWideFindingID picks any finding the catalog marks cluster-wide, so the +// tests don't hard-code one that might be reclassified. +func clusterWideFindingID(t *testing.T) string { + t.Helper() + for _, id := range []string{"checksum_failures", "connection_saturation", "wraparound_risk", "archiver_failing", "replication_lag"} { + if findings.ClusterWide(id) { + return id + } + } + t.Skip("no known cluster-wide finding id available") + return "" +} diff --git a/cmd/pgbot/inspect.go b/cmd/pgbot/inspect.go index ddb1cc4..a9fd697 100644 --- a/cmd/pgbot/inspect.go +++ b/cmd/pgbot/inspect.go @@ -43,7 +43,8 @@ type inspectFlags struct { failOn string // exit non-zero on findings at/above this severity (B5-1) format string // text|json|sarif|junit (B5-2) allDatabases bool // inspect every database in the cluster (B3) - parallel int // max concurrent database inspections (B3); default 1 = serial + allInstances bool // Aurora: inspect every writer/reader instance behind the endpoint (experimental) + parallel int // max concurrent target inspections (B3); default 1 = serial profile string // full (default) | schema: emit only schema-scoped findings (D3-1) failOnNew string // path to a base report; act only on findings new vs it (D3-2) } @@ -84,7 +85,8 @@ func newInspectCmd() *cobra.Command { fl.StringVar(&f.profile, "profile", "full", "which findings to run: full (a live database) | schema (catalog-only, safe on an empty CI database)") fl.StringVar(&f.failOnNew, "fail-on-new", "", "path to a base report (JSON); mark findings already in it preexisting and act only on new ones") fl.BoolVar(&f.allDatabases, "all-databases", false, "inspect every database in the cluster (cluster-wide findings reported once)") - fl.IntVar(&f.parallel, "parallel", 1, "max databases inspected concurrently under --all-databases (default 1 = serial)") + fl.BoolVar(&f.allInstances, "all-instances", false, "Aurora: discover every writer and reader instance behind the cluster endpoint and inspect each (experimental; composes with --all-databases)") + fl.IntVar(&f.parallel, "parallel", 1, "max targets inspected concurrently under --all-databases / --all-instances (default 1 = serial)") return cmd } @@ -105,7 +107,7 @@ func runInspect(cmd *cobra.Command, args []string, f inspectFlags) error { if connString == "" { return fmt.Errorf("no connection string (pass one or set $DATABASE_URL)") } - if f.allDatabases { + if f.allDatabases || f.allInstances { return runInspectAll(cmd.Context(), connString, f) } diff --git a/docs/providers.md b/docs/providers.md index c375098..099b6ce 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -40,6 +40,7 @@ detection works even when the host is a bare IP or sits behind a proxy. - **Host metrics:** CPU / memory / disk IOPS live in **CloudWatch**, not Postgres — out of reach over a connection string. Everything pgbot computes over SQL works. - **`pg_monitor`:** grant as the master user (`rds_superuser`): `GRANT pg_monitor TO ;`. RDS does not expose OS-superuser, but `pg_monitor` is fully grantable, which is all pgbot needs. - **`pg_stat_statements`:** add `pg_stat_statements` to `shared_preload_libraries` in the **DB parameter group**, **reboot** the instance, then `CREATE EXTENSION pg_stat_statements;`. This is exactly the string pgbot prints on the degraded path. +- **Aurora clusters (experimental):** an Aurora endpoint stands for several instances, and pg_stat_* on one says nothing about the others. `pgbot inspect "$CLUSTER_URL" --all-instances` asks `aurora_replica_status()` for the members, derives each instance endpoint from the cluster endpoint's DNS name (`...rds.amazonaws.com`; a custom domain is followed through its CNAME), verifies with `aurora_db_instance_identifier()` that each derived endpoint reached the member it names, and inspects every one — writer first, then readers — using the same credentials and TLS settings. It needs only SQL and DNS: no AWS credentials, CLI, SDK, or API. It refuses an RDS Proxy endpoint (the members are hidden behind it) and any endpoint it cannot derive from, and any member it cannot reach makes the run report partial coverage and exit 3. Add `--all-databases` to cross members with databases. Each instance is its own server for cluster-wide findings, since parameter groups differ per instance. - **Pooler:** the default endpoint is a direct connection. RDS Proxy is opt-in; when used it is a transaction pooler and pgbot will note it (rates stay correct). - **Idle/stats:** always-on instance; cumulative stats persist normally. - **Capability-gated:** `pg_stat_io` requires PG 16+; `pg_stat_wal` requires PG 14+. Aurora reports storage differently from community Postgres — WAL/IO sections may read differently and need live confirmation. diff --git a/internal/conn/aurora.go b/internal/conn/aurora.go new file mode 100644 index 0000000..c17d555 --- /dev/null +++ b/internal/conn/aurora.go @@ -0,0 +1,164 @@ +package conn + +// Aurora instance discovery for `pgbot inspect --all-instances`. An Aurora +// cluster endpoint stands for several DB instances (one writer, N readers), and +// pg_stat_* on one of them says nothing about the others. Discovery needs only +// what the issue asked for — SQL and DNS: aurora_replica_status() lists the +// members, and every instance endpoint is derivable from the cluster endpoint's +// name. No AWS credentials, CLI, SDK, or RDS API. + +import ( + "context" + "errors" + "fmt" + "net" + "sort" + "strings" +) + +// AuroraInstance is one DB instance of an Aurora cluster. +type AuroraInstance struct { + ID string // DB instance identifier, e.g. "prod-db-instance-1" + Writer bool // the current writer (aurora_replica_status: session_id = MASTER_SESSION_ID) + Host string // the instance endpoint to reach it at; see AuroraInstanceHost +} + +// Role is what Server.InstanceRole reports. +func (i AuroraInstance) Role() string { + if i.Writer { + return "writer" + } + return "reader" +} + +// auroraStatusRow is the part of aurora_replica_status() discovery reads. +type auroraStatusRow struct { + ServerID string + SessionID string +} + +// AuroraInstances lists the cluster's members: writer first, then readers by ID. +// It refuses anything that is not Aurora rather than guessing. +func AuroraInstances(ctx context.Context, t *Target) ([]AuroraInstance, error) { + if t.Caps.Provider != ProviderAurora { + return nil, errors.New("not an Aurora cluster (aurora_version() is missing) — --all-instances needs a native Aurora endpoint") + } + rows, err := t.Pool.Query(ctx, `SELECT server_id, coalesce(session_id, '') FROM aurora_replica_status() ORDER BY server_id`) + if err != nil { + return nil, fmt.Errorf("aurora_replica_status(): %w", err) + } + defer rows.Close() + var status []auroraStatusRow + for rows.Next() { + var r auroraStatusRow + if err := rows.Scan(&r.ServerID, &r.SessionID); err != nil { + return nil, err + } + status = append(status, r) + } + if err := rows.Err(); err != nil { + return nil, err + } + return auroraInstancesFromStatus(status) +} + +// auroraInstancesFromStatus turns aurora_replica_status() rows into instances. +// Exactly one writer is expected; anything else is a topology pgbot does not +// understand (a failover in flight, a global-database secondary) and is reported +// rather than inspected under a wrong role. +func auroraInstancesFromStatus(status []auroraStatusRow) ([]AuroraInstance, error) { + var out []AuroraInstance + writers := 0 + for _, r := range status { + id := strings.TrimSpace(r.ServerID) + if id == "" { + continue + } + inst := AuroraInstance{ID: id, Writer: r.SessionID == "MASTER_SESSION_ID"} + if inst.Writer { + writers++ + } + out = append(out, inst) + } + if len(out) == 0 { + return nil, errors.New("aurora_replica_status() listed no instances") + } + if writers != 1 { + return nil, fmt.Errorf("aurora_replica_status() reports %d writers among %d instances; expected exactly one — refusing to guess roles", writers, len(out)) + } + sort.SliceStable(out, func(i, j int) bool { + if out[i].Writer != out[j].Writer { + return out[i].Writer + } + return out[i].ID < out[j].ID + }) + return out, nil +} + +// AuroraInstanceIdentifier returns the identifier of the instance this Target is +// connected to — the check that a derived endpoint reached the member it names. +func AuroraInstanceIdentifier(ctx context.Context, t *Target) (string, error) { + var id string + if err := t.Pool.QueryRow(ctx, `SELECT aurora_db_instance_identifier()`).Scan(&id); err != nil { + return "", fmt.Errorf("cannot verify the instance identity (aurora_db_instance_identifier()): %w", err) + } + return id, nil +} + +// AuroraInstanceHost derives an instance endpoint from a cluster, reader, custom, +// or instance endpoint of the same cluster. RDS names are +// +// .