Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/migrate-all-tenants/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/migrate-all-tenants/main.go |
| Original SHA-256 | 039c68b15ac2d7fc21b56ed787b265805c67ecdd2e0d9874a73a3e4a7cf58a3a |
| Git blob SHA | 386da220034396892f47473d2d29d9f37c27a861 |
| Reading-copy SHA-256 | 039c68b15ac2d7fc21b56ed787b265805c67ecdd2e0d9874a73a3e4a7cf58a3a |
Recovered source transcript
Download complete source text ↓352 lines · 13.6 KB
// Command migrate-all-tenants applies the migration set to the control plane database AND to every active
// tenant database in the database per tenant model. It reads the tenant directory in the control plane to
// discover the tenant databases, then for each target applies BOTH the public schema migrations and the
// segregated eeo migrations (two separate goose version tables), so a tenant database can never drift one
// schema relative to the other.
//
// It is a migrator role operator tool (it needs BYPASSRLS to write the directory bookkeeping and to run
// DDL). It is idempotent and resumable: goose records applied versions, so a re run is a no op on a
// healthy database. The failure surface is hard: it records which database errored or is behind, keeps
// going to the rest, and exits non zero if ANY database failed or is still behind head, so a partial fan
// out can never report success.
//
// Two modes:
//
// migrate-all-tenants apply pending migrations to the control plane and every active tenant
// migrate-all-tenants -check dry run: report each database's version and whether it is behind, NO writes
//
// It uses the per instance goose Provider (NOT the process global goose API, whose SetBaseFS and
// SetTableName state would race and cross contaminate the public versus eeo table name across N databases
// and under concurrency, the pressly/goose multi database guidance), and a Postgres session locker so two
// operators (or a concurrent provision-tenant) cannot migrate the same database at once.
package main
import (
"context"
"database/sql"
"errors"
"flag"
"fmt"
"io/fs"
"log"
"os"
"sort"
"strings"
"sync"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool"
_ "github.com/jackc/pgx/v5/stdlib"
goose "github.com/pressly/goose/v3"
goosedb "github.com/pressly/goose/v3/database"
gooselock "github.com/pressly/goose/v3/lock"
"golang.org/x/sync/errgroup"
dbembed "github.com/missionworkforce/missionmatch/db"
"github.com/missionworkforce/missionmatch/internal/platform/database"
"github.com/missionworkforce/missionmatch/internal/platform/directory"
)
type target struct {
name string // human label (control plane, or the tenant database name)
dsn string // migrator DSN to this database
org *uuid.UUID // nil for the control plane; set for a tenant (to record bookkeeping)
err error // set when the target could not even be ADDRESSED, so it is reported, never skipped
}
type result struct {
name string
pubVersion int64
eeoVersion int64
pending bool // dry run: any pending migration
appliedPub int
appliedEEO int
err error
}
func main() {
controlDSN := flag.String("control-dsn", "", "migrator DSN to the control plane database, defaults to MM_MIGRATOR_DATABASE_URL")
check := flag.Bool("check", false, "dry run: report each database's version and whether it is behind, make no changes")
concurrency := flag.Int("concurrency", 1, "migrate up to N databases at once (distinct databases only)")
flag.Parse()
if *controlDSN == "" {
*controlDSN = os.Getenv("MM_MIGRATOR_DATABASE_URL")
}
if *controlDSN == "" {
log.Fatal("migrate-all-tenants: -control-dsn (or MM_MIGRATOR_DATABASE_URL) is required")
}
if *concurrency < 1 {
*concurrency = 1
}
ctx := context.Background()
// The control plane pool: it must be the BYPASSRLS migrator so it can read the directory and run DDL.
pool, err := pgxpool.New(ctx, *controlDSN)
if err != nil {
log.Fatalf("migrate-all-tenants: control plane pool: %v", err)
}
defer pool.Close()
if err := database.AssertMigratorRole(ctx, pool); err != nil {
log.Fatalf("migrate-all-tenants: %v", err)
}
dir := directory.New(pool)
// Targets: the control plane first (the foundation), then every active tenant database.
//
// A tenant DSN is built from THE DIRECTORY ROW, host, port, and database name together, and never by
// assuming the tenant lives on the control plane's host.
//
// It used to swap only the database name into the control plane DSN, keeping that host and port. That
// silently defeated the entire point of tenant_directory carrying db_host and db_port: a tenant on any
// other machine was NEVER migrated, and because a database of the same name usually existed on the
// control plane host, the tool connected there instead, migrated the control plane a second time, and
// reported APPLIED for both. A fan out that cannot reach a tenant reported success.
//
// The cost was not theoretical. Production runs its control plane on one machine and its only tenant on
// another; the two drifted TWELVE migrations apart with every run reporting OK, until a deploy needed a
// column the control plane did not have and every authenticated session broke.
targets := []target{{name: "control plane", dsn: *controlDSN, org: nil}}
tenants, err := dir.List(ctx, directory.StatusActive)
if err != nil {
// BOOTSTRAP CASE. tenant_directory is itself created by a migration, so a database that is behind
// head (a fresh environment, or staging catching up) does not have the table the fan out reads. That
// must not stop the database from migrating ITSELF: refusing here means the one command that would
// create the directory can never run, which is a deadlock rather than a safety check. A missing
// relation is therefore treated as "no tenants yet", and every other error still fails hard, because
// a directory that exists but cannot be read is a real fault worth stopping for.
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && pgErr.Code == "42P01" {
log.Printf("migrate-all-tenants: tenant_directory does not exist yet, migrating the control plane only (bootstrap)")
tenants = nil
} else {
log.Fatalf("migrate-all-tenants: list tenants: %v", err)
}
}
for _, t := range tenants {
org := t.OrganizationID
dsn, derr := tenantDSN(*controlDSN, t)
if derr != nil {
// FAIL THIS TARGET LOUDLY rather than falling back to a DSN that would reach some other
// database. A silent fallback is precisely the bug being fixed here, and a tenant that cannot
// be addressed must be reported as a failure so the non zero exit code carries it.
targets = append(targets, target{name: t.DBName, dsn: "", org: &org, err: derr})
continue
}
targets = append(targets, target{name: t.DBName, dsn: dsn, org: &org})
}
log.Printf("migrate-all-tenants: %d targets (control plane + %d active tenants), check=%v concurrency=%d", len(targets), len(tenants), *check, *concurrency)
results := make([]result, len(targets))
var g errgroup.Group
g.SetLimit(*concurrency)
var mu sync.Mutex // guards the directory bookkeeping writes
for i := range targets {
i := i
g.Go(func() error {
r := migrateOne(ctx, targets[i], *check)
results[i] = r
// Record the applied version in the directory for a tenant that advanced (apply mode only).
if !*check && r.err == nil && targets[i].org != nil {
mu.Lock()
if e := dir.SetMigrated(ctx, *targets[i].org, r.pubVersion); e != nil {
log.Printf("migrate-all-tenants: WARNING could not record migration bookkeeping for %s: %v", targets[i].name, e)
}
mu.Unlock()
}
return nil // never abort the group on one database; the summary and exit code carry the verdict
})
}
_ = g.Wait()
// Summary and a hard exit code.
failed, behind := report(results, *check)
if failed > 0 || (*check && behind > 0) {
os.Exit(1)
}
}
// migrateOne applies (or, in check mode, inspects) the two migration sets on one database using per
// instance goose Providers and a session lock.
func migrateOne(ctx context.Context, t target, check bool) result {
r := result{name: t.name}
// A target that could not be ADDRESSED is a failure, reported with its reason and carried into the exit
// code. It is never quietly dropped from the fan out, because a tenant missing from the summary reads
// as a tenant that had nothing to do.
if t.err != nil {
r.err = t.err
return r
}
sqlDB, err := sql.Open("pgx", t.dsn)
if err != nil {
r.err = fmt.Errorf("open: %w", err)
return r
}
defer func() { _ = sqlDB.Close() }()
sqlDB.SetMaxOpenConns(5)
sqlDB.SetConnMaxLifetime(5 * time.Minute)
locker, err := gooselock.NewPostgresSessionLocker()
if err != nil {
r.err = fmt.Errorf("locker: %w", err)
return r
}
pubFS, err := fs.Sub(dbembed.Migrations, "migrations")
if err != nil {
r.err = err
return r
}
pub, err := goose.NewProvider(goose.DialectPostgres, sqlDB, pubFS, goose.WithSessionLocker(locker))
if err != nil {
r.err = fmt.Errorf("public provider: %w", err)
return r
}
defer func() { _ = pub.Close() }()
eeoStore, err := goosedb.NewStore(goosedb.DialectPostgres, "goose_db_version_eeo")
if err != nil {
r.err = fmt.Errorf("eeo store: %w", err)
return r
}
eeoFS, err := fs.Sub(dbembed.EEOMigrations, "migrations_eeo")
if err != nil {
r.err = err
return r
}
// Empty dialect because the table name is carried by the custom store.
eeo, err := goose.NewProvider("", sqlDB, eeoFS, goose.WithStore(eeoStore), goose.WithSessionLocker(locker))
if err != nil {
r.err = fmt.Errorf("eeo provider: %w", err)
return r
}
defer func() { _ = eeo.Close() }()
if check {
pubPending, e1 := pub.HasPending(ctx)
eeoPending, e2 := eeo.HasPending(ctx)
if e1 != nil || e2 != nil {
r.err = firstErr(e1, e2)
return r
}
r.pending = pubPending || eeoPending
r.pubVersion, _ = pub.GetDBVersion(ctx)
r.eeoVersion, _ = eeo.GetDBVersion(ctx)
return r
}
// Apply public then eeo, so the public foundation exists before the segregated eeo schema.
rp, err := pub.Up(ctx)
if err != nil {
r.err = fmt.Errorf("public migrations: %w", err)
return r
}
re, err := eeo.Up(ctx)
if err != nil {
r.err = fmt.Errorf("eeo migrations: %w", err)
return r
}
r.appliedPub, r.appliedEEO = len(rp), len(re)
r.pubVersion, _ = pub.GetDBVersion(ctx)
r.eeoVersion, _ = eeo.GetDBVersion(ctx)
return r
}
// report prints a per database table and returns how many failed and how many are behind.
func report(results []result, check bool) (failed, behind int) {
sorted := append([]result(nil), results...)
sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].name < sorted[j].name })
log.Printf("migrate-all-tenants: results")
for _, r := range sorted {
switch {
case r.err != nil:
failed++
log.Printf(" FAILED %-28s public=v%d eeo=v%d %v", r.name, r.pubVersion, r.eeoVersion, r.err)
case check && r.pending:
behind++
log.Printf(" BEHIND %-28s public=v%d eeo=v%d (has pending migrations)", r.name, r.pubVersion, r.eeoVersion)
case check:
log.Printf(" OK %-28s public=v%d eeo=v%d (at head)", r.name, r.pubVersion, r.eeoVersion)
default:
log.Printf(" APPLIED %-28s public=v%d (+%d) eeo=v%d (+%d)", r.name, r.pubVersion, r.appliedPub, r.eeoVersion, r.appliedEEO)
}
}
if failed > 0 {
log.Printf("migrate-all-tenants: %d database(s) FAILED", failed)
}
if check && behind > 0 {
log.Printf("migrate-all-tenants: %d database(s) BEHIND head", behind)
}
if failed == 0 && behind == 0 {
log.Printf("migrate-all-tenants: all databases OK")
}
return failed, behind
}
func firstErr(errs ...error) error {
for _, e := range errs {
if e != nil {
return e
}
}
return nil
}
// swapDB reuses a migrator DSN and swaps only the database name, so a tenant database is reached with the
// same migrator credential (the migrator is a cluster role). Mirrors the helper in cmd/provision-tenant.
// tenantDSN builds the migrator DSN for one tenant FROM ITS DIRECTORY ROW.
//
// Precedence, and the order matters:
//
// 1. MM_TENANT_MIGRATOR_DSN_<REF>, when set, is used verbatim. A tenant whose migrator credential differs
// from the control plane's (a managed database on another cloud, a rotated password) needs a way to say
// so, and this is it. The reference is folded exactly as internal/platform/secrets folds it, so one
// naming convention covers the application and this tool.
//
// 2. Otherwise the control plane's migrator CREDENTIAL is reused against the tenant's OWN host, port, and
// database name. That is the common case: one cluster role, several machines.
//
// What it will NOT do is keep the control plane's host. Silently connecting to the wrong machine and
// reporting success is the failure this function exists to make impossible, so a directory row with no host
// is an ERROR rather than an assumption.
func tenantDSN(controlDSN string, t directory.TenantConn) (string, error) {
if t.SecretRef != "" {
if override := os.Getenv("MM_TENANT_MIGRATOR_DSN_" + envKey(t.SecretRef)); override != "" {
return override, nil
}
}
if t.Host == "" || t.Port == 0 {
return "", fmt.Errorf("tenant %s has no db_host or db_port in tenant_directory, refusing to guess (set MM_TENANT_MIGRATOR_DSN_%s or repair the directory row)", t.DBName, envKey(t.SecretRef))
}
cfg, err := pgx.ParseConfig(controlDSN)
if err != nil {
return "", fmt.Errorf("parse control plane dsn: %w", err)
}
out := fmt.Sprintf("host=%s port=%d user=%s dbname=%s", t.Host, t.Port, cfg.User, t.DBName)
if cfg.Password != "" {
out += " password=" + cfg.Password
}
if cfg.TLSConfig == nil {
out += " sslmode=disable"
}
return out, nil
}
// envKey folds a secret reference into an environment variable suffix, upper cased with any character that
// is not A to Z or 0 to 9 becoming an underscore. It mirrors internal/platform/secrets exactly, so an
// operator learns ONE naming rule rather than two that look alike and differ in an edge case.
func envKey(ref string) string {
out := make([]rune, 0, len(ref))
for _, r := range strings.ToUpper(strings.TrimSpace(ref)) {
if (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
out = append(out, r)
continue
}
out = append(out, '_')
}
return string(out)
}