deps: upgrade to pgx v5 and tern v2 (MARSOHS-1643)

README.md

@@ -16,6 +16,8 @@ fork is tracked under Jira epic
   serves the UI and the runner API, alerts on failures. State is in Postgres.
 - **Runner** (`tester run`): claims runs from the server, executes the test
   binary, streams results back. Any number of runners can share a server.
+- **Migrate** (`tester migrate`): applies pending database migrations and
+  exits, for running as a pre-deploy step instead of at server startup.
 - **Configuration**: one JSON file shared by server and runners.
 
 ```jsonc
@@ -76,11 +78,12 @@ tester serve \
   --base-url https://e2e.example.com \
   --api-key "$API_KEY" \
   --config /etc/tester/config.json \
-  --pg-dsn "$PG_DSN"
+  --pg-dsn "$PG_DSN" \
+  --migrate-on-start=false   # default true; see Database migrations
 ```
 
 Every flag can also be set as an environment variable prefixed with `SERVE_`
-(`SERVE_API_KEY`, `SERVE_PG_DSN`, ...).
+(`SERVE_API_KEY`, `SERVE_PG_DSN`, `SERVE_MIGRATE_ON_START`, ...).
 
 Slack alerting and the `/tester` slash command need `--slack-access-token` and
 `--slack-signing-secret`. Okta login for the UI needs `--okta-client-id`,
@@ -89,12 +92,70 @@ Slack alerting and the `/tester` slash command need `--slack-access-token` and
 
 #### Database migrations
 
-`tester serve` runs the migrations in `db/pg_migrations.go` at startup via
+Migrations live in `db/pg_migrations.go` and are applied via
 [tern](https://github.com/jackc/tern), under an advisory lock and each in its
-own transaction. Long-running DDL such as index builds on large tables should
-be applied by hand first with `CREATE INDEX CONCURRENTLY` (which cannot run in
-a transaction) and then declared in a migration with `IF NOT EXISTS` and the
-same name, so the migration is a no-op in production.
+own transaction. Every migration that runs is logged. Long-running DDL such as
+index builds on large tables should be applied by hand first with
+`CREATE INDEX CONCURRENTLY` (which cannot run in a transaction) and then
+declared in a migration with `IF NOT EXISTS` and the same name, so the
+migration is a no-op in production.
+
+There are two ways to run them:
+
+- **At server startup** (the default, `--migrate-on-start=true`):
+  `tester serve` migrates before it listens. Simple, but
+  the migration then runs inside the web container's health-check window (a
+  slow one makes the deploy look unhealthy and roll back mid-migration),
+  several web replicas serialize on the advisory lock, and `worker` replicas
+  roll independently so they may briefly run against a not-yet-migrated
+  schema.
+- **As a separate step** with `tester migrate`, which applies pending
+  migrations and exits 0 (non-zero on failure). Run it once per deploy before
+  anything else starts, and start the server with `--migrate-on-start=false`
+  (or `SERVE_MIGRATE_ON_START=false`): `tester serve` then only checks the
+  schema version and refuses to start (`database schema is at version N, this
+  binary expects M`) if the migrate step was skipped, rather than serving
+  against a stale schema. A schema *newer* than the binary is accepted so the
+  image can be rolled back.
+
+```sh
+tester migrate --pg-dsn "$PG_DSN"   # or MIGRATE_PG_DSN / SERVE_PG_DSN in the environment
+```
+
+`migrate` does not read the config file; it only needs the DSN. It falls back
+to `SERVE_PG_DSN` so the job can copy the web service's environment verbatim.
+
+On App Platform this is a `PRE_DEPLOY` job, which runs to completion before
+any component of the new deployment is started. The app spec lives in
+`digitalocean/e2e`, not here; the job uses the same image and DB env as `web`
+(`tester` is `/bin/tester` in the image, and `serve`/`run` are invoked the
+same way):
+
+```yaml
+jobs:
+  - name: migrate
+    kind: PRE_DEPLOY
+    image:                    # same image reference as the web/worker components
+      registry_type: DOCR
+      registry: do-e2e-canaries
+      repository: <e2e image repository>
+      tag: <same tag as web/worker>
+    run_command: tester migrate
+    envs:
+      - key: SERVE_PG_DSN     # same DB env as web; MIGRATE_PG_DSN also works
+        scope: RUN_TIME
+        value: ${db.DATABASE_URL}
+```
+
+Rollout order for switching an existing app over:
+
+1. Deploy an image containing `tester migrate` with `--migrate-on-start` left
+   at its default (true). Nothing changes yet.
+2. Add the `PRE_DEPLOY` job to the app spec. Both the job and the server now
+   migrate; the second is a no-op.
+3. On the `web` component of the app spec, set `SERVE_MIGRATE_ON_START=false`
+   in `envs` (or add `--migrate-on-start=false` to its `run_command`). No
+   image rebuild is needed. The server now only verifies the schema.
 
 ### Runner
 

cmd/tester/main.go

