Organize DB Schema Migration Code and DB Init Checks

Makefile

@@ -78,7 +78,7 @@ perms-table:
 gen: cleangen proto api migrations fmt
 
 migrations:
-	$(MAKE) --environment-overrides -C internal/db/migrations/genmigrations migrations
+	$(MAKE) --environment-overrides -C internal/db/schema/migrations/generate migrations
 
 ### oplog requires protoc-gen-go v1.20.0 or later
 # GO111MODULE=on go get -u github.com/golang/protobuf/[email protected]

go.mod

@@ -9,6 +9,7 @@ replace github.com/hashicorp/boundary/sdk => ./sdk
 require (
 	github.com/armon/go-metrics v0.3.5
 	github.com/bufbuild/buf v0.33.0
+	github.com/dhui/dktest v0.3.3
 	github.com/fatih/color v1.10.0
 	github.com/favadi/protoc-go-inject-tag v1.1.0
 	github.com/go-bindata/go-bindata/v3 v3.1.3

internal/cmd/base/servers.go

@@ -18,6 +18,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/db/schema"
 	"github.com/hashicorp/boundary/internal/docker"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/types/scope"
@@ -442,7 +443,7 @@ func (b *Server) ConnectToDatabase(dialect string) error {
 	return nil
 }
 
-func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
+func (b *Server) CreateDevDatabase(ctx context.Context, dialect string, opt ...Option) error {
 	opts := getOpts(opt...)
 
 	var container, url string
@@ -470,17 +471,25 @@ func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
 			return fmt.Errorf("unable to start dev database with dialect %s: %w", dialect, err)
 		}
 
-		_, err := db.InitStore(dialect, c, url)
+		_, err := schema.InitStore(ctx, dialect, url)
 		if err != nil {
-			return fmt.Errorf("unable to initialize dev database with dialect %s: %w", dialect, err)
+			err = fmt.Errorf("unable to initialize dev database with dialect %s: %w", dialect, err)
+			if c != nil {
+				err = multierror.Append(err, c())
+			}
+			return err
 		}
 
 		b.DevDatabaseCleanupFunc = c
 		b.DatabaseUrl = url
 
 	default:
-		if _, err := db.InitStore(dialect, c, b.DatabaseUrl); err != nil {
-			return fmt.Errorf("error initializing store: %w", err)
+		if _, err := schema.InitStore(ctx, dialect, b.DatabaseUrl); err != nil {
+			err = fmt.Errorf("error initializing store: %w", err)
+			if c != nil {
+				err = multierror.Append(err, c())
+			}
+			return err
 		}
 	}
 
@@ -492,16 +501,25 @@ func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
 	}
 
 	if err := b.ConnectToDatabase(dialect); err != nil {
+		if c != nil {
+			err = multierror.Append(err, c())
+		}
 		return err
 	}
 
 	b.Database.LogMode(true)
 
-	if err := b.CreateGlobalKmsKeys(context.Background()); err != nil {
+	if err := b.CreateGlobalKmsKeys(ctx); err != nil {
+		if c != nil {
+			err = multierror.Append(err, c())
+		}
 		return err
 	}
 
-	if _, err := b.CreateInitialLoginRole(context.Background()); err != nil {
+	if _, err := b.CreateInitialLoginRole(ctx); err != nil {
+		if c != nil {
+			err = multierror.Append(err, c())
+		}
 		return err
 	}
 
@@ -512,7 +530,7 @@ func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
 		return nil
 	}
 
-	if _, _, err := b.CreateInitialAuthMethod(context.Background()); err != nil {
+	if _, _, err := b.CreateInitialAuthMethod(ctx); err != nil {
 		return err
 	}
 
@@ -523,7 +541,7 @@ func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
 		return nil
 	}
 
-	if _, _, err := b.CreateInitialScopes(context.Background()); err != nil {
+	if _, _, err := b.CreateInitialScopes(ctx); err != nil {
 		return err
 	}
 
@@ -545,7 +563,7 @@ func (b *Server) CreateDevDatabase(dialect string, opt ...Option) error {
 		return nil
 	}
 
-	if _, err := b.CreateInitialTarget(context.Background()); err != nil {
+	if _, err := b.CreateInitialTarget(ctx); err != nil {
 		return err
 	}
 

internal/cmd/commands.go

@@ -36,20 +36,14 @@ func initCommands(ui, serverCmdUi cli.Ui, runOpts *RunOptions) {
 	Commands = map[string]cli.CommandFactory{
 		"server": func() (cli.Command, error) {
 			return &server.Command{
-				Server: base.NewServer(&base.Command{
-					UI:         serverCmdUi,
-					ShutdownCh: base.MakeShutdownCh(),
-				}),
+				Server:    base.NewServer(base.NewCommand(serverCmdUi)),
 				SighupCh:  MakeSighupCh(),
 				SigUSR2Ch: MakeSigUSR2Ch(),
 			}, nil
 		},
 		"dev": func() (cli.Command, error) {
 			return &dev.Command{
-				Server: base.NewServer(&base.Command{
-					UI:         serverCmdUi,
-					ShutdownCh: base.MakeShutdownCh(),
-				}),
+				Server:    base.NewServer(base.NewCommand(serverCmdUi)),
 				SighupCh:  MakeSighupCh(),
 				SigUSR2Ch: MakeSigUSR2Ch(),
 			}, nil

internal/cmd/commands/database/init.go

@@ -7,9 +7,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
-	"github.com/hashicorp/boundary/internal/db"
-	"github.com/hashicorp/boundary/internal/db/migrations"
-	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/db/schema"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/boundary/sdk/wrapper"
 	wrapping "github.com/hashicorp/go-kms-wrapping"
@@ -186,8 +184,10 @@ func (c *InitCommand) Run(args []string) (retCode int) {
 		}()
 	}
 
-	if migrations.DevMigration != c.flagAllowDevMigrations {
-		if migrations.DevMigration {
+	dialect := "postgres"
+
+	if schema.DevMigration(dialect) != c.flagAllowDevMigrations {
+		if schema.DevMigration(dialect) {
 			c.UI.Error(base.WrapAtLength("This version of the binary has " +
 				"dev database schema updates which may not be supported in the " +
 				"next official release. To proceed anyways please use the " +
@@ -263,34 +263,63 @@ func (c *InitCommand) Run(args []string) (retCode int) {
 		return 1
 	}
 
-	migrationUrl, err := config.ParseAddress(migrationUrlToParse)
-	if err != nil && err != config.ErrNotAUrl {
-		c.UI.Error(fmt.Errorf("Error parsing migration url: %w", err).Error())
+	// This database is used to keep an exclusive lock on the database for the
+	// remainder of the command
+	dBase, err := sql.Open(dialect, dbaseUrl)
+	if err != nil {
+		c.UI.Error(fmt.Errorf("Error establishing db connection for locking: %w", err).Error())
+		return 1
+	}
+	man, err := schema.NewManager(c.Context, dialect, dBase)
+	if err != nil {
+		c.UI.Error(fmt.Errorf("Error setting up schema manager for locking: %w", err).Error())
 		return 1
 	}
-
-	// Core migrations using the migration URL
 	{
-		c.srv.DatabaseUrl = strings.TrimSpace(migrationUrl)
-		ldb, err := sql.Open("postgres", c.srv.DatabaseUrl)
+		st, err := man.CurrentState(c.Context)
 		if err != nil {
-			c.UI.Error(fmt.Errorf("Error opening database to check init status: %w", err).Error())
+			c.UI.Error(fmt.Errorf("Error getting database state: %w", err).Error())
 			return 1
 		}
-		_, err = ldb.QueryContext(c.Context, "select version from schema_migrations")
-		switch {
-		case err == nil:
-			if base.Format(c.UI) == "table" {
-				c.UI.Info("Database already initialized.")
-				return 0
-			}
-		case errors.IsMissingTableError(err):
-			// Doesn't exist so we continue on
-		default:
-			c.UI.Error(fmt.Errorf("Error querying database for init status: %w", err).Error())
+		if st.Dirty {
+			c.UI.Error(base.WrapAtLength("Database is in a bad initialization " +
+				"state.  Please revert back to the last known good state."))
+			return 1
+		}
+		if st.InitializationStarted {
+			// TODO: Separate from the "dirty" bit maintained by the schema
+			//  manager maintain a bit which indicates that this full command
+			//  was completed successfully (with all default resources being created).
+			//  Use that bit to determine if a previous init was completed
+			//  successfully or not.
+			c.UI.Error(base.WrapAtLength("Database has already been " +
+				"initialized. If the initialization did not complete successfully " +
+				"please revert the database to its fresh state."))
 			return 1
 		}
-		ran, err := db.InitStore("postgres", nil, c.srv.DatabaseUrl)
+	}
+
+	// This is an advisory locks on the DB which is released when the db session ends.
+	if err := man.ExclusiveLock(c.Context); err != nil {
+		c.UI.Error(fmt.Errorf("Error capturing an exclusive lock: %w", err).Error())
+		return 1
+	}
+	defer func() {
+		if err := man.ExclusiveUnlock(c.Context); err != nil {
+			c.UI.Error(fmt.Errorf("Unable to release exclusive lock to the database: %w", err).Error())
+		}
+	}()
+
+	migrationUrl, err := config.ParseAddress(migrationUrlToParse)
+	if err != nil && err != config.ErrNotAUrl {
+		c.UI.Error(fmt.Errorf("Error parsing migration url: %w", err).Error())
+		return 1
+	}
+
+	// Core migrations using the migration URL
+	{
+		migrationUrl = strings.TrimSpace(migrationUrl)
+		ran, err := schema.InitStore(c.Context, dialect, migrationUrl)
 		if err != nil {
 			c.UI.Error(fmt.Errorf("Error running database migrations: %w", err).Error())
 			return 1
@@ -308,7 +337,7 @@ func (c *InitCommand) Run(args []string) (retCode int) {
 
 	// Everything after is done with normal database URL and is affecting actual data
 	c.srv.DatabaseUrl = strings.TrimSpace(dbaseUrl)
-	if err := c.srv.ConnectToDatabase("postgres"); err != nil {
+	if err := c.srv.ConnectToDatabase(dialect); err != nil {
 		c.UI.Error(fmt.Errorf("Error connecting to database after migrations: %w", err).Error())
 		return 1
 	}

internal/cmd/commands/dev/dev.go

@@ -399,7 +399,7 @@ func (c *Command) Run(args []string) int {
 		if c.flagDisableDatabaseDestruction {
 			opts = append(opts, base.WithSkipDatabaseDestruction())
 		}
-		if err := c.CreateDevDatabase("postgres", opts...); err != nil {
+		if err := c.CreateDevDatabase(c.Context, "postgres", opts...); err != nil {
 			if err == docker.ErrDockerUnsupported {
 				c.UI.Error("Automatically starting a Docker container running Postgres is not currently supported on this platform. Please use -database-url to pass in a URL (or an env var or file reference to a URL) for connecting to an existing empty database.")
 				return 1
@@ -417,7 +417,7 @@ func (c *Command) Run(args []string) int {
 			return 1
 		}
 		c.DatabaseUrl = strings.TrimSpace(dbaseUrl)
-		if err := c.CreateDevDatabase("postgres"); err != nil {
+		if err := c.CreateDevDatabase(c.Context, "postgres"); err != nil {
 			c.UI.Error(fmt.Errorf("Error connecting to database: %w", err).Error())
 			return 1
 		}

internal/cmd/commands/server/server.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/hashicorp/boundary/internal/db/schema"
 	"github.com/hashicorp/boundary/internal/servers/controller"
 	"github.com/hashicorp/boundary/internal/servers/worker"
 	"github.com/hashicorp/boundary/sdk/wrapper"
@@ -341,6 +342,46 @@ func (c *Command) Run(args []string) int {
 			c.UI.Error(fmt.Errorf("Error connecting to database: %w", err).Error())
 			return 1
 		}
+
+		sMan, err := schema.NewManager(c.Context, "postgres", c.Database.DB())
+		if err != nil {
+			c.UI.Error(fmt.Errorf("Can't get schema manager: %w.", err).Error())
+			return 1
+		}
+		// This is an advisory locks on the DB which is released when the db session ends.
+		if err := sMan.SharedLock(c.Context); err != nil {
+			c.UI.Error(fmt.Errorf("Unable to gain shared access to the database: %w", err).Error())
+			return 1
+		}
+		defer func() {
+			if err := sMan.SharedUnlock(c.Context); err != nil {
+				c.UI.Error(fmt.Errorf("Unable to release shared lock to the database: %w", err).Error())
+			}
+		}()
+		ckState, err := sMan.CurrentState(c.Context)
+		if err != nil {
+			c.UI.Error(fmt.Errorf("Error checking schema state: %w", err).Error())
+			return 1
+		}
+		if !ckState.InitializationStarted {
+			c.UI.Error("Database has not been initialized. Please run `boundary database init`.")
+			return 1
+		}
+		if ckState.Dirty {
+			c.UI.Error(base.WrapAtLength("Database is in a bad state. Please revert the database into the last known good state."))
+			return 1
+		}
+		if ckState.BinarySchemaVersion > ckState.DatabaseSchemaVersion {
+			// TODO: Add the command to migrate up the schema version once that command exists.
+			c.UI.Error("Older schema version is than is expected from this binary.")
+			return 1
+		}
+		if ckState.BinarySchemaVersion < ckState.DatabaseSchemaVersion {
+			c.UI.Error(base.WrapAtLength(fmt.Sprintf("Newer schema version (%d) "+
+				"than this binary expects. Please use a newer version of the boundary "+
+				"binary.", ckState.DatabaseSchemaVersion)))
+			return 1
+		}
 	}
 
 	defer func() {