Merge pull request #1130 from percona/PBM-1514-Improve-resync-performance

Author: veceraj

cmd/pbm-agent/delete.go

@@ -297,7 +297,7 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 		l.Error(err.Error())
 	}
 
-	err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me)
+	err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me, false)
 	if err != nil {
 		l.Error("storage resync: " + err.Error())
 	}

cmd/pbm-agent/resync.go

@@ -71,7 +71,7 @@ func (a *Agent) Resync(ctx context.Context, cmd *ctrl.ResyncCmd, opid ctrl.OPID,
 	} else if cmd.Name != "" {
 		err = a.handleSyncProfile(ctx, cmd.Name, cmd.Clear)
 	} else {
-		err = a.handleSyncMainStorage(ctx)
+		err = a.handleSyncMainStorage(ctx, cmd.SkipRestores)
 	}
 	if err != nil {
 		l.Error(err.Error())
@@ -134,13 +134,13 @@ func (a *Agent) helpSyncProfileBackups(ctx context.Context, profile *config.Conf
 	return errors.Wrapf(err, "sync backup list for %q", profile.Name)
 }
 
-func (a *Agent) handleSyncMainStorage(ctx context.Context) error {
+func (a *Agent) handleSyncMainStorage(ctx context.Context, skipRestores bool) error {
 	cfg, err := config.GetConfig(ctx, a.leadConn)
 	if err != nil {
 		return errors.Wrap(err, "get config")
 	}
 
-	err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me)
+	err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me, skipRestores)
 	if err != nil {
 		return errors.Wrap(err, "resync")
 	}

cmd/pbm/config.go

@@ -18,13 +18,14 @@ import (
 )
 
 type configOpts struct {
-	rsync    bool
-	wait     bool
-	waitTime time.Duration
-	list     bool
-	file     string
-	set      map[string]string
-	key      string
+	rsync        bool
+	skipRestores bool
+	wait         bool
+	waitTime     time.Duration
+	list         bool
+	file         string
+	set          map[string]string
+	key          string
 }
 
 type confKV struct {
@@ -76,7 +77,7 @@ func runConfig(
 			}
 		}
 		if rsnc {
-			if _, err := pbm.SyncFromStorage(ctx); err != nil {
+			if _, err := pbm.SyncFromStorage(ctx, false); err != nil {
 				return nil, errors.Wrap(err, "resync")
 			}
 		}
@@ -91,7 +92,7 @@ func runConfig(
 		}
 		return confKV{c.key, fmt.Sprint(k)}, nil
 	case c.rsync:
-		cid, err := pbm.SyncFromStorage(ctx)
+		cid, err := pbm.SyncFromStorage(ctx, c.skipRestores)
 		if err != nil {
 			return nil, errors.Wrap(err, "resync")
 		}
@@ -143,7 +144,7 @@ func runConfig(
 
 		// resync storage only if Storage options have changed
 		if !reflect.DeepEqual(newCfg.Storage, oldCfg.Storage) {
-			if _, err := pbm.SyncFromStorage(ctx); err != nil {
+			if _, err := pbm.SyncFromStorage(ctx, false); err != nil {
 				return nil, errors.Wrap(err, "resync")
 			}
 		}

cmd/pbm/main.go

@@ -368,6 +368,10 @@ func (app *pbmApp) buildConfigCmd() *cobra.Command {
 		Use:   "config [key]",
 		Short: "Set, change or list the config",
 		RunE: app.wrapRunE(func(cmd *cobra.Command, args []string) (fmt.Stringer, error) {
+			if cfg.skipRestores && !cfg.rsync {
+				return nil, errors.New("--skip-restores cannot be used without --force-resync")
+			}
+
 			if len(args) == 1 {
 				cfg.key = args[0]
 			}
@@ -376,6 +380,9 @@ func (app *pbmApp) buildConfigCmd() *cobra.Command {
 	}
 
 	configCmd.Flags().BoolVar(&cfg.rsync, "force-resync", false, "Resync backup list with the current store")
+	configCmd.Flags().BoolVar(
+		&cfg.skipRestores, "skip-restores", false, "Skip physical restore metadata during force-resync",
+	)
 	configCmd.Flags().BoolVar(&cfg.list, "list", false, "List current settings")
 	configCmd.Flags().StringVar(&cfg.file, "file", "", "Upload config from YAML file")
 	configCmd.Flags().StringToStringVar(&cfg.set, "set", nil, "Set the option value <key.name=value>")

e2e-tests/pkg/pbm/mongo_pbm.go

@@ -71,7 +71,7 @@ func (m *MongoPBM) StoreResync(ctx context.Context) error {
 		return errors.Wrap(err, "get config")
 	}
 
-	return resync.Resync(ctx, m.conn, &cfg.Storage, "")
+	return resync.Resync(ctx, m.conn, &cfg.Storage, "", false)
 }
 
 func (m *MongoPBM) Conn() connect.Client {

pbm/ctrl/cmd.go

@@ -121,9 +121,10 @@ type ProfileCmd struct {
 }
 
 type ResyncCmd struct {
-	Name  string `bson:"name,omitempty"`
-	All   bool   `bson:"all,omitempty"`
-	Clear bool   `bson:"clear,omitempty"`
+	Name         string `bson:"name,omitempty"`
+	All          bool   `bson:"all,omitempty"`
+	Clear        bool   `bson:"clear,omitempty"`
+	SkipRestores bool   `bson:"skipRestores,omitempty"`
 }
 
 type BackupCmd struct {

pbm/ctrl/send.go

@@ -93,37 +93,15 @@ func SendRemoveConfigProfile(ctx context.Context, m connect.Client, name string)
 	return sendCommand(ctx, m, cmd)
 }
 
-func SendResync(ctx context.Context, m connect.Client) (OPID, error) {
-	return sendCommand(ctx, m, Cmd{Cmd: CmdResync})
-}
-
-func SendSyncMetaFrom(ctx context.Context, m connect.Client, name string) (OPID, error) {
-	opts := &ResyncCmd{}
-	if name != "" {
-		opts.Name = name
-	} else {
-		opts.All = true
-	}
-
+func SendResync(ctx context.Context, m connect.Client, opts *ResyncCmd) (OPID, error) {
 	cmd := Cmd{
-		Cmd:    CmdResync,
-		Resync: opts,
+		Cmd: CmdResync,
 	}
-	return sendCommand(ctx, m, cmd)
-}
 
-func SendClearMetaFrom(ctx context.Context, m connect.Client, name string) (OPID, error) {
-	opts := &ResyncCmd{Clear: true}
-	if name != "" {
-		opts.Name = name
-	} else {
-		opts.All = true
+	if opts != nil {
+		cmd.Resync = opts
 	}
 
-	cmd := Cmd{
-		Cmd:    CmdResync,
-		Resync: opts,
-	}
 	return sendCommand(ctx, m, cmd)
 }
 

pbm/resync/rsync.go

@@ -26,7 +26,13 @@ import (
 //
 // It checks for read and write permissions, drops all meta from the database
 // and populate it again by reading meta from the storage.
-func Resync(ctx context.Context, conn connect.Client, cfg *config.StorageConf, node string) error {
+func Resync(
+	ctx context.Context,
+	conn connect.Client,
+	cfg *config.StorageConf,
+	node string,
+	skipRestores bool,
+) error {
 	l := log.LogEventFromContext(ctx)
 
 	stg, err := util.StorageFromConfig(cfg, node, l)
@@ -62,7 +68,7 @@ func Resync(ctx context.Context, conn connect.Client, cfg *config.StorageConf, n
 		l.Error("failed sync oplog range: %v", err)
 	}
 
-	err = resyncPhysicalRestores(ctx, conn, stg)
+	err = resyncPhysicalRestores(ctx, conn, stg, skipRestores)
 	if err != nil {
 		l.Error("failed sync physical restore metadata: %v", err)
 	}
@@ -244,6 +250,7 @@ func resyncPhysicalRestores(
 	ctx context.Context,
 	conn connect.Client,
 	stg storage.Storage,
+	skipRestores bool,
 ) error {
 	_, err := conn.RestoresCollection().DeleteMany(ctx, bson.D{})
 	if err != nil {
@@ -262,7 +269,7 @@ func resyncPhysicalRestores(
 		return nil
 	}
 
-	restoreMeta, err := getAllRestoreMetaFromStorage(ctx, stg)
+	restoreMeta, err := getAllRestoreMetaFromStorage(ctx, stg, skipRestores)
 	if err != nil {
 		return errors.Wrap(err, "get all restore meta from storage")
 	}
@@ -315,6 +322,7 @@ func getAllBackupMetaFromStorage(
 func getAllRestoreMetaFromStorage(
 	ctx context.Context,
 	stg storage.Storage,
+	skipRestores bool,
 ) ([]*restore.RestoreMeta, error) {
 	l := log.LogEventFromContext(ctx)
 
@@ -323,8 +331,23 @@ func getAllRestoreMetaFromStorage(
 		return nil, errors.Wrap(err, "get physical restores list from the storage")
 	}
 
-	rv := make([]*restore.RestoreMeta, 0, len(restoreMeta))
-	for _, file := range restoreMeta {
+	var targets []storage.FileInfo
+
+	if skipRestores && len(restoreMeta) > 0 {
+		l.Debug("only processing last restore")
+		latest := restoreMeta[0]
+		for _, f := range restoreMeta[1:] {
+			if f.Name > latest.Name {
+				latest = f
+			}
+		}
+		targets = []storage.FileInfo{latest}
+	} else {
+		targets = restoreMeta
+	}
+
+	rv := make([]*restore.RestoreMeta, 0, len(targets))
+	for _, file := range targets {
 		filename := strings.TrimSuffix(file.Name, ".json")
 		meta, err := restore.GetPhysRestoreMeta(filename, stg, l)
 		if err != nil {

sdk/impl.go

@@ -262,8 +262,12 @@ func (c *Client) GetRestoreByOpID(ctx context.Context, opid string) (*RestoreMet
 	return restore.GetRestoreMetaByOPID(ctx, c.conn, opid)
 }
 
-func (c *Client) SyncFromStorage(ctx context.Context) (CommandID, error) {
-	opid, err := ctrl.SendResync(ctx, c.conn)
+func (c *Client) SyncFromStorage(ctx context.Context, skipRestores bool) (CommandID, error) {
+	var opts *ctrl.ResyncCmd
+	if skipRestores {
+		opts = &ctrl.ResyncCmd{SkipRestores: true}
+	}
+	opid, err := ctrl.SendResync(ctx, c.conn, opts)
 	return CommandID(opid.String()), err
 }
 
@@ -272,12 +276,12 @@ func (c *Client) SyncFromExternalStorage(ctx context.Context, name string) (Comm
 		return NoOpID, errors.New("name is not provided")
 	}
 
-	opid, err := ctrl.SendSyncMetaFrom(ctx, c.conn, name)
+	opid, err := ctrl.SendResync(ctx, c.conn, &ctrl.ResyncCmd{Name: name})
 	return CommandID(opid.String()), err
 }
 
 func (c *Client) SyncFromAllExternalStorages(ctx context.Context) (CommandID, error) {
-	opid, err := ctrl.SendSyncMetaFrom(ctx, c.conn, "")
+	opid, err := ctrl.SendResync(ctx, c.conn, &ctrl.ResyncCmd{All: true})
 	return CommandID(opid.String()), err
 }
 
@@ -286,12 +290,12 @@ func (c *Client) ClearSyncFromExternalStorage(ctx context.Context, name string)
 		return NoOpID, errors.New("name is not provided")
 	}
 
-	opid, err := ctrl.SendClearMetaFrom(ctx, c.conn, name)
+	opid, err := ctrl.SendResync(ctx, c.conn, &ctrl.ResyncCmd{Name: name, Clear: true})
 	return CommandID(opid.String()), err
 }
 
 func (c *Client) ClearSyncFromAllExternalStorages(ctx context.Context) (CommandID, error) {
-	opid, err := ctrl.SendClearMetaFrom(ctx, c.conn, "")
+	opid, err := ctrl.SendResync(ctx, c.conn, &ctrl.ResyncCmd{All: true, Clear: true})
 	return CommandID(opid.String()), err
 }