@@ -20,6 +20,7 @@ func init() {
 
 	rootCmd.AddCommand(serveCmd)
 	rootCmd.AddCommand(runCmd)
+	rootCmd.AddCommand(migrateCmd)
 }
 
 func main() {

cmd/tester/migrate.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/digitalocean/tester/db"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+var migrateCmd = &cobra.Command{
+	Use:   "migrate",
+	Short: "apply pending database migrations and exit",
+	Long: `migrate applies the database migrations that "tester serve" would
+otherwise run at startup, then exits 0. It is meant to run once per deploy,
+before the server and runners start (an App Platform PRE_DEPLOY job), so that
+the server can be started with --migrate-on-start=false.
+
+The DSN is taken from --pg-dsn, MIGRATE_PG_DSN, or, failing both, SERVE_PG_DSN
+so the job can reuse the web service's environment as is.`,
+	Args: cobra.ExactArgs(0),
+	Run: func(cmd *cobra.Command, args []string) {
+		dsn := viper.GetString("migrate-pg-dsn")
+		if dsn == "" {
+			dsn = viper.GetString("serve-pg-dsn")
+		}
+		if dsn == "" {
+			log.Fatal("no postgresql dsn: set --pg-dsn, MIGRATE_PG_DSN or SERVE_PG_DSN")
+		}
+
+		ctx := context.Background()
+		pool, err := pgxpool.New(ctx, dsn)
+		if err != nil {
+			log.Fatalf("failed to configure db pool: %s", err)
+		}
+		defer pool.Close()
+		err = pool.Ping(ctx)
+		if err != nil {
+			log.Fatalf("failed to connect to db: %s", err)
+		}
+
+		dbStore := db.NewPG(pool)
+		log.Print("applying database migrations")
+		err = dbStore.Init(ctx)
+		if err != nil {
+			log.Fatalf("failed to migrate db: %s", err)
+		}
+		log.Print("database schema is current")
+	},
+}
+
+func init() {
+	migrateCmd.Flags().String("pg-dsn", "", "The postgresql dsn to use.")
+	viper.BindPFlag("migrate-pg-dsn", migrateCmd.Flags().Lookup("pg-dsn"))
+}

cmd/tester/serve.go

@@ -20,7 +20,7 @@ import (
 	"github.com/digitalocean/tester/http/okta"
 	"github.com/digitalocean/tester/scheduler"
 	"github.com/digitalocean/tester/slack"
-	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -64,16 +64,31 @@ var serveCmd = &cobra.Command{
 			log.Fatalf("failed to listen on %s", viper.GetString("serve-addr"))
 		}
 
-		pool, err := pgxpool.Connect(context.Background(), viper.GetString("serve-pg-dsn"))
+		pool, err := pgxpool.New(context.Background(), viper.GetString("serve-pg-dsn"))
 		if err != nil {
-			log.Fatalf("failed to connect to db at %s: %s", viper.GetString("serve-addr"), err)
+			log.Fatalf("failed to configure db pool: %s", err)
 		}
 		defer pool.Close()
+		// pgxpool.New connects lazily; ping so a bad DSN fails at startup.
+		err = pool.Ping(context.Background())
+		if err != nil {
+			log.Fatalf("failed to connect to db: %s", err)
+		}
 
 		dbStore := db.NewPG(pool)
-		err = dbStore.Init(context.Background())
-		if err != nil {
-			log.Fatalf("failed to init db: %s", err)
+		if viper.GetBool("serve-migrate-on-start") {
+			err = dbStore.Init(context.Background())
+			if err != nil {
+				log.Fatalf("failed to init db: %s", err)
+			}
+		} else {
+			// Migrations are someone else's job (`tester migrate` in a
+			// pre-deploy step); refuse to serve against a stale schema.
+			log.Print("--migrate-on-start=false, skipping migrations")
+			err = dbStore.CheckSchema(context.Background())
+			if err != nil {
+				log.Fatalf("database schema is not current: %s", err)
+			}
 		}
 
 		var httpOpts []testerhttp.Option
@@ -231,6 +246,9 @@ func init() {
 	serveCmd.Flags().String("pg-dsn", "", "The postgresql dsn to use.")
 	viper.BindPFlag("serve-pg-dsn", serveCmd.Flags().Lookup("pg-dsn"))
 
+	serveCmd.Flags().Bool("migrate-on-start", true, "Apply pending database migrations at startup. Set false when `tester migrate` runs them in a pre-deploy job; the server then only verifies the schema is current.")
+	viper.BindPFlag("serve-migrate-on-start", serveCmd.Flags().Lookup("migrate-on-start"))
+
 	serveCmd.Flags().String("api-key", "", "Symmetric key for API Auth")
 	viper.BindPFlag("serve-api-key", serveCmd.Flags().Lookup("api-key"))
 

cmd/tester/serve_test.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestServe_migrateOnStart pins the flag/env wiring: default true, overridable
+// by SERVE_MIGRATE_ON_START, and the flag wins over the environment.
+func TestServe_migrateOnStart(t *testing.T) {
+	flag := serveCmd.Flags().Lookup("migrate-on-start")
+	require.NotNil(t, flag)
+	t.Cleanup(func() {
+		require.NoError(t, flag.Value.Set(flag.DefValue))
+		flag.Changed = false
+	})
+
+	assert.True(t, viper.GetBool("serve-migrate-on-start"), "default must be true")
+
+	t.Setenv("SERVE_MIGRATE_ON_START", "false")
+	assert.False(t, viper.GetBool("serve-migrate-on-start"), "env must override default")
+
+	require.NoError(t, serveCmd.ParseFlags([]string{"--migrate-on-start=true"}))
+	assert.True(t, viper.GetBool("serve-migrate-on-start"), "flag must override env")
+
+	require.NoError(t, serveCmd.ParseFlags([]string{"--migrate-on-start=false"}))
+	assert.False(t, viper.GetBool("serve-migrate-on-start"))
+}

db/pg.go

@@ -4,17 +4,18 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"log"
 	"math"
 	"strings"
 	"time"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/digitalocean/tester"
 	"github.com/google/uuid"
-	"github.com/jackc/pgconn"
-	"github.com/jackc/pgx/v4"
-	"github.com/jackc/pgx/v4/pgxpool"
-	"github.com/jackc/tern/migrate"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/jackc/tern/v2/migrate"
 )
 
 var psq = sq.StatementBuilder.PlaceholderFormat(sq.Dollar)
@@ -39,23 +40,71 @@ func NewPG(pool *pgxpool.Pool) *PG {
 	}
 }
 
+// migrator builds a tern Migrator on conn with pgMigrations appended. Every
+// migration that actually runs is logged, so both `tester serve` (when it
+// migrates at startup) and `tester migrate` show what they applied.
+func (p *PG) migrator(ctx context.Context, conn *pgx.Conn) (*migrate.Migrator, error) {
+	m, err := migrate.NewMigrator(ctx, conn, "versions")
+	if err != nil {
+		return nil, err
+	}
+	m.OnStart = func(sequence int32, name, direction, _ string) {
+		log.Printf("db: applying migration %d %q (%s)", sequence, name, direction)
+	}
+	for _, migration := range pgMigrations {
+		m.AppendMigration(migration.name, migration.up, migration.down)
+	}
+	return m, nil
+}
+
+// Init applies any pending migrations from pgMigrations, under tern's
+// advisory lock and each in its own transaction. It is a no-op when the
+// schema is already current.
 func (p *PG) Init(ctx context.Context) error {
 	conn, err := p.pool.Acquire(ctx)
 	if err != nil {
 		return err
 	}
 	defer conn.Release()
 
-	m, err := migrate.NewMigrator(ctx, conn.Conn(), "versions")
+	m, err := p.migrator(ctx, conn.Conn())
 	if err != nil {
 		return err
 	}
+	return m.Migrate(ctx)
+}
 
-	for _, migration := range pgMigrations {
-		m.AppendMigration(migration.name, migration.up, migration.down)
+// CheckSchema verifies that every migration in pgMigrations has been applied
+// without applying anything. It is for processes that must not migrate (a
+// server whose migrations run in a separate pre-deploy job): the returned
+// error names the current and expected versions so a deploy that started the
+// server before migrating fails fast instead of serving a stale schema.
+//
+// A schema newer than this binary knows about is tolerated so that rolling
+// back to a previous image does not take the server down; migrations are
+// expected to stay backwards compatible for one release. Building the
+// migrator takes tern's advisory lock briefly, so this blocks while another
+// process is mid-migration and reports the post-migration version.
+func (p *PG) CheckSchema(ctx context.Context) error {
+	conn, err := p.pool.Acquire(ctx)
+	if err != nil {
+		return err
 	}
+	defer conn.Release()
 
-	return m.Migrate(ctx)
+	m, err := p.migrator(ctx, conn.Conn())
+	if err != nil {
+		return err
+	}
+	current, err := m.GetCurrentVersion(ctx)
+	if err != nil {
+		return fmt.Errorf("reading schema version: %w", err)
+	}
+	expected := int32(len(pgMigrations))
+	if current < expected {
+		return fmt.Errorf("database schema is at version %d, this binary expects %d: run `tester migrate`", current, expected)
+	}
+	return nil
 }
 
 func (p *PG) tx(ctx context.Context, f func(tx pgx.Tx) error) error {
@@ -208,7 +257,7 @@ func (p *PG) StartRun(ctx context.Context, id uuid.UUID, runner string) error {
 
 		uq := psq.Update("runs").
 			Set("started_at", p.now()).
-			Set("meta", r.Meta).
+			Set("meta", jsonb(r.Meta)).
 			Where("id = ?", id)
 
 		sql, args, err = uq.ToSql()
@@ -320,7 +369,7 @@ func (p *PG) ResetRun(ctx context.Context, id uuid.UUID) error {
 			"started_at":  sql.NullTime{},
 			"finished_at": sql.NullTime{},
 			"error":       sql.NullString{},
-			"meta":        tester.RunMeta{},
+			"meta":        jsonb(tester.RunMeta{}),
 			"enqueued_at": p.now(),
 		}).
 		Set("reset_count", sq.Expr("reset_count + 1")).