[PBM-907] slice oplog during logical backup (#902)

Author: defbin

.github/workflows/reviewdog.yml

@@ -75,7 +75,7 @@ jobs:
         with:
           go-version: "1.19"
       - run: go install golang.org/x/tools/cmd/goimports@latest
-      - run: go install mvdan.cc/gofumpt@latest
+      - run: go install mvdan.cc/gofumpt@v0.5.0
       - run: goimports -w -local "github.com/percona" $(find . -not -path "*/vendor/*" -name "*.go")
       - run: gofumpt -w -extra $(find . -not -path "*/vendor/*" -name "*.go")
       - uses: reviewdog/action-suggester@v1

cmd/pbm-agent/agent.go

@@ -182,7 +182,7 @@ func (a *Agent) Resync(ctx context.Context, opid ctrl.OPID, ep config.Epoch) {
 		Epoch:   &epts,
 	})
 
-	got, err := a.acquireLock(ctx, lock, l, nil)
+	got, err := a.acquireLock(ctx, lock, l)
 	if err != nil {
 		l.Error("acquiring lock: %v", err)
 		return
@@ -194,7 +194,7 @@ func (a *Agent) Resync(ctx context.Context, opid ctrl.OPID, ep config.Epoch) {
 
 	defer func() {
 		if err := lock.Release(); err != nil {
-			l.Error("reslase lock %v: %v", lock, err)
+			l.Error("release lock %v: %v", lock, err)
 		}
 	}()
 
@@ -215,16 +215,10 @@ func (a *Agent) Resync(ctx context.Context, opid ctrl.OPID, ep config.Epoch) {
 	l.Debug("epoch set to %v", epch)
 }
 
-type lockAquireFn func(context.Context) (bool, error)
-
 // acquireLock tries to acquire the lock. If there is a stale lock
 // it tries to mark op that held the lock (backup, [pitr]restore) as failed.
-func (a *Agent) acquireLock(ctx context.Context, l *lock.Lock, lg log.LogEvent, acquireFn lockAquireFn) (bool, error) {
-	if acquireFn == nil {
-		acquireFn = l.Acquire
-	}
-
-	got, err := acquireFn(ctx)
+func (a *Agent) acquireLock(ctx context.Context, l *lock.Lock, lg log.LogEvent) (bool, error) {
+	got, err := l.Acquire(ctx)
 	if err == nil {
 		return got, nil
 	}
@@ -248,14 +242,14 @@ func (a *Agent) acquireLock(ctx context.Context, l *lock.Lock, lg log.LogEvent,
 	case ctrl.CmdRestore:
 		fn = markRestoreStale
 	default:
-		return acquireFn(ctx)
+		return l.Acquire(ctx)
 	}
 
 	if err := fn(ctx, l, lck.OPID); err != nil {
 		lg.Warning("failed to mark stale op '%s' as failed: %v", lck.OPID, err)
 	}
 
-	return acquireFn(ctx)
+	return l.Acquire(ctx)
 }
 
 func (a *Agent) HbPause() {

cmd/pbm-agent/backup.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/percona/percona-backup-mongodb/internal/backup"
 	"github.com/percona/percona-backup-mongodb/internal/config"
 	"github.com/percona/percona-backup-mongodb/internal/ctrl"
@@ -17,36 +19,27 @@ import (
 )
 
 type currentBackup struct {
-	header *ctrl.BackupCmd
 	cancel context.CancelFunc
 }
 
-func (a *Agent) setBcp(b *currentBackup) bool {
+func (a *Agent) setBcp(b *currentBackup) {
 	a.mx.Lock()
 	defer a.mx.Unlock()
-	if a.bcp != nil {
-		return false
-	}
 
 	a.bcp = b
-	return true
-}
-
-func (a *Agent) unsetBcp() {
-	a.mx.Lock()
-	a.bcp = nil
-	a.mx.Unlock()
 }
 
 // CancelBackup cancels current backup
 func (a *Agent) CancelBackup() {
 	a.mx.Lock()
 	defer a.mx.Unlock()
+
 	if a.bcp == nil {
 		return
 	}
 
 	a.bcp.cancel()
+	a.bcp = nil
 }
 
 // Backup starts backup
@@ -67,6 +60,7 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		l.Error("get node info: %v", err)
 		return
 	}
+	// TODO: do the check on the agent start only
 	if nodeInfo.IsStandalone() {
 		l.Error("mongod node can not be used to fetch a consistent backup because it has no oplog. " +
 			"Please restart it as a primary in a single-node replicaset " +
@@ -78,7 +72,7 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 	canRunBackup, err := topo.NodeSuitsExt(ctx, a.nodeConn, nodeInfo, cmd.Type)
 	if err != nil {
 		l.Error("node check: %v", err)
-		if !isClusterLeader {
+		if errors.Is(err, context.Canceled) || !isClusterLeader {
 			return
 		}
 	}
@@ -89,10 +83,8 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		}
 	}
 
-	// wakeup the slicer not to wait for the tick
-	if p := a.getPitr(); p != nil {
-		p.w <- &opid
-	}
+	// wakeup the slicer to not wait for the tick
+	go a.sliceNow(opid)
 
 	var bcp *backup.Backup
 	switch cmd.Type {
@@ -117,6 +109,8 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		l.Error("backups cannot be saved because PBM storage configuration hasn't been set yet")
 		return
 	}
+
+	bcp.SetSlicerInterval(cfg.BackupSlicerInterval())
 	bcp.SetTimeouts(cfg.Backup.Timeouts)
 
 	if isClusterLeader {
@@ -187,21 +181,28 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 			l.Error("get cluster members: %v", err)
 			return
 		}
-		for _, sh := range shards {
-			go func(rs string) {
-				err := a.nominateRS(ctx, cmd.Name, rs, nodes.RS(rs), l)
-				if err != nil {
-					l.Error("nodes nomination for %s: %v", rs, err)
-				}
-			}(sh.RS)
+
+		errGrp, grpCtx := errgroup.WithContext(ctx)
+		for i := range shards {
+			rs := shards[i].RS
+
+			errGrp.Go(func() error {
+				err := a.nominateRS(grpCtx, cmd.Name, rs, nodes.RS(rs))
+				return errors.Wrapf(err, "nodes nomination for %s", rs)
+			})
+		}
+
+		err = errGrp.Wait()
+		if err != nil {
+			l.Error(err.Error())
+			return
 		}
 	}
 
-	nominated, err := a.waitNomination(ctx, cmd.Name, nodeInfo.SetName, nodeInfo.Me, l)
+	nominated, err := a.waitNomination(ctx, cmd.Name, nodeInfo.SetName, nodeInfo.Me)
 	if err != nil {
 		l.Error("wait for nomination: %v", err)
 	}
-
 	if !nominated {
 		l.Debug("skip after nomination, probably started by another node")
 		return
@@ -216,13 +217,7 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		Epoch:   &epoch,
 	})
 
-	// install a backup lock despite having PITR one
-	got, err := a.acquireLock(ctx, lck, l, func(ctx context.Context) (bool, error) {
-		return lck.Rewrite(ctx, &lock.LockHeader{
-			Replset: a.brief.SetName,
-			Type:    ctrl.CmdPITR,
-		})
-	})
+	got, err := a.acquireLock(ctx, lck, l)
 	if err != nil {
 		l.Error("acquiring lock: %v", err)
 		return
@@ -231,20 +226,27 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		l.Debug("skip: lock not acquired")
 		return
 	}
+	defer func() {
+		l.Debug("releasing lock")
+		err = lck.Release()
+		if err != nil {
+			l.Error("unable to release backup lock %v: %v", lck, err)
+		}
+	}()
 
 	err = backup.SetRSNomineeACK(ctx, a.leadConn, cmd.Name, nodeInfo.SetName, nodeInfo.Me)
 	if err != nil {
 		l.Warning("set nominee ack: %v", err)
 	}
 
 	bcpCtx, cancel := context.WithCancel(ctx)
-	a.setBcp(&currentBackup{
-		header: cmd,
-		cancel: cancel,
-	})
+	defer cancel()
+
+	a.setBcp(&currentBackup{cancel: cancel})
+	defer a.setBcp(nil)
+
 	l.Info("backup started")
 	err = bcp.Run(bcpCtx, cmd, opid, l)
-	a.unsetBcp()
 	if err != nil {
 		if errors.Is(err, storage.ErrCancelled) || errors.Is(err, context.Canceled) {
 			l.Info("backup was canceled")
@@ -254,18 +256,14 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 	} else {
 		l.Info("backup finished")
 	}
-
-	l.Debug("releasing lock")
-	err = lck.Release()
-	if err != nil {
-		l.Error("unable to release backup lock %v: %v", lck, err)
-	}
 }
 
 const renominationFrame = 5 * time.Second
 
-func (a *Agent) nominateRS(ctx context.Context, bcp, rs string, nodes [][]string, l log.LogEvent) error {
+func (a *Agent) nominateRS(ctx context.Context, bcp, rs string, nodes [][]string) error {
+	l := log.LogEventFromContext(ctx)
 	l.Debug("nomination list for %s: %v", rs, nodes)
+
 	err := backup.SetRSNomination(ctx, a.leadConn, bcp, rs)
 	if err != nil {
 		return errors.Wrap(err, "set nomination meta")
@@ -298,7 +296,9 @@ func (a *Agent) nominateRS(ctx context.Context, bcp, rs string, nodes [][]string
 	return nil
 }
 
-func (a *Agent) waitNomination(ctx context.Context, bcp, rs, node string, l log.LogEvent) (bool, error) {
+func (a *Agent) waitNomination(ctx context.Context, bcp, rs, node string) (bool, error) {
+	l := log.LogEventFromContext(ctx)
+
 	tk := time.NewTicker(time.Millisecond * 500)
 	defer tk.Stop()
 

cmd/pbm-agent/delete.go

@@ -43,15 +43,15 @@ func (a *Agent) Delete(ctx context.Context, d *ctrl.DeleteBackupCmd, opid ctrl.O
 	}
 
 	epts := ep.TS()
-	lock := lock.NewOpLock(a.leadConn, lock.LockHeader{
+	lock := lock.NewLock(a.leadConn, lock.LockHeader{
 		Replset: a.brief.SetName,
 		Node:    a.brief.Me,
 		Type:    ctrl.CmdDeleteBackup,
 		OPID:    opid.String(),
 		Epoch:   &epts,
 	})
 
-	got, err := a.acquireLock(ctx, lock, l, nil)
+	got, err := a.acquireLock(ctx, lock, l)
 	if err != nil {
 		l.Error("acquire lock: %v", err)
 		return
@@ -70,8 +70,10 @@ func (a *Agent) Delete(ctx context.Context, d *ctrl.DeleteBackupCmd, opid ctrl.O
 	case d.OlderThan > 0:
 		t := time.Unix(d.OlderThan, 0).UTC()
 		obj := t.Format("2006-01-02T15:04:05Z")
+
 		l = logger.NewEvent(string(ctrl.CmdDeleteBackup), obj, opid.String(), ep.TS())
 		ctx := log.SetLogEventToContext(ctx, l)
+
 		l.Info("deleting backups older than %v", t)
 		err := backup.DeleteBackupBefore(ctx, a.leadConn, t, "")
 		if err != nil {
@@ -81,6 +83,7 @@ func (a *Agent) Delete(ctx context.Context, d *ctrl.DeleteBackupCmd, opid ctrl.O
 	case d.Backup != "":
 		l = logger.NewEvent(string(ctrl.CmdDeleteBackup), d.Backup, opid.String(), ep.TS())
 		ctx := log.SetLogEventToContext(ctx, l)
+
 		l.Info("deleting backup")
 		err := backup.DeleteBackup(ctx, a.leadConn, d.Backup)
 		if err != nil {
@@ -119,15 +122,15 @@ func (a *Agent) DeletePITR(ctx context.Context, d *ctrl.DeletePITRCmd, opid ctrl
 	}
 
 	epts := ep.TS()
-	lock := lock.NewOpLock(a.leadConn, lock.LockHeader{
+	lock := lock.NewLock(a.leadConn, lock.LockHeader{
 		Replset: a.brief.SetName,
 		Node:    a.brief.Me,
 		Type:    ctrl.CmdDeletePITR,
 		OPID:    opid.String(),
 		Epoch:   &epts,
 	})
 
-	got, err := a.acquireLock(ctx, lock, l, nil)
+	got, err := a.acquireLock(ctx, lock, l)
 	if err != nil {
 		l.Error("acquire lock: %v", err)
 		return
@@ -173,11 +176,6 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 
 	ctx = log.SetLogEventToContext(ctx, l)
 
-	if d == nil {
-		l.Error("missed command")
-		return
-	}
-
 	nodeInfo, err := topo.GetNodeInfoExt(ctx, a.nodeConn)
 	if err != nil {
 		l.Error("get node info data: %v", err)
@@ -189,15 +187,15 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 	}
 
 	epts := ep.TS()
-	lock := lock.NewOpLock(a.leadConn, lock.LockHeader{
+	lock := lock.NewLock(a.leadConn, lock.LockHeader{
 		Replset: a.brief.SetName,
 		Node:    a.brief.Me,
 		Type:    ctrl.CmdCleanup,
 		OPID:    opid.String(),
 		Epoch:   &epts,
 	})
 
-	got, err := a.acquireLock(ctx, lock, l, nil)
+	got, err := a.acquireLock(ctx, lock, l)
 	if err != nil {
 		l.Error("acquire lock: %v", err)
 		return

cmd/pbm-agent/oplog.go

@@ -50,7 +50,7 @@ func (a *Agent) OplogReplay(ctx context.Context, r *ctrl.ReplayCmd, opID ctrl.OP
 		Epoch:   &epoch,
 	})
 
-	nominated, err := a.acquireLock(ctx, lck, l, nil)
+	nominated, err := a.acquireLock(ctx, lck, l)
 	if err != nil {
 		l.Error("acquiring lock: %s", err.Error())
 		return