diff --git a/server/channels/store/sqlstore/migrate.go b/server/channels/store/sqlstore/migrate.go new file mode 100644 index 0000000000..6f1c0955e1 --- /dev/null +++ b/server/channels/store/sqlstore/migrate.go @@ -0,0 +1,229 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package sqlstore + +import ( + "context" + "fmt" + "log" + "path" + "sort" + "strconv" + "sync" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/v8/channels/db" + "github.com/mattermost/morph" + "github.com/mattermost/morph/drivers" + ms "github.com/mattermost/morph/drivers/mysql" + ps "github.com/mattermost/morph/drivers/postgres" + "github.com/mattermost/morph/models" + mbindata "github.com/mattermost/morph/sources/embedded" +) + +type Migrator struct { + engine *morph.Morph + store *SqlStore +} + +func NewMigrator(settings model.SqlSettings, dryRun bool) (*Migrator, error) { + ss := &SqlStore{ + rrCounter: 0, + srCounter: 0, + settings: &settings, + quitMonitor: make(chan struct{}), + wgMonitor: &sync.WaitGroup{}, + } + + ss.initConnection() + + ver, err := ss.GetDbVersion(true) + if err != nil { + return nil, fmt.Errorf("error while getting DB version: %w", err) + } + + ok, err := ss.ensureMinimumDBVersion(ver) + if !ok { + return nil, fmt.Errorf("error while checking DB version: %w", err) + } + + err = ss.ensureDatabaseCollation() + if err != nil { + return nil, fmt.Errorf("error while checking DB collation: %w", err) + } + + engine, err := ss.initMorph(dryRun) + if err != nil { + return nil, fmt.Errorf("failed to initialize morph: %w", err) + } + + return &Migrator{ + engine: engine, + store: ss, + }, nil +} + +func (m *Migrator) Close() error { + if err := m.engine.Close(); err != nil { + return fmt.Errorf("failed to close morph engine: %w", err) + } + + m.store.Close() + + return nil +} + +func (m *Migrator) GetFileName(plan *models.Plan) (string, error) { + if len(plan.Migrations) == 0 { + return "", fmt.Errorf("plan is empty") + } + + to := plan.Migrations[len(plan.Migrations)-1].Version + from, err := m.store.GetDBSchemaVersion() + if err != nil { + return "", err + } + + return fmt.Sprintf("migration_plan_%d_%d", from, to), nil +} + +func (ss *SqlStore) initMorph(dryRun bool) (*morph.Morph, error) { + assets := db.Assets() + + assetsList, err := assets.ReadDir(path.Join("migrations", ss.DriverName())) + if err != nil { + return nil, err + } + + assetNamesForDriver := make([]string, len(assetsList)) + for i, entry := range assetsList { + assetNamesForDriver[i] = entry.Name() + } + + src, err := mbindata.WithInstance(&mbindata.AssetSource{ + Names: assetNamesForDriver, + AssetFunc: func(name string) ([]byte, error) { + return assets.ReadFile(path.Join("migrations", ss.DriverName(), name)) + }, + }) + if err != nil { + return nil, err + } + + var driver drivers.Driver + switch ss.DriverName() { + case model.DatabaseDriverMysql: + dataSource, rErr := ResetReadTimeout(*ss.settings.DataSource) + if rErr != nil { + mlog.Fatal("Failed to reset read timeout from datasource.", mlog.Err(rErr), mlog.String("src", *ss.settings.DataSource)) + return nil, rErr + } + dataSource, err = AppendMultipleStatementsFlag(dataSource) + if err != nil { + return nil, err + } + db, err2 := SetupConnection("master", dataSource, ss.settings, DBPingAttempts) + if err2 != nil { + return nil, err2 + } + + driver, err = ms.WithInstance(db) + if err != nil { + return nil, err + } + defer db.Close() + case model.DatabaseDriverPostgres: + driver, err = ps.WithInstance(ss.GetMasterX().DB.DB) + default: + err = fmt.Errorf("unsupported database type %s for migration", ss.DriverName()) + } + if err != nil { + return nil, err + } + + opts := []morph.EngineOption{ + morph.WithLogger(log.New(&morphWriter{}, "", log.Lshortfile)), + morph.WithLock("mm-lock-key"), + morph.SetStatementTimeoutInSeconds(*ss.settings.MigrationsStatementTimeoutSeconds), + morph.SetDryRun(dryRun), + } + + engine, err := morph.New(context.Background(), driver, src, opts...) + if err != nil { + return nil, err + } + + return engine, nil +} + +func (ss *SqlStore) migrate(direction migrationDirection, dryRun bool) error { + engine, err := ss.initMorph(dryRun) + if err != nil { + return err + } + defer engine.Close() + + switch direction { + case migrationsDirectionDown: + _, err = engine.ApplyDown(-1) + return err + default: + return engine.ApplyAll() + } +} + +func (m *Migrator) GeneratePlan(recover bool) (*models.Plan, error) { + diff, err := m.engine.Diff(models.Up) + if err != nil { + return nil, err + } + + plan, err := m.engine.GeneratePlan(diff, recover) + if err != nil { + return nil, err + } + + return plan, nil +} + +// MigrateWithPlan migrates the database to the latest version using the provided plan. +func (m *Migrator) MigrateWithPlan(plan *models.Plan, dryRun bool) error { + return m.engine.ApplyPlan(plan) +} + +func (m *Migrator) DowngradeMigrations(dryRun bool, versions ...string) error { + migrations, err := m.engine.Diff(models.Down) + if err != nil { + return err + } + + migrationsToDowngrade := make([]*models.Migration, 0, len(versions)) + for _, version := range versions { + for _, migration := range migrations { + versionNumber, sErr := strconv.Atoi(version) + if sErr != nil { + return sErr + } + if migration.Version == uint32(versionNumber) { + migrationsToDowngrade = append(migrationsToDowngrade, migration) + } + } + } + + sort.Slice(migrationsToDowngrade, func(i, j int) bool { + return migrationsToDowngrade[i].Version > migrationsToDowngrade[j].Version + }) + + if len(migrationsToDowngrade) != len(versions) { + mlog.Warn("could not match give migration versions, going to downgrade only the migrations those are available.") + } + + plan, err := m.engine.GeneratePlan(migrationsToDowngrade, false) + if err != nil { + return err + } + + return m.engine.ApplyPlan(plan) +} diff --git a/server/channels/store/sqlstore/migrate_test.go b/server/channels/store/sqlstore/migrate_test.go new file mode 100644 index 0000000000..b5749c3e8f --- /dev/null +++ b/server/channels/store/sqlstore/migrate_test.go @@ -0,0 +1,33 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package sqlstore + +import ( + "testing" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/stretchr/testify/assert" +) + +func TestUpAndDownMigrations(t *testing.T) { + testDrivers := []string{ + model.DatabaseDriverPostgres, + model.DatabaseDriverMysql, + } + + for _, driver := range testDrivers { + t.Run("Should be reversible for "+driver, func(t *testing.T) { + settings, err := makeSqlSettings(driver) + if err != nil { + t.Skip(err) + } + + store := New(*settings, nil) + defer store.Close() + + err = store.migrate(migrationsDirectionDown, false) + assert.NoError(t, err, "downing migrations should not error") + }) + } +} diff --git a/server/channels/store/sqlstore/store.go b/server/channels/store/sqlstore/store.go index e5cd249f15..3ea68e4605 100644 --- a/server/channels/store/sqlstore/store.go +++ b/server/channels/store/sqlstore/store.go @@ -8,31 +8,22 @@ import ( "database/sql" dbsql "database/sql" "fmt" - "log" - "path" "strconv" "strings" "sync" "sync/atomic" "time" - "github.com/mattermost/morph" sq "github.com/mattermost/squirrel" - "github.com/mattermost/morph/drivers" - ms "github.com/mattermost/morph/drivers/mysql" - ps "github.com/mattermost/morph/drivers/postgres" - "github.com/go-sql-driver/mysql" _ "github.com/golang-migrate/migrate/v4/source/file" "github.com/jmoiron/sqlx" "github.com/lib/pq" - mbindata "github.com/mattermost/morph/sources/embedded" "github.com/pkg/errors" "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/public/shared/mlog" - "github.com/mattermost/mattermost/server/v8/channels/db" "github.com/mattermost/mattermost/server/v8/channels/store" "github.com/mattermost/mattermost/server/v8/einterfaces" ) @@ -177,7 +168,7 @@ func New(settings model.SqlSettings, metrics einterfaces.MetricsInterface) *SqlS mlog.Fatal("Error while checking DB collation.", mlog.Err(err)) } - err = store.migrate(migrationsDirectionUp) + err = store.migrate(migrationsDirectionUp, false) if err != nil { mlog.Fatal("Failed to apply database migrations.", mlog.Err(err)) } @@ -1191,76 +1182,6 @@ func (ss *SqlStore) hasLicense() bool { return hasLicense } -func (ss *SqlStore) migrate(direction migrationDirection) error { - assets := db.Assets() - - assetsList, err := assets.ReadDir(path.Join("migrations", ss.DriverName())) - if err != nil { - return err - } - - assetNamesForDriver := make([]string, len(assetsList)) - for i, entry := range assetsList { - assetNamesForDriver[i] = entry.Name() - } - - src, err := mbindata.WithInstance(&mbindata.AssetSource{ - Names: assetNamesForDriver, - AssetFunc: func(name string) ([]byte, error) { - return assets.ReadFile(path.Join("migrations", ss.DriverName(), name)) - }, - }) - if err != nil { - return err - } - - var driver drivers.Driver - switch ss.DriverName() { - case model.DatabaseDriverMysql: - dataSource, rErr := ResetReadTimeout(*ss.settings.DataSource) - if rErr != nil { - mlog.Fatal("Failed to reset read timeout from datasource.", mlog.Err(rErr), mlog.String("src", *ss.settings.DataSource)) - return rErr - } - dataSource, err = AppendMultipleStatementsFlag(dataSource) - if err != nil { - return err - } - db, err2 := SetupConnection("master", dataSource, ss.settings, DBPingAttempts) - if err2 != nil { - return err2 - } - driver, err = ms.WithInstance(db) - defer db.Close() - case model.DatabaseDriverPostgres: - driver, err = ps.WithInstance(ss.GetMasterX().DB.DB) - default: - err = fmt.Errorf("unsupported database type %s for migration", ss.DriverName()) - } - if err != nil { - return err - } - - opts := []morph.EngineOption{ - morph.WithLogger(log.New(&morphWriter{}, "", log.Lshortfile)), - morph.WithLock("mm-lock-key"), - morph.SetStatementTimeoutInSeconds(*ss.settings.MigrationsStatementTimeoutSeconds), - } - engine, err := morph.New(context.Background(), driver, src, opts...) - if err != nil { - return err - } - defer engine.Close() - - switch direction { - case migrationsDirectionDown: - _, err = engine.ApplyDown(-1) - return err - default: - return engine.ApplyAll() - } -} - func convertMySQLFullTextColumnsToPostgres(columnNames string) string { columns := strings.Split(columnNames, ", ") concatenatedColumnNames := "" diff --git a/server/channels/store/sqlstore/store_test.go b/server/channels/store/sqlstore/store_test.go index e591a6a144..8e319d4abd 100644 --- a/server/channels/store/sqlstore/store_test.go +++ b/server/channels/store/sqlstore/store_test.go @@ -558,27 +558,6 @@ func TestIsBinaryParamEnabled(t *testing.T) { } -func TestUpAndDownMigrations(t *testing.T) { - testDrivers := []string{ - model.DatabaseDriverPostgres, - model.DatabaseDriverMysql, - } - - for _, driver := range testDrivers { - t.Run("Should be reversible for "+driver, func(t *testing.T) { - settings, err := makeSqlSettings(driver) - if err != nil { - t.Skip(err) - } - store := New(*settings, nil) - defer store.Close() - - err = store.migrate(migrationsDirectionDown) - assert.NoError(t, err, "downing migrations should not error") - }) - } -} - func TestGetAllConns(t *testing.T) { t.Parallel() testCases := []struct { @@ -777,7 +756,7 @@ func TestReplicaLagQuery(t *testing.T) { require.NoError(t, store.initConnection()) store.stores.post = newSqlPostStore(store, mockMetrics) - err = store.migrate(migrationsDirectionUp) + err = store.migrate(migrationsDirectionUp, false) require.NoError(t, err) defer store.Close() diff --git a/server/cmd/mattermost/commands/db.go b/server/cmd/mattermost/commands/db.go index 6e8801b33c..5ec5bc7d5e 100644 --- a/server/cmd/mattermost/commands/db.go +++ b/server/cmd/mattermost/commands/db.go @@ -4,16 +4,23 @@ package commands import ( + "bytes" + "encoding/json" "fmt" "strconv" + "strings" "github.com/pkg/errors" "github.com/spf13/cobra" + "github.com/mattermost/mattermost/server/public/model" "github.com/mattermost/mattermost/server/v8/channels/app" "github.com/mattermost/mattermost/server/v8/channels/audit" "github.com/mattermost/mattermost/server/v8/channels/store/sqlstore" "github.com/mattermost/mattermost/server/v8/config" + "github.com/mattermost/mattermost/server/v8/platform/shared/filestore" + "github.com/mattermost/morph" + "github.com/mattermost/morph/models" ) var DbCmd = &cobra.Command{ @@ -53,6 +60,15 @@ var MigrateCmd = &cobra.Command{ RunE: migrateCmdF, } +var DowngradeCmd = &cobra.Command{ + Use: "downgrade", + Short: "Downgrade the database with the given plan or migration numbers", + Long: "Downgrade the database with the given plan or migration numbers. " + + "The plan will be read from filestore hence the path should be relative to file store root.", + RunE: downgradeCmdF, + Args: cobra.ExactArgs(1), +} + var DBVersionCmd = &cobra.Command{ Use: "version", Short: "Returns the recent applied version number", @@ -62,11 +78,18 @@ var DBVersionCmd = &cobra.Command{ func init() { ResetCmd.Flags().Bool("confirm", false, "Confirm you really want to delete everything and a DB backup has been performed.") DBVersionCmd.Flags().Bool("all", false, "Returns all applied migrations") + MigrateCmd.Flags().Bool("auto-recover", false, "Recover the database to it's existing state after a failed migration.") + MigrateCmd.Flags().Bool("save-plan", false, "Saves the migration plan into file store so that it can be used in the future.") + MigrateCmd.Flags().Bool("dry-run", false, "Runs the migration plan without applying it.") + + DowngradeCmd.Flags().Bool("auto-recover", false, "Recover the database to it's existing state after a failed migration.") + DowngradeCmd.Flags().Bool("dry-run", false, "Runs the migration plan without applying it.") DbCmd.AddCommand( InitDbCmd, ResetCmd, MigrateCmd, + DowngradeCmd, DBVersionCmd, ) @@ -95,7 +118,7 @@ func initDbCmdF(command *cobra.Command, _ []string) error { sqlStore := sqlstore.New(configStore.Get().SqlSettings, nil) defer sqlStore.Close() - fmt.Println("Database store correctly initialised") + CommandPrettyPrintln("Database store correctly initialised") return nil } @@ -134,20 +157,129 @@ func resetCmdF(command *cobra.Command, args []string) error { func migrateCmdF(command *cobra.Command, args []string) error { cfgDSN := getConfigDSN(command, config.GetEnvironment()) + recoverFlag, _ := command.Flags().GetBool("auto-recover") + savePlan, _ := command.Flags().GetBool("save-plan") + dryRun, _ := command.Flags().GetBool("dry-run") cfgStore, err := config.NewStoreFromDSN(cfgDSN, true, nil, true) if err != nil { return errors.Wrap(err, "failed to load configuration") } config := cfgStore.Get() - store := sqlstore.New(config.SqlSettings, nil) - defer store.Close() + migrator, err := sqlstore.NewMigrator(config.SqlSettings, dryRun) + if err != nil { + return errors.Wrap(err, "failed to create migrator") + } + defer migrator.Close() + + plan, err := migrator.GeneratePlan(recoverFlag) + if err != nil { + return errors.Wrap(err, "failed to generate migration plan") + } + + if len(plan.Migrations) == 0 { + CommandPrettyPrintln("No migrations to apply.") + return nil + } + + if savePlan || recoverFlag { + backend, err2 := filestore.NewFileBackend(ConfigToFileBackendSettings(&config.FileSettings, false, true)) + if err2 != nil { + return fmt.Errorf("failed to initialize filebackend: %w", err2) + } + + b, mErr := json.MarshalIndent(plan, "", " ") + if mErr != nil { + return fmt.Errorf("failed to marshal plan: %w", mErr) + } + + fileName, err2 := migrator.GetFileName(plan) + if err2 != nil { + return fmt.Errorf("failed to generate plan file: %w", err2) + } + + _, err = backend.WriteFile(bytes.NewReader(b), fileName+".json") + if err != nil { + return fmt.Errorf("failed to write migration plan: %w", err) + } + + CommandPrettyPrintln( + fmt.Sprintf("%s\nThe migration plan has been saved. File: %q.\nNote that "+ + " migration plan is saved into file store, so the filepath will be relative to root of file store\n%s", + strings.Repeat("*", 80), fileName+".json", strings.Repeat("*", 80))) + } + + err = migrator.MigrateWithPlan(plan, dryRun) + if err != nil { + return errors.Wrap(err, "failed to migrate with the plan") + } CommandPrettyPrintln("Database successfully migrated") return nil } +func downgradeCmdF(command *cobra.Command, args []string) error { + cfgDSN := getConfigDSN(command, config.GetEnvironment()) + cfgStore, err := config.NewStoreFromDSN(cfgDSN, true, nil, true) + if err != nil { + return errors.Wrap(err, "failed to load configuration") + } + config := cfgStore.Get() + + dryRun, _ := command.Flags().GetBool("dry-run") + recoverFlag, _ := command.Flags().GetBool("auto-recover") + + backend, err2 := filestore.NewFileBackend(ConfigToFileBackendSettings(&config.FileSettings, false, true)) + if err2 != nil { + return fmt.Errorf("failed to initialize filebackend: %w", err2) + } + + migrator, err := sqlstore.NewMigrator(config.SqlSettings, dryRun) + if err != nil { + return errors.Wrap(err, "failed to create migrator") + } + defer migrator.Close() + + // check if the input is version numbers or a file + // if the input is given as a file, we assume it's a migration plan + versions := strings.Split(args[0], ",") + if _, sErr := strconv.Atoi(versions[0]); sErr == nil { + CommandPrettyPrintln("Database will be downgraded with the following versions: ", versions) + + err = migrator.DowngradeMigrations(dryRun, versions...) + if err != nil { + return errors.Wrap(err, "failed to downgrade migrations") + } + + CommandPrettyPrintln("Database successfully downgraded") + return nil + } + + b, err := backend.ReadFile(args[0]) + if err != nil { + return fmt.Errorf("failed to read plan: %w", err) + } + + var plan models.Plan + err = json.Unmarshal(b, &plan) + if err != nil { + return fmt.Errorf("failed to unmarshal plan: %w", err) + } + + morph.SwapPlanDirection(&plan) + plan.Auto = recoverFlag + + err = migrator.MigrateWithPlan(&plan, dryRun) + if err != nil { + return errors.Wrap(err, "failed to migrate with the plan") + } + + CommandPrettyPrintln("Database successfully downgraded") + + return nil +} + func dbVersionCmdF(command *cobra.Command, args []string) error { cfgDSN := getConfigDSN(command, config.GetEnvironment()) cfgStore, err := config.NewStoreFromDSN(cfgDSN, true, nil, true) @@ -175,7 +307,32 @@ func dbVersionCmdF(command *cobra.Command, args []string) error { if err != nil { return errors.Wrap(err, "failed to get schema version") } + CommandPrettyPrintln("Current database schema version is: " + strconv.Itoa(v)) return nil } + +func ConfigToFileBackendSettings(s *model.FileSettings, enableComplianceFeature bool, skipVerify bool) filestore.FileBackendSettings { + if *s.DriverName == model.ImageDriverLocal { + return filestore.FileBackendSettings{ + DriverName: *s.DriverName, + Directory: *s.Directory, + } + } + return filestore.FileBackendSettings{ + DriverName: *s.DriverName, + AmazonS3AccessKeyId: *s.AmazonS3AccessKeyId, + AmazonS3SecretAccessKey: *s.AmazonS3SecretAccessKey, + AmazonS3Bucket: *s.AmazonS3Bucket, + AmazonS3PathPrefix: *s.AmazonS3PathPrefix, + AmazonS3Region: *s.AmazonS3Region, + AmazonS3Endpoint: *s.AmazonS3Endpoint, + AmazonS3SSL: s.AmazonS3SSL == nil || *s.AmazonS3SSL, + AmazonS3SignV2: s.AmazonS3SignV2 != nil && *s.AmazonS3SignV2, + AmazonS3SSE: s.AmazonS3SSE != nil && *s.AmazonS3SSE && enableComplianceFeature, + AmazonS3Trace: s.AmazonS3Trace != nil && *s.AmazonS3Trace, + AmazonS3RequestTimeoutMilliseconds: *s.AmazonS3RequestTimeoutMilliseconds, + SkipVerify: skipVerify, + } +}