CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adding Migrate Command (#872)

Browse Source
This commit is contained in:
Todd Knight authored and GitHub committed 2021-01-25 14:43:50 -07:00
1 parent 5de4befb58
commit 1f9fcad1cd
25 files changed
+590 -650
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

CHANGELOG.md
+6
View File
@@ -24,9 +24,15 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
* api/cli: On listing/reading, return a list of actions the user is authorized
to perform on the identified resources or their associated collections
([PR](https://github.com/hashicorp/boundary/pull/870))
* cli: Add a `database migrate` command which updates a database's schema to
the version supported by the boundary binary ([PR](https://github.com/hashicorp/boundary/pull/872)).
### Bug Fixes
* controller/db: Correctly check if db init previously completed successfully
when starting a controller or when running `database init`
([Issue](https://github.com/hashicorp/boundary/issues/805))
([PR](https://github.com/hashicorp/boundary/pull/842))
* cli: When `output-curl-string` is used with `update` or `add-/remove-/set-`
commands and automatic versioning is being used (that is, no `-version` flag
is given), it will now display the final call instead of the `GET` that
internal/cmd/base/servers.go
+4 -2
View File
@@ -471,7 +471,8 @@ func (b *Server) CreateDevDatabase(ctx context.Context, dialect string, opt ...O
return fmt.Errorf("unable to start dev database with dialect %s: %w", dialect, err)
}
_, err := schema.InitStore(ctx, dialect, url)
// Let migrate store manage the dirty bit since dev DBs should be ephemeral anyways.
_, err := schema.MigrateStore(ctx, dialect, url)
if err != nil {
err = fmt.Errorf("unable to initialize dev database with dialect %s: %w", dialect, err)
if c != nil {
@@ -484,7 +485,8 @@ func (b *Server) CreateDevDatabase(ctx context.Context, dialect string, opt ...O
b.DatabaseUrl = url
default:
if _, err := schema.InitStore(ctx, dialect, b.DatabaseUrl); err != nil {
// Let migrate store manage the dirty bit since dev DBs should be ephemeral anyways.
if _, err := schema.MigrateStore(ctx, dialect, b.DatabaseUrl); err != nil {
err = fmt.Errorf("error initializing store: %w", err)
if c != nil {
err = multierror.Append(err, c())
internal/cmd/commands.go
+5
View File
@@ -286,6 +286,11 @@ func initCommands(ui, serverCmdUi cli.Ui, runOpts *RunOptions) {
Command: base.NewCommand(ui),
}, nil
},
"database migrate": func() (cli.Command, error) {
return &database.MigrateCommand{
Command: base.NewCommand(ui),
}, nil
},
"groups": func() (cli.Command, error) {
return &groups.Command{
internal/cmd/commands/database/funcs.go
+60
View File
@@ -1,11 +1,71 @@
package database
import (
"context"
"database/sql"
"fmt"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/db/schema"
"github.com/hashicorp/boundary/internal/errors"
"github.com/mitchellh/cli"
)
// migrateDatabase updates the schema to the most recent version known by the binary.
// It owns the reporting to the UI any errors.
// Returns a cleanup function which must be called even if an error is returned and
// an error code where a non-zero value indicates an error happened.
func migrateDatabase(ctx context.Context, ui cli.Ui, dialect, u string) (func(), int) {
noop := func() {}
// This database is used to keep an exclusive lock on the database for the
// remainder of the command
dBase, err := sql.Open(dialect, u)
if err != nil {
ui.Error(fmt.Errorf("Error establishing db connection: %w", err).Error())
return noop, 1
}
if err := dBase.PingContext(ctx); err != nil {
ui.Error(fmt.Sprintf("Unable to connect to the database at %q", u))
return noop, 1
}
man, err := schema.NewManager(ctx, dialect, dBase)
if err != nil {
if errors.Match(errors.T(errors.MigrationLock), err) {
ui.Error("Unable to capture a lock on the database.")
} else {
ui.Error(fmt.Errorf("Error setting up schema manager: %w", err).Error())
}
return noop, 1
}
// This is an advisory lock on the DB which is released when the DB session ends.
if err := man.ExclusiveLock(ctx); err != nil {
ui.Error("Unable to capture a lock on the database.")
return noop, 1
}
unlock := func() {
// We don't report anything since this should resolve itself anyways.
_ = man.ExclusiveUnlock(ctx)
}
st, err := man.CurrentState(ctx)
if err != nil {
ui.Error(fmt.Errorf("Error getting database state: %w", err).Error())
return unlock, 1
}
if st.Dirty {
ui.Error(base.WrapAtLength("Database is in a bad state. Please revert back to the last known good state."))
return unlock, 1
}
if err := man.RollForward(ctx); err != nil {
ui.Error(fmt.Errorf("Error running database migrations: %w", err).Error())
return unlock, 1
}
if base.Format(ui) == "table" {
ui.Info("Migrations successfully run.")
}
return unlock, 0
}
type RoleInfo struct {
RoleId string `json:"scope_id"`
Name string `json:"name"`
internal/cmd/commands/database/funcs_test.go
+108
View File
@@ -0,0 +1,108 @@
package database
import (
"context"
"database/sql"
"testing"
"github.com/hashicorp/boundary/internal/cmd/base"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/schema"
"github.com/mitchellh/cli"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestMigrateDatabase(t *testing.T) {
ctx := context.Background()
dialect := "postgres"
cases := []struct {
name string
urlProvider func() string
expectedCode int
expectedOutput string
expectedError string
}{
{
name: "basic",
urlProvider: func() string {
c, u, _, err := db.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
return u
},
expectedCode: 0,
expectedOutput: "Migrations successfully run.\n",
},
{
name: "bad_url",
urlProvider: func() string { return "badurl" },
expectedCode: 1,
expectedError: "Unable to connect to the database at \"badurl\"\n",
},
{
name: "cant_get_lock",
urlProvider: func() string {
c, u, _, err := db.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
dBase, err := sql.Open(dialect, u)
require.NoError(t, err)
man, err := schema.NewManager(ctx, dialect, dBase)
require.NoError(t, err)
// This is an advisory lock on the DB which is released when the DB session ends.
err = man.ExclusiveLock(ctx)
require.NoError(t, err)
return u
},
expectedCode: 1,
expectedError: "Unable to capture a lock on the database.\n",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
u := tc.urlProvider()
ui := cli.NewMockUi()
clean, errCode := migrateDatabase(ctx, ui, dialect, u)
clean()
assert.EqualValues(t, tc.expectedCode, errCode)
assert.Equal(t, tc.expectedOutput, ui.OutputWriter.String())
assert.Equal(t, tc.expectedError, ui.ErrorWriter.String())
})
}
}
func TestVerifyOplogIsEmpty(t *testing.T) {
dialect := "postgres"
ctx := context.Background()
c, u, _, err := db.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
_ = c
dBase, err := sql.Open(dialect, u)
require.NoError(t, err)
man, err := schema.NewManager(ctx, dialect, dBase)
require.NoError(t, err)
require.NoError(t, man.RollForward(ctx))
cmd := InitCommand{Command: base.NewCommand(cli.NewMockUi())}
cmd.srv = base.NewServer(&base.Command{UI: cmd.UI})
cmd.srv.DatabaseUrl = u
require.NoError(t, cmd.srv.ConnectToDatabase(dialect))
assert.NoError(t, cmd.verifyOplogIsEmpty())
}
internal/cmd/commands/database/init.go
+36 -70
View File
@@ -1,13 +1,12 @@
package database
import (
"database/sql"
"fmt"
"strings"
"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/errors"
"github.com/hashicorp/boundary/internal/types/scope"
"github.com/hashicorp/boundary/sdk/wrapper"
wrapping "github.com/hashicorp/go-kms-wrapping"
@@ -194,9 +193,8 @@ func (c *InitCommand) Run(args []string) (retCode int) {
"'-allow-development-migrations' flag."))
return 2
} else {
c.UI.Error(base.WrapAtLength("The '-allow-development-migrations' " +
c.UI.Warn(base.WrapAtLength("The '-allow-development-migrations' " +
"flag was set but this binary has no dev database schema updates."))
return 3
}
}
@@ -239,12 +237,6 @@ func (c *InitCommand) Run(args []string) (retCode int) {
return 1
}
urlToParse := c.Config.Controller.Database.Url
if urlToParse == "" {
c.UI.Error(`"url" not specified in "database" config block"`)
return 1
}
var migrationUrlToParse string
if c.Config.Controller.Database.MigrationUrl != "" {
migrationUrlToParse = c.Config.Controller.Database.MigrationUrl
@@ -254,7 +246,12 @@ func (c *InitCommand) Run(args []string) (retCode int) {
}
// Fallback to using database URL for everything
if migrationUrlToParse == "" {
migrationUrlToParse = urlToParse
migrationUrlToParse = c.Config.Controller.Database.Url
}
if migrationUrlToParse == "" {
c.UI.Error(base.WrapAtLength(`neither "url" nor "migration_url" correctly set in "database" config block nor was the "migration-url" flag used`))
return 1
}
migrationUrl, err := config.ParseAddress(migrationUrlToParse)
@@ -262,77 +259,32 @@ func (c *InitCommand) Run(args []string) (retCode int) {
c.UI.Error(fmt.Errorf("Error parsing migration url: %w", err).Error())
return 1
}
migrationUrl = strings.TrimSpace(migrationUrl)
// This database is used to keep an exclusive lock on the database for the
// remainder of the command
dBase, err := sql.Open(dialect, migrationUrl)
if err != nil {
c.UI.Error(fmt.Errorf("Error establishing db connection for locking: %w", err).Error())
clean, errCode := migrateDatabase(c.Context, c.UI, dialect, migrationUrl)
defer clean()
if errCode != 0 {
return errCode
}
urlToParse := c.Config.Controller.Database.Url
if urlToParse == "" {
c.UI.Error(`"url" not specified in "database" config block`)
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
}
{
st, err := man.CurrentState(c.Context)
if err != nil {
c.UI.Error(fmt.Errorf("Error getting database state: %w", err).Error())
return 1
}
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
}
}
// 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())
}
}()
{
if err := man.RollForward(c.Context); err != nil {
c.UI.Error(fmt.Errorf("Error running database migrations: %w", err).Error())
return 1
}
if base.Format(c.UI) == "table" {
c.UI.Info("Migrations successfully run.")
}
}
dbaseUrl, err := config.ParseAddress(urlToParse)
c.srv.DatabaseUrl, err = config.ParseAddress(urlToParse)
if err != nil && err != config.ErrNotAUrl {
c.UI.Error(fmt.Errorf("Error parsing database url: %w", err).Error())
return 1
}
// Everything after is done with normal database URL and is affecting actual data
c.srv.DatabaseUrl = strings.TrimSpace(dbaseUrl)
if err := c.srv.ConnectToDatabase(dialect); err != nil {
c.UI.Error(fmt.Errorf("Error connecting to database after migrations: %w", err).Error())
return 1
}
if err := c.verifyOplogIsEmpty(); err != nil {
c.UI.Error(fmt.Sprintf("The database appears to have already been initialized: %v", err))
return 1
}
if err := c.srv.CreateGlobalKmsKeys(c.Context); err != nil {
c.UI.Error(fmt.Errorf("Error creating global-scope KMS keys: %w", err).Error())
return 1
@@ -538,3 +490,17 @@ func (c *InitCommand) ParseFlagsAndConfig(args []string) int {
return 0
}
func (c *InitCommand) verifyOplogIsEmpty() error {
const op = "database.(InitCommand).verifyOplogIsEmpty"
r := c.srv.Database.DB().QueryRowContext(c.Context, "select not exists(select 1 from oplog_entry limit 1)")
if r.Err() != nil {
return r.Err()
}
var empty bool
r.Scan(&empty)
if !empty {
return errors.New(errors.MigrationIntegrity, op, "oplog_entry is not empty")
}
return nil
}
internal/cmd/commands/database/migrate.go
+253
View File
@@ -0,0 +1,253 @@
package database
import (
"fmt"
"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/sdk/wrapper"
wrapping "github.com/hashicorp/go-kms-wrapping"
"github.com/hashicorp/vault/sdk/helper/mlock"
"github.com/mitchellh/cli"
"github.com/posener/complete"
)
var (
_ cli.Command = (*MigrateCommand)(nil)
_ cli.CommandAutocomplete = (*MigrateCommand)(nil)
)
type MigrateCommand struct {
*base.Command
srv *base.Server
SighupCh chan struct{}
ReloadedCh chan struct{}
SigUSR2Ch chan struct{}
Config *config.Config
configWrapper wrapping.Wrapper
flagConfig string
flagConfigKms string
flagLogLevel string
flagLogFormat string
flagMigrationUrl string
flagAllowDevMigrations bool
}
func (c *MigrateCommand) Synopsis() string {
return "Migrate Boundary's database to the most recent schema supported by this binary."
}
func (c *MigrateCommand) Help() string {
return base.WrapForHelpText([]string{
"Usage: boundary database migrate [options]",
"",
" Migrate Boundary's database:",
"",
" $ boundary database migrate -config=/etc/boundary/controller.hcl",
"",
" For a full list of examples, please see the documentation.",
}) + c.Flags().Help()
}
func (c *MigrateCommand) Flags() *base.FlagSets {
set := c.FlagSet(base.FlagSetHTTP)
f := set.NewFlagSet("Command options")
f.StringVar(&base.StringVar{
Name: "config",
Target: &c.flagConfig,
Completion: complete.PredictOr(
complete.PredictFiles("*.hcl"),
complete.PredictFiles("*.json"),
),
Usage: "Path to the configuration file.",
})
f.StringVar(&base.StringVar{
Name: "config-kms",
Target: &c.flagConfigKms,
Completion: complete.PredictOr(
complete.PredictFiles("*.hcl"),
complete.PredictFiles("*.json"),
),
Usage: `Path to a configuration file containing a "kms" block marked for "config" purpose, to perform decryption of the main configuration file. If not set, will look for such a block in the main configuration file, which has some drawbacks; see the help output for "boundary config encrypt -h" for details.`,
})
f.StringVar(&base.StringVar{
Name: "log-level",
Target: &c.flagLogLevel,
EnvVar: "BOUNDARY_LOG_LEVEL",
Completion: complete.PredictSet("trace", "debug", "info", "warn", "err"),
Usage: "Log verbosity level. Supported values (in order of more detail to less) are " +
"\"trace\", \"debug\", \"info\", \"warn\", and \"err\".",
})
f.StringVar(&base.StringVar{
Name: "log-format",
Target: &c.flagLogFormat,
Completion: complete.PredictSet("standard", "json"),
Usage: `Log format. Supported values are "standard" and "json".`,
})
f = set.NewFlagSet("Migration options")
f.BoolVar(&base.BoolVar{
Name: "allow-development-migrations",
Target: &c.flagAllowDevMigrations,
Usage: "If set the migrate command will continue even if the schema includes database update steps that may not be supported in the next official release. Boundary does not provide a rollback mechanism so a backup should be taken independently if needed.",
})
f.StringVar(&base.StringVar{
Name: "migration-url",
Target: &c.flagMigrationUrl,
Usage: `If set, overrides a migration URL set in config, and specifies the URL used to connect to the database for migration. This can allow different permissions for the user running initialization or migration vs. normal operation. This can refer to a file on disk (file://) from which a URL will be read; an env var (env://) from which the URL will be read; or a direct database URL.`,
})
return set
}
func (c *MigrateCommand) AutocompleteArgs() complete.Predictor {
return complete.PredictNothing
}
func (c *MigrateCommand) AutocompleteFlags() complete.Flags {
return c.Flags().Completions()
}
func (c *MigrateCommand) Run(args []string) (retCode int) {
if result := c.ParseFlagsAndConfig(args); result > 0 {
return result
}
if c.configWrapper != nil {
defer func() {
if err := c.configWrapper.Finalize(c.Context); err != nil {
c.UI.Warn(fmt.Errorf("Error finalizing config kms: %w", err).Error())
}
}()
}
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 " +
"'-allow-development-migrations' flag."))
return 2
} else {
c.UI.Warn(base.WrapAtLength("The '-allow-development-migrations' " +
"flag was set but this binary has no dev database schema updates."))
}
}
c.srv = base.NewServer(&base.Command{UI: c.UI})
if err := c.srv.SetupLogging(c.flagLogLevel, c.flagLogFormat, c.Config.LogLevel, c.Config.LogFormat); err != nil {
c.UI.Error(err.Error())
return 1
}
// If mlockall(2) isn't supported, show a warning. We disable this in dev
// because it is quite scary to see when first using Boundary. We also disable
// this if the user has explicitly disabled mlock in configuration.
if !c.Config.DisableMlock && !mlock.Supported() {
c.UI.Warn(base.WrapAtLength(
"WARNING! mlock is not supported on this system! An mlockall(2)-like " +
"syscall to prevent memory from being swapped to disk is not " +
"supported on this system. For better security, only run Boundary on " +
"systems where this call is supported. If you are running Boundary" +
"in a Docker container, provide the IPC_LOCK cap to the container."))
}
if c.Config.Controller == nil {
c.UI.Error(`"controller" config block not found`)
return 1
}
if c.Config.Controller.Database == nil {
c.UI.Error(`"controller.database" config block not found`)
return 1
}
var migrationUrlToParse string
if c.Config.Controller.Database.MigrationUrl != "" {
migrationUrlToParse = c.Config.Controller.Database.MigrationUrl
}
if c.flagMigrationUrl != "" {
migrationUrlToParse = c.flagMigrationUrl
}
// Fallback to using database URL for everything
if migrationUrlToParse == "" {
migrationUrlToParse = c.Config.Controller.Database.Url
}
if migrationUrlToParse == "" {
c.UI.Error(base.WrapAtLength(`neither "url" nor "migration_url" correctly set in "database" config block nor was the "migration-url" flag used`))
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())
return 1
}
clean, errCode := migrateDatabase(c.Context, c.UI, dialect, migrationUrl)
defer clean()
if errCode != 0 {
return errCode
}
return 0
}
func (c *MigrateCommand) ParseFlagsAndConfig(args []string) int {
var err error
f := c.Flags()
if err = f.Parse(args); err != nil {
c.UI.Error(err.Error())
return 1
}
// Validation
switch {
case len(c.flagConfig) == 0:
c.UI.Error("Must specify a config file using -config")
return 1
}
wrapperPath := c.flagConfig
if c.flagConfigKms != "" {
wrapperPath = c.flagConfigKms
}
wrapper, err := wrapper.GetWrapperFromPath(wrapperPath, "config")
if err != nil {
c.UI.Error(err.Error())
return 1
}
if wrapper != nil {
c.configWrapper = wrapper
if err := wrapper.Init(c.Context); err != nil {
c.UI.Error(fmt.Errorf("Could not initialize kms: %w", err).Error())
return 1
}
}
c.Config, err = config.LoadFile(c.flagConfig, wrapper)
if err != nil {
c.UI.Error("Error parsing config: " + err.Error())
return 1
}
return 0
}
internal/cmd/commands/dev/dev.go
+1 -2
View File
@@ -411,12 +411,11 @@ func (c *Command) Run(args []string) int {
c.ShutdownFuncs = append(c.ShutdownFuncs, c.DestroyDevDatabase)
}
default:
dbaseUrl, err := config.ParseAddress(c.flagDatabaseUrl)
c.DatabaseUrl, err = config.ParseAddress(c.flagDatabaseUrl)
if err != nil && err != config.ErrNotAUrl {
c.UI.Error(fmt.Errorf("Error parsing database url: %w", err).Error())
return 1
}
c.DatabaseUrl = strings.TrimSpace(dbaseUrl)
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
+31 -8
View File
@@ -9,9 +9,13 @@ 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"
"github.com/hashicorp/boundary/internal/db/schema"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
"github.com/hashicorp/boundary/internal/servers/controller"
"github.com/hashicorp/boundary/internal/servers/worker"
"github.com/hashicorp/boundary/internal/types/scope"
"github.com/hashicorp/boundary/sdk/wrapper"
"github.com/hashicorp/go-hclog"
wrapping "github.com/hashicorp/go-kms-wrapping"
@@ -332,12 +336,12 @@ func (c *Command) Run(args []string) int {
c.UI.Error(`"url" not specified in "controller.database" config block"`)
return 1
}
dbaseUrl, err := config.ParseAddress(c.Config.Controller.Database.Url)
var err error
c.DatabaseUrl, err = config.ParseAddress(c.Config.Controller.Database.Url)
if err != nil && err != config.ErrNotAUrl {
c.UI.Error(fmt.Errorf("Error parsing database url: %w", err).Error())
return 1
}
c.DatabaseUrl = strings.TrimSpace(dbaseUrl)
if err := c.ConnectToDatabase("postgres"); err != nil {
c.UI.Error(fmt.Errorf("Error connecting to database: %w", err).Error())
return 1
@@ -363,17 +367,12 @@ func (c *Command) Run(args []string) int {
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.")
c.UI.Error(base.WrapAtLength("Database schema must be updated to use this version. Run 'boundary database migrate' to update the database."))
return 1
}
if ckState.BinarySchemaVersion < ckState.DatabaseSchemaVersion {
@@ -382,6 +381,10 @@ func (c *Command) Run(args []string) int {
"binary.", ckState.DatabaseSchemaVersion)))
return 1
}
if err := c.verifyKmsSetup(); err != nil {
c.UI.Error(base.WrapAtLength("Database is in a bad state. Please revert the database into the last known good state."))
return 1
}
}
defer func() {
@@ -630,3 +633,23 @@ func (c *Command) Reload() error {
return reloadErrors.ErrorOrNil()
}
func (c *Command) verifyKmsSetup() error {
const op = "server.(Command).verifyKmsExists"
rw := db.New(c.Database)
kmsRepo, err := kms.NewRepository(rw, rw)
if err != nil {
return fmt.Errorf("error creating kms repository: %w", err)
}
rks, err := kmsRepo.ListRootKeys(c.Context, kms.WithLimit(1))
if err != nil {
return err
}
for _, rk := range rks {
if rk.GetScopeId() == scope.Global.String() {
return nil
}
}
return errors.New(errors.MigrationIntegrity, op, "can't find global scoped root key")
}
internal/cmd/config/config.go
+2 -2
View File
@@ -378,9 +378,9 @@ func ParseAddress(addr string) (string, error) {
if err != nil {
return addr, fmt.Errorf("error reading file at %s: %w", addr, err)
}
return string(contents), nil
return strings.TrimSpace(string(contents)), nil
case "env":
return os.Getenv(strings.TrimPrefix(addr, "env://")), nil
return strings.TrimSpace(os.Getenv(strings.TrimPrefix(addr, "env://"))), nil
}
return addr, nil
internal/db/schema/manager.go
+11 -13
View File
@@ -22,7 +22,7 @@ type driver interface {
// A value of -1 indicates no version is set.
SetVersion(context.Context, int, bool) error
// A value of -1 indicates no version is set.
Version(context.Context) (int, bool, error)
CurrentState(context.Context) (int, bool, error)
}
// Manager provides a way to run operations and retrieve information regarding
@@ -70,22 +70,22 @@ func (b *Manager) CurrentState(ctx context.Context) (*State, error) {
dbS := State{
BinarySchemaVersion: BinarySchemaVersion(b.dialect),
}
v, dirty, err := b.driver.Version(ctx)
v, dirty, err := b.driver.CurrentState(ctx)
if err != nil {
return nil, err
}
dbS.DatabaseSchemaVersion = v
dbS.Dirty = dirty
if v == nilVersion {
return &dbS, nil
}
dbS.InitializationStarted = true
dbS.DatabaseSchemaVersion = v
dbS.Dirty = dirty
return &dbS, nil
}
// SharedLock attempts to obtain a shared lock on the database. This can fail if
// an exclusive lock is already held with the same key. An error is returned if
// a lock was unable to be obtained.
// SharedLock attempts to obtain a shared lock on the database. This can fail
// if an exclusive lock is already held. If the lock can't be obtained an
// error is returned.
func (b *Manager) SharedLock(ctx context.Context) error {
const op = "schema.(Manager).SharedLock"
if err := b.driver.TrySharedLock(ctx); err != nil {
@@ -140,7 +140,7 @@ func (b *Manager) RollForward(ctx context.Context) error {
b.driver.Unlock(ctx)
}()
curVersion, dirty, err := b.driver.Version(ctx)
curVersion, dirty, err := b.driver.CurrentState(ctx)
if err != nil {
return errors.Wrap(err, op)
}
@@ -149,11 +149,7 @@ func (b *Manager) RollForward(ctx context.Context) error {
return errors.New(errors.NotSpecificIntegrity, op, fmt.Sprintf("schema is dirty with version %d", curVersion))
}
sp, err := newStatementProvider(b.dialect, curVersion)
if err != nil {
return errors.Wrap(err, op)
}
return b.runMigrations(ctx, sp)
return b.runMigrations(ctx, newStatementProvider(b.dialect, curVersion))
}
// runMigrations passes migration queries to a database driver and manages
@@ -174,6 +170,8 @@ func (b *Manager) runMigrations(ctx context.Context, qp *statementProvider) erro
return errors.Wrap(err, op)
}
// TODO: Wrap all the queries provided by the statementProvider into a
// single transaction which includes setting the version number.
if err := b.driver.Run(ctx, bytes.NewReader(qp.ReadUp())); err != nil {
return errors.Wrap(err, op)
}
internal/db/schema/manager_test.go
+27 -26
View File
@@ -13,27 +13,29 @@ import (
)
func TestNewManager(t *testing.T) {
c, u, _, err := docker.StartDbInDocker("postgres")
dialect := "postgres"
c, u, _, err := docker.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
d, err := sql.Open("postgres", u)
d, err := sql.Open(dialect, u)
require.NoError(t, err)
ctx := context.Background()
_, err = NewManager(ctx, "postgres", d)
_, err = NewManager(ctx, dialect, d)
require.NoError(t, err)
_, err = NewManager(ctx, "unknown", d)
assert.True(t, errors.Match(errors.T(errors.InvalidParameter), err))
d.Close()
_, err = NewManager(ctx, "postgres", d)
_, err = NewManager(ctx, dialect, d)
assert.True(t, errors.Match(errors.T(errors.Op("schema.NewManager")), err))
}
func TestCurrentState(t *testing.T) {
c, u, _, err := docker.StartDbInDocker("postgres")
dialect := "postgres"
c, u, _, err := docker.StartDbInDocker(dialect)
t.Cleanup(func() {
if err := c(); err != nil {
t.Fatalf("Got error at cleanup: %v", err)
@@ -44,13 +46,14 @@ func TestCurrentState(t *testing.T) {
require.NoError(t, c())
})
ctx := context.Background()
d, err := sql.Open("postgres", u)
d, err := sql.Open(dialect, u)
require.NoError(t, err)
m, err := NewManager(ctx, "postgres", d)
m, err := NewManager(ctx, dialect, d)
require.NoError(t, err)
want := &State{
BinarySchemaVersion: BinarySchemaVersion("postgres"),
BinarySchemaVersion: BinarySchemaVersion(dialect),
DatabaseSchemaVersion: nilVersion,
}
s, err := m.CurrentState(ctx)
require.NoError(t, err)
@@ -62,7 +65,7 @@ func TestCurrentState(t *testing.T) {
want = &State{
InitializationStarted: true,
BinarySchemaVersion: BinarySchemaVersion("postgres"),
BinarySchemaVersion: BinarySchemaVersion(dialect),
Dirty: true,
DatabaseSchemaVersion: 2,
}
@@ -72,19 +75,18 @@ func TestCurrentState(t *testing.T) {
}
func TestRollForward(t *testing.T) {
c, u, _, err := docker.StartDbInDocker("postgres")
dialect := "postgres"
c, u, _, err := docker.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
d, err := sql.Open("postgres", u)
d, err := sql.Open(dialect, u)
require.NoError(t, err)
ctx := context.Background()
m, err := NewManager(ctx, "postgres", d)
m, err := NewManager(ctx, dialect, d)
require.NoError(t, err)
assert.NoError(t, m.RollForward(ctx))
// Now set to dirty at an early version
testDriver, err := postgres.New(ctx, d)
require.NoError(t, err)
@@ -99,7 +101,7 @@ func TestRollForward_NotFromFresh(t *testing.T) {
nState := createPartialMigrationState(oState, 8)
migrationStates[dialect] = nState
c, u, _, err := docker.StartDbInDocker("postgres")
c, u, _, err := docker.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
@@ -113,7 +115,7 @@ func TestRollForward_NotFromFresh(t *testing.T) {
require.NoError(t, err)
assert.NoError(t, m.RollForward(ctx))
ver, dirty, err := m.driver.Version(ctx)
ver, dirty, err := m.driver.CurrentState(ctx)
assert.NoError(t, err)
assert.Equal(t, nState.binarySchemaVersion, ver)
assert.False(t, dirty)
@@ -124,7 +126,7 @@ func TestRollForward_NotFromFresh(t *testing.T) {
newM, err := NewManager(ctx, dialect, d)
require.NoError(t, err)
assert.NoError(t, newM.RollForward(ctx))
ver, dirty, err = newM.driver.Version(ctx)
ver, dirty, err = newM.driver.CurrentState(ctx)
assert.NoError(t, err)
assert.Equal(t, oState.binarySchemaVersion, ver)
assert.False(t, dirty)
@@ -132,19 +134,20 @@ func TestRollForward_NotFromFresh(t *testing.T) {
func TestManager_ExclusiveLock(t *testing.T) {
ctx := context.Background()
c, u, _, err := docker.StartDbInDocker("postgres")
dialect := "postgres"
c, u, _, err := docker.StartDbInDocker(dialect)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, c())
})
d1, err := sql.Open("postgres", u)
d1, err := sql.Open(dialect, u)
require.NoError(t, err)
m1, err := NewManager(ctx, "postgres", d1)
m1, err := NewManager(ctx, dialect, d1)
require.NoError(t, err)
d2, err := sql.Open("postgres", u)
d2, err := sql.Open(dialect, u)
require.NoError(t, err)
m2, err := NewManager(ctx, "postgres", d2)
m2, err := NewManager(ctx, dialect, d2)
require.NoError(t, err)
assert.NoError(t, m1.ExclusiveLock(ctx))
@@ -182,9 +185,8 @@ func TestManager_SharedLock(t *testing.T) {
// Creates a new migrationState only with the versions <= the provided maxVer
func createPartialMigrationState(om migrationState, maxVer int) migrationState {
nState := migrationState{
devMigration: om.devMigration,
upMigrations: make(map[int][]byte),
downMigrations: make(map[int][]byte),
devMigration: om.devMigration,
upMigrations: make(map[int][]byte),
}
for k := range om.upMigrations {
if k > maxVer {
@@ -192,7 +194,6 @@ func createPartialMigrationState(om migrationState, maxVer int) migrationState {
continue
}
nState.upMigrations[k] = om.upMigrations[k]
nState.downMigrations[k] = om.downMigrations[k]
if nState.binarySchemaVersion < k {
nState.binarySchemaVersion = k
}
internal/db/schema/migrations/generate/generate.go
-10
View File
@@ -34,7 +34,6 @@ func generate(dialect string) {
Content string
}
var upContents []ContentValues
var downContents []ContentValues
isDev := false
var lRelVer, largestSchemaVersion int
@@ -102,27 +101,21 @@ func generate(dialect string) {
Content: string(contents),
}
switch {
case strings.Contains(nameParts[1], ".down."):
downContents = append(downContents, cv)
case strings.Contains(nameParts[1], ".up."):
upContents = append(upContents, cv)
}
}
}
// fmt.Printf("Got upcontent: %#v\n\n downcontents: %#v", upContents[:1], downContents[:1])
outBuf := bytes.NewBuffer(nil)
if err := migrationsTemplate.Execute(outBuf, struct {
Type string
UpValues []ContentValues
DownValues []ContentValues
DevMigration bool
BinarySchemaVersion int
}{
Type: dialect,
UpValues: upContents,
DownValues: downContents,
DevMigration: isDev,
BinarySchemaVersion: largestSchemaVersion,
}); err != nil {
@@ -150,9 +143,6 @@ func init() {
upMigrations: map[int][]byte{
{{range .UpValues }}{{ template "Content" . }}{{end}}
},
downMigrations: map[int][]byte{
{{range .DownValues }}{{ template "Content" . }}{{end}}
},
}
}
`))
internal/db/schema/postgres/postgres.go
+10 -12
View File
@@ -52,6 +52,7 @@ const (
var defaultMigrationsTable = "boundary_schema_version"
// Postgres is a driver usable by a boundary schema manager.
// This struct is not thread safe.
type Postgres struct {
// Locking and unlocking need to use the same connection
conn *sql.Conn
@@ -86,7 +87,8 @@ func New(ctx context.Context, instance *sql.DB) (*Postgres, error) {
// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
func (p *Postgres) TrySharedLock(ctx context.Context) error {
const op = "postgres.(Postgres).TrySharedLock"
r := p.conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock_shared($1)", schemaAccessLockId)
const query = "SELECT pg_try_advisory_lock_shared($1)"
r := p.conn.QueryRowContext(ctx, query, schemaAccessLockId)
if r.Err() != nil {
return errors.Wrap(r.Err(), op)
}
@@ -104,8 +106,8 @@ func (p *Postgres) TrySharedLock(ctx context.Context) error {
// https://www.postgresql.org/docs/9.6/static/explicit-locking.html#ADVISORY-LOCKS
func (p *Postgres) TryLock(ctx context.Context) error {
const op = "postgres.(Postgres).TryLock"
r := p.conn.QueryRowContext(ctx, "SELECT pg_try_advisory_lock($1)", schemaAccessLockId)
const query = "SELECT pg_try_advisory_lock($1)"
r := p.conn.QueryRowContext(ctx, query, schemaAccessLockId)
if r.Err() != nil {
return errors.Wrap(r.Err(), op)
}
@@ -123,13 +125,10 @@ func (p *Postgres) TryLock(ctx context.Context) error {
// if we were unable to get the lock before the context cancels.
func (p *Postgres) Lock(ctx context.Context) error {
const op = "postgres.(Postgres).Lock"
// This will wait indefinitely until the Lock can be acquired.
query := `SELECT pg_advisory_lock($1)`
const query = "SELECT pg_advisory_lock($1)"
if _, err := p.conn.ExecContext(ctx, query, schemaAccessLockId); err != nil {
return errors.Wrap(err, op)
}
return nil
}
@@ -137,8 +136,7 @@ func (p *Postgres) Lock(ctx context.Context) error {
// release the lock before the context cancels.
func (p *Postgres) Unlock(ctx context.Context) error {
const op = "postgres.(Postgres).Unlock"
query := `SELECT pg_advisory_unlock($1)`
const query = `SELECT pg_advisory_unlock($1)`
if _, err := p.conn.ExecContext(ctx, query, schemaAccessLockId); err != nil {
return errors.Wrap(err, op)
}
@@ -266,9 +264,9 @@ func (p *Postgres) SetVersion(ctx context.Context, version int, dirty bool) erro
return nil
}
// Version returns the version, if the database is currently in a dirty state, and any error.
// CurrentState returns the version, if the database is currently in a dirty state, and any error.
// A version value of -1 indicates no version is set.
func (p *Postgres) Version(ctx context.Context) (version int, dirty bool, err error) {
func (p *Postgres) CurrentState(ctx context.Context) (version int, dirty bool, err error) {
const op = "postgres.(Postgres).Version"
query := `SELECT Version, dirty FROM ` + pq.QuoteIdentifier(defaultMigrationsTable) + ` LIMIT 1`
err = p.conn.QueryRowContext(ctx, query).Scan(&version, &dirty)
@@ -337,7 +335,7 @@ func (p *Postgres) drop(ctx context.Context) (err error) {
// convention of "caller locks" in the postgres type.
func (p *Postgres) ensureVersionTable(ctx context.Context) (err error) {
const op = "postgres.(Postgres).ensureVersionTable"
if err = p.Lock(ctx); err != nil {
if err = p.TryLock(ctx); err != nil {
return errors.Wrap(err, op)
}
internal/db/schema/postgres/postgres_test.go
+7 -51
View File
@@ -40,7 +40,6 @@ import (
"log"
"strconv"
"strings"
"sync"
"testing"
"github.com/dhui/dktest"
@@ -136,10 +135,10 @@ func TestVersion_NoVersionTable(t *testing.T) {
t.Error(err)
}
}()
// Drop the version table so calls to Version don't rely on that
// Drop the version table so calls to CurrentState don't rely on that
d.drop(ctx)
v, dirt, err := d.Version(ctx)
v, dirt, err := d.CurrentState(ctx)
assert.NoError(t, err)
assert.Equal(t, v, nilVersion)
assert.False(t, dirt)
@@ -218,7 +217,7 @@ func TestWithSchema(t *testing.T) {
}
}()
version, _, err := d2.Version(ctx)
version, _, err := d2.CurrentState(ctx)
if err != nil {
t.Fatal(err)
}
@@ -226,11 +225,11 @@ func TestWithSchema(t *testing.T) {
t.Fatal("expected NilVersion")
}
// now update Version and compare
// now update CurrentState and compare
if err := d2.SetVersion(ctx, 2, false); err != nil {
t.Fatal(err)
}
version, _, err = d2.Version(ctx)
version, _, err = d2.CurrentState(ctx)
if err != nil {
t.Fatal(err)
}
@@ -238,8 +237,8 @@ func TestWithSchema(t *testing.T) {
t.Fatal("expected Version 2")
}
// meanwhile, the public schema still has the other Version
version, _, err = d.Version(ctx)
// meanwhile, the public schema still has the other CurrentState
version, _, err = d.CurrentState(ctx)
if err != nil {
t.Fatal(err)
}
@@ -287,49 +286,6 @@ func TestPostgres_Lock(t *testing.T) {
})
}
func TestWithInstance_Concurrent(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ip, port, err := c.FirstPort()
if err != nil {
t.Fatal(err)
}
// The number of concurrent processes running New
const concurrency = 30
// We can instantiate a single database handle because it is
// actually a connection pool, and so, each of the below go
// routines will have a high probability of using a separate
// connection, which is something we want to exercise.
db, err := sql.Open("postgres", pgConnectionString(ip, port))
if err != nil {
t.Fatal(err)
}
defer func() {
if err := db.Close(); err != nil {
t.Error(err)
}
}()
db.SetMaxIdleConns(concurrency)
db.SetMaxOpenConns(concurrency)
var wg sync.WaitGroup
defer wg.Wait()
wg.Add(concurrency)
for i := 0; i < concurrency; i++ {
go func(i int) {
defer wg.Done()
_, err := New(context.Background(), db)
if err != nil {
t.Errorf("process %d error: %s", i, err)
}
}(i)
}
})
}
func TestRun_Error(t *testing.T) {
dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
ctx := context.Background()
internal/db/schema/postgres/testing.go
+3 -3
View File
@@ -49,14 +49,14 @@ func test(t *testing.T, d *Postgres, migration []byte) {
testNilVersion(t, d) // test first
testLockAndUnlock(t, d)
testRun(t, d, bytes.NewReader(migration))
testSetVersion(t, d) // also tests Version()
testSetVersion(t, d) // also tests CurrentState()
// drop breaks the driver, so test it last.
testDrop(t, d)
}
func testNilVersion(t *testing.T, d *Postgres) {
ctx := context.Background()
v, _, err := d.Version(ctx)
v, _, err := d.CurrentState(ctx)
if err != nil {
t.Fatal(err)
}
@@ -136,7 +136,7 @@ func testSetVersion(t *testing.T, d *Postgres) {
if err != tc.expectedErr {
t.Fatal("Got unexpected error:", err, "!=", tc.expectedErr)
}
v, dirty, readErr := d.Version(ctx)
v, dirty, readErr := d.CurrentState(ctx)
if readErr != tc.expectedReadErr {
t.Fatal("Got unexpected error:", readErr, "!=", tc.expectedReadErr)
}
internal/db/schema/postgres_migration.gen.go
-365
View File
@@ -4988,371 +4988,6 @@ create table server_tag (
primary key(server_id, key, value)
);
commit;
`),
},
downMigrations: map[int][]byte{
1: []byte(`
begin;
drop domain wt_timestamp;
drop domain wt_public_id;
drop domain wt_private_id;
drop domain wt_scope_id;
drop domain wt_user_id;
drop domain wt_version;
drop function default_create_time;
drop function update_time_column;
drop function update_version_column;
drop function immutable_columns;
commit;
`),
2: []byte(`
begin;
drop table oplog_metadata cascade;
drop table oplog_ticket cascade;
drop table oplog_entry cascade;
commit;
`),
3: []byte(`
begin;
drop table db_test_rental cascade;
drop table db_test_car cascade;
drop table db_test_user cascade;
drop table db_test_scooter cascade;
commit;
`),
6: []byte(`
BEGIN;
drop table iam_group cascade;
drop table iam_user cascade;
drop table iam_scope_project cascade;
drop table iam_scope_org cascade;
drop table iam_scope_global cascade;
drop table iam_scope cascade;
drop table iam_scope_type_enm cascade;
drop table iam_role cascade;
drop view iam_principal_role cascade;
drop table iam_group_role cascade;
drop table iam_user_role cascade;
drop table iam_group_member_user cascade;
drop view iam_group_member cascade;
drop table iam_role_grant cascade;
drop function iam_sub_names cascade;
drop function iam_immutable_scope_type_func cascade;
drop function iam_sub_scopes_func cascade;
drop function iam_immutable_role_principal cascade;
drop function iam_user_role_scope_check cascade;
drop function iam_group_role_scope_check cascade;
drop function iam_group_member_scope_check cascade;
drop function iam_immutable_group_member cascade;
drop function get_scoped_member_id cascade;
drop function grant_scope_id_valid cascade;
drop function disallow_global_scope_deletion cascade;
drop function user_scope_id_valid cascade;
drop function iam_immutable_role_grant cascade;
drop function disallow_iam_predefined_user_deletion cascade;
drop function recovery_user_not_allowed cascade;
COMMIT;
`),
7: []byte(`
begin;
drop function update_iam_user_auth_account;
drop function insert_auth_account_subtype;
drop function insert_auth_method_subtype;
drop table auth_account cascade;
drop table auth_method cascade;
commit;
`),
8: []byte(`
begin;
drop table server;
commit;
`),
11: []byte(`
begin;
drop view auth_token_account cascade;
drop table auth_token cascade;
drop function update_last_access_time cascade;
drop function immutable_auth_token_columns cascade;
drop function expire_time_not_older_than_token cascade;
commit;
`),
12: []byte(`
begin;
drop table auth_password_credential;
drop table auth_password_conf cascade;
drop table if exists auth_password_account;
drop table if exists auth_password_method;
drop function insert_auth_password_credential_subtype;
drop function insert_auth_password_conf_subtype;
commit;
`),
13: []byte(`
begin;
drop table auth_password_argon2_cred;
drop table auth_password_argon2_conf;
commit;
`),
14: []byte(`
begin;
drop view auth_password_current_conf;
drop view auth_password_conf_union;
commit;
`),
20: []byte(`
begin;
drop table host_set;
drop table host;
drop table host_catalog;
drop function insert_host_set_subtype;
drop function insert_host_subtype;
drop function insert_host_catalog_subtype;
delete
from oplog_ticket
where name in (
'host_catalog',
'host',
'host_set'
);
commit;
`),
22: []byte(`
begin;
drop table static_host_set_member cascade;
drop table static_host_set cascade;
drop table static_host cascade;
drop table static_host_catalog cascade;
delete
from oplog_ticket
where name in (
'static_host_catalog',
'static_host',
'static_host_set',
'static_host_set_member'
);
commit;
`),
30: []byte(`
begin;
drop function kms_version_column cascade;
commit;
`),
31: []byte(`
begin;
drop table kms_root_key cascade;
drop table kms_root_key_version cascade;
drop table kms_database_key cascade;
drop table kms_database_key_version cascade;
drop table kms_oplog_key cascade;
drop table kms_oplog_key_version cascade;
drop table kms_session_key cascade;
drop table kms_session_key_version cascade;
commit;
`),
40: []byte(`
begin;
drop function insert_target_subtype;
drop function delete_target_subtype;
drop function target_scope_valid;
drop function target_host_set_scope_valid
commit;
`),
41: []byte(`
begin;
drop table target cascade;
drop table target_host_set cascade;
drop table target_tcp;
drop view target_all_subtypes;
drop view target_host_set_catalog;
delete
from oplog_ticket
where name in (
'target_tcp'
);
commit;
`),
50: []byte(`
begin;
drop table session_state;
drop table session_state_enm;
drop table session;
drop table session_termination_reason_enm;
drop function insert_session_state;
drop function insert_new_session_state;
drop function insert_session;
drop function update_session_state_on_termination_reason;
drop function insert_session_state;
delete
from oplog_ticket
where name in (
'session'
);
commit;
`),
51: []byte(`
begin;
drop table session_connection_state;
drop table session_connection_state_enm;
drop table session_connection;
drop table session_connection_closed_reason_enm;
drop function insert_session_connection_state;
drop function insert_new_connection_state;
drop function update_connection_state_on_closed_reason;
commit;
`),
60: []byte(`
begin;
drop function wh_current_time_id;
drop function wh_current_date_id;
drop function wh_time_id;
drop function wh_date_id;
drop domain wh_dim_text;
drop domain wh_timestamp;
drop domain wh_public_id;
drop domain wh_dim_id;
drop function wh_dim_id;
drop domain wh_bytes_transmitted;
drop domain wh_inet_port;
commit;
`),
62: []byte(`
begin;
drop table wh_time_of_day_dimension;
drop table wh_date_dimension;
commit;
`),
65: []byte(`
begin;
drop view whx_user_dimension_target;
drop view whx_user_dimension_source;
drop view whx_host_dimension_target;
drop view whx_host_dimension_source;
drop table wh_user_dimension;
drop table wh_host_dimension;
commit;
`),
66: []byte(`
begin;
drop function wh_upsert_user;
drop function wh_upsert_host;
commit;
`),
68: []byte(`
begin;
drop table wh_session_connection_accumulating_fact;
drop table wh_session_accumulating_fact;
commit;
`),
69: []byte(`
begin;
drop trigger wh_insert_session_connection_state on session_connection_state;
drop function wh_insert_session_connection_state;
drop trigger wh_insert_session_state on session_state;
drop function wh_insert_session_state;
drop trigger wh_update_session_connection on session_connection;
drop function wh_update_session_connection;
drop trigger wh_insert_session_connection on session_connection;
drop function wh_insert_session_connection;
drop trigger wh_insert_session on session;
drop function wh_insert_session;
drop function wh_rollup_connections;
commit;
`),
1001: []byte(`
begin;
drop domain wt_bexprfilter;
drop table server_tag;
drop domain wt_tagpair;
commit;
`),
},
internal/db/schema/schema.go
+7 -4
View File
@@ -7,10 +7,12 @@ import (
"github.com/hashicorp/boundary/internal/errors"
)
// InitStore executes the migrations needed to initialize the store. It
// returns true if migrations actually ran; false if the database is already current.
func InitStore(ctx context.Context, dialect string, url string) (bool, error) {
const op = "schema.InitStore"
// MigrateStore executes the migrations needed to initialize the store. It
// returns true if migrations actually ran; false if the database is already current
// or if there was an error.
func MigrateStore(ctx context.Context, dialect string, url string) (bool, error) {
const op = "schema.MigrateStore"
d, err := sql.Open(dialect, url)
if err != nil {
return false, errors.Wrap(err, op)
@@ -36,5 +38,6 @@ func InitStore(ctx context.Context, dialect string, url string) (bool, error) {
if err := sMan.RollForward(ctx); err != nil {
return false, errors.Wrap(err, op)
}
return true, nil
}
internal/db/schema/schema_test.go
+7 -7
View File
@@ -10,7 +10,7 @@ import (
"github.com/stretchr/testify/require"
)
func TestInitStore(t *testing.T) {
func TestMigrateStore(t *testing.T) {
dialect := "postgres"
ctx := context.Background()
@@ -25,25 +25,25 @@ func TestInitStore(t *testing.T) {
nState := createPartialMigrationState(oState, 8)
migrationStates[dialect] = nState
ran, err := InitStore(ctx, dialect, u)
ran, err := MigrateStore(ctx, dialect, u)
assert.NoError(t, err)
assert.True(t, ran)
ran, err = InitStore(ctx, dialect, u)
ran, err = MigrateStore(ctx, dialect, u)
assert.NoError(t, err)
assert.False(t, ran)
// Reset the possible migration state to contain everything
migrationStates[dialect] = oState
ran, err = InitStore(ctx, dialect, u)
ran, err = MigrateStore(ctx, dialect, u)
assert.NoError(t, err)
assert.True(t, ran)
ran, err = InitStore(ctx, dialect, u)
ran, err = MigrateStore(ctx, dialect, u)
assert.NoError(t, err)
assert.False(t, ran)
}
func TestInitStore_Dirty(t *testing.T) {
func TestMigrateStore_Dirty(t *testing.T) {
dialect := "postgres"
ctx := context.Background()
@@ -59,7 +59,7 @@ func TestInitStore_Dirty(t *testing.T) {
m, err := NewManager(ctx, dialect, db)
m.driver.SetVersion(ctx, -1, true)
b, err := InitStore(ctx, dialect, u)
b, err := MigrateStore(ctx, dialect, u)
assert.Error(t, err)
assert.False(t, b)
}
internal/db/schema/state.go
+2 -11
View File
@@ -6,7 +6,7 @@ const nilVersion = -1
// contains the internal representation of a schema in the current binary.
type migrationState struct {
// devMigration is true if the database schema that would be applied by
// InitStore would be from files in the /dev directory which indicates it would
// MigrateStore would be from files in the /dev directory which indicates it would
// not be safe to run in a non dev environment.
devMigration bool
@@ -14,8 +14,7 @@ type migrationState struct {
// this binary.
binarySchemaVersion int
upMigrations map[int][]byte
downMigrations map[int][]byte
upMigrations map[int][]byte
}
// migrationStates is populated by the generated migration code with the key being the dialect.
@@ -29,14 +28,6 @@ func getUpMigration(dialect string) map[int][]byte {
return ms.upMigrations
}
func getDownMigration(dialect string) map[int][]byte {
ms, ok := migrationStates[dialect]
if !ok {
return nil
}
return ms.downMigrations
}
// DevMigration returns true iff the provided dialect has changes which are still in development.
func DevMigration(dialect string) bool {
ms, ok := migrationStates[dialect]
internal/db/schema/statement.go
+5 -14
View File
@@ -1,10 +1,7 @@
package schema
import (
"fmt"
"sort"
"github.com/hashicorp/boundary/internal/errors"
)
// statementProvider provides the migration statements in order.
@@ -13,20 +10,14 @@ import (
type statementProvider struct {
pos int
versions []int
up, down map[int][]byte
up map[int][]byte
}
func newStatementProvider(dialect string, curVer int) (*statementProvider, error) {
op := errors.Op("schema.newStatementProvider")
func newStatementProvider(dialect string, curVer int) *statementProvider {
const op = "schema.newStatementProvider"
qp := statementProvider{pos: -1}
qp.up, qp.down = getUpMigration(dialect), getDownMigration(dialect)
if len(qp.up) != len(qp.down) {
return nil, errors.New(errors.MigrationIntegrity, op, fmt.Sprintf("Mismatch up/down size: up %d vs. down %d", len(qp.up), len(qp.down)))
}
qp.up = getUpMigration(dialect)
for k := range qp.up {
if _, ok := qp.down[k]; !ok {
return nil, errors.New(errors.MigrationIntegrity, op, fmt.Sprintf("Up key %d doesn't exist in down %v", k, qp.down))
}
qp.versions = append(qp.versions, k)
}
sort.Ints(qp.versions)
@@ -35,7 +26,7 @@ func newStatementProvider(dialect string, curVer int) (*statementProvider, error
qp.versions = qp.versions[1:]
}
return &qp, nil
return &qp
}
func (q *statementProvider) Next() bool {
internal/db/schema/statement_test.go
+2 -47
View File
@@ -15,15 +15,9 @@ func TestStatementProvider(t *testing.T) {
2: []byte("two"),
3: []byte("three"),
},
downMigrations: map[int][]byte{
1: []byte("down one"),
2: []byte("down two"),
3: []byte("down three"),
},
}
st, err := newStatementProvider(testDialect, 1)
assert.NoError(t, err)
st := newStatementProvider(testDialect, 1)
assert.Equal(t, -1, st.Version())
assert.Equal(t, []byte(nil), st.ReadUp())
@@ -43,45 +37,6 @@ func TestStatementProvider(t *testing.T) {
assert.Equal(t, -1, st.Version())
assert.Equal(t, []byte(nil), st.ReadUp())
st, err = newStatementProvider("unknown_dialect", nilVersion)
assert.NoError(t, err)
st = newStatementProvider("unknown_dialect", nilVersion)
assert.False(t, st.Next())
}
func TestStatementProvider_error(t *testing.T) {
cases := []struct {
name string
in migrationState
}{
{
name: "mismatchLength",
in: migrationState{
binarySchemaVersion: 5,
upMigrations: map[int][]byte{
1: []byte("one"),
},
downMigrations: map[int][]byte{},
},
},
{
name: "mismatchVersions",
in: migrationState{
binarySchemaVersion: 5,
upMigrations: map[int][]byte{
1: []byte("one"),
},
downMigrations: map[int][]byte{
2: []byte("two"),
},
},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
migrationStates[tc.name] = tc.in
defer delete(migrationStates, tc.name)
_, err := newStatementProvider(tc.name, -1)
assert.Error(t, err)
})
}
}
internal/db/testing.go
+1 -1
View File
@@ -41,7 +41,7 @@ func TestSetup(t *testing.T, dialect string, opt ...TestOption) (*gorm.DB, strin
url = opts.withTestDatabaseUrl
}
_, err = schema.InitStore(ctx, dialect, url)
_, err = schema.MigrateStore(ctx, dialect, url)
if err != nil {
t.Fatalf("Couldn't init store on existing db: %v", err)
}
internal/oplog/README.md
+1 -1
View File
@@ -68,7 +68,7 @@ tx.Commit()
## TBD/TODO
We need to discuss and decide how Boundary is going to handle the following oplog things:
* SQL migrations: you'll find the package's [SQL migrations](https://github.com/golang-migrate/migrate) under: ./migrations/postgres We need to decide how Boundary will manage migrations across the system and we will likely need to reference this package's migrations somehow.
* SQL migrations: you'll find the package's SQL migrations under: ./db/schema We need to decide how Boundary will manage migrations across the system and we will likely need to reference this package's migrations somehow.
## oplog entry
```
internal/servers/controller/testing.go
+1 -1
View File
@@ -427,7 +427,7 @@ func NewTestController(t *testing.T, opts *TestControllerOpts) *TestController {
if opts.DatabaseUrl != "" {
tc.b.DatabaseUrl = opts.DatabaseUrl
if _, err := schema.InitStore(ctx, "postgres", tc.b.DatabaseUrl); err != nil {
if _, err := schema.MigrateStore(ctx, "postgres", tc.b.DatabaseUrl); err != nil {
t.Fatal(err)
}
if err := tc.b.ConnectToDatabase("postgres"); err != nil {