[MM-47976] cmd/mattermost/db: add downgrade commands and enable plan saving (#21779)

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)
+}

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")
+		})
+	}
+}

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 := ""

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()

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,
+	}
+}