[PBM-1234, PBM-1236] Unable to delete incremental backups (#905)

Author: defbin

cmd/pbm-agent/agent.go

@@ -206,7 +206,7 @@ func (a *Agent) Resync(ctx context.Context, opid ctrl.OPID, ep config.Epoch) {
 	}
 	l.Info("succeed")
 
-	epch, err := config.ResetEpoch(a.leadConn)
+	epch, err := config.ResetEpoch(ctx, a.leadConn)
 	if err != nil {
 		l.Error("reset epoch: %v", err)
 		return

cmd/pbm-agent/backup.go

@@ -83,8 +83,10 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		}
 	}
 
-	// wakeup the slicer to not wait for the tick
-	go a.sliceNow(opid)
+	if cmd.Type == defs.LogicalBackup {
+		// wakeup the slicer to not wait for the tick
+		go a.sliceNow(opid)
+	}
 
 	var bcp *backup.Backup
 	switch cmd.Type {

cmd/pbm-agent/delete.go

@@ -5,6 +5,7 @@ import (
 	"runtime"
 	"time"
 
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/percona/percona-backup-mongodb/pbm/backup"
@@ -74,8 +75,20 @@ func (a *Agent) Delete(ctx context.Context, d *ctrl.DeleteBackupCmd, opid ctrl.O
 		l = logger.NewEvent(string(ctrl.CmdDeleteBackup), obj, opid.String(), ep.TS())
 		ctx := log.SetLogEventToContext(ctx, l)
 
+		ct, err := topo.GetClusterTime(ctx, a.leadConn)
+		if err != nil {
+			l.Error("get cluster time: %v", err)
+			return
+		}
+		if d.OlderThan > int64(ct.T) {
+			providedTime := t.Format(time.RFC3339)
+			realTime := time.Unix(int64(ct.T), 0).UTC().Format(time.RFC3339)
+			l.Error("provided time %q is after now %q", providedTime, realTime)
+			return
+		}
+
 		l.Info("deleting backups older than %v", t)
-		err := backup.DeleteBackupBefore(ctx, a.leadConn, t, "")
+		err = backup.DeleteBackupBefore(ctx, a.leadConn, t, "")
 		if err != nil {
 			l.Error("deleting: %v", err)
 			return
@@ -148,11 +161,28 @@ func (a *Agent) DeletePITR(ctx context.Context, d *ctrl.DeletePITRCmd, opid ctrl
 	if d.OlderThan > 0 {
 		t := time.Unix(d.OlderThan, 0).UTC()
 		obj := t.Format("2006-01-02T15:04:05Z")
+
 		l = logger.NewEvent(string(ctrl.CmdDeletePITR), obj, opid.String(), ep.TS())
+		ctx := log.SetLogEventToContext(ctx, l)
+
+		var ct primitive.Timestamp
+		ct, err = topo.GetClusterTime(ctx, a.leadConn)
+		if err != nil {
+			l.Error("get cluster time: %v", err)
+			return
+		}
+		if d.OlderThan > int64(ct.T) {
+			providedTime := t.Format(time.RFC3339)
+			realTime := time.Unix(int64(ct.T), 0).UTC().Format(time.RFC3339)
+			l.Error("provided time %q is after now %q", providedTime, realTime)
+			return
+		}
+
 		l.Info("deleting pitr chunks older than %v", t)
 		err = oplogtmp.DeletePITR(ctx, a.leadConn, &t, l)
 	} else {
 		l = logger.NewEvent(string(ctrl.CmdDeletePITR), "_all_", opid.String(), ep.TS())
+		ctx := log.SetLogEventToContext(ctx, l)
 		l.Info("deleting all pitr chunks")
 		err = oplogtmp.DeletePITR(ctx, a.leadConn, nil, l)
 	}
@@ -210,6 +240,18 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 		}
 	}()
 
+	ct, err := topo.GetClusterTime(ctx, a.leadConn)
+	if err != nil {
+		l.Error("get cluster time: %v", err)
+		return
+	}
+	if d.OlderThan.T > ct.T {
+		providedTime := time.Unix(int64(ct.T), 0).UTC().Format(time.RFC3339)
+		realTime := time.Unix(int64(ct.T), 0).UTC().Format(time.RFC3339)
+		l.Error("provided time %q is after now %q", providedTime, realTime)
+		return
+	}
+
 	stg, err := util.GetStorage(ctx, a.leadConn, l)
 	if err != nil {
 		l.Error("get storage: " + err.Error())

cmd/pbm-agent/oplog.go

@@ -77,7 +77,7 @@ func (a *Agent) OplogReplay(ctx context.Context, r *ctrl.ReplayCmd, opID ctrl.OP
 	}
 	l.Info("oplog replay successfully finished")
 
-	resetEpoch, err := config.ResetEpoch(a.leadConn)
+	resetEpoch, err := config.ResetEpoch(ctx, a.leadConn)
 	if err != nil {
 		l.Error("reset epoch: %s", err.Error())
 		return

cmd/pbm-agent/restore.go

@@ -118,7 +118,7 @@ func canSlicingNow(ctx context.Context, conn connect.Client) error {
 			return errors.Wrap(err, "get backup metadata")
 		}
 
-		if bcp.Type != defs.PhysicalBackup && bcp.Type != defs.ExternalBackup {
+		if bcp.Type == defs.LogicalBackup {
 			return lock.ConcurrentOpError{l.LockHeader}
 		}
 	}
@@ -442,7 +442,7 @@ func (a *Agent) Restore(ctx context.Context, r *ctrl.RestoreCmd, opid ctrl.OPID,
 	}
 
 	if bcpType == defs.LogicalBackup && nodeInfo.IsLeader() {
-		epch, err := config.ResetEpoch(a.leadConn)
+		epch, err := config.ResetEpoch(ctx, a.leadConn)
 		if err != nil {
 			l.Error("reset epoch: %v", err)
 		}

cmd/pbm/delete.go

@@ -111,6 +111,11 @@ func deleteManyBackup(ctx context.Context, pbm sdk.Client, d *deleteBcpOpts) (sd
 	if err != nil {
 		return sdk.NoOpID, errors.Wrap(err, "parse --older-than")
 	}
+	if n := time.Now().UTC(); ts.T > uint32(n.Unix()) {
+		providedTime := time.Unix(int64(ts.T), 0).UTC().Format(time.RFC3339)
+		realTime := n.Format(time.RFC3339)
+		return sdk.NoOpID, errors.Errorf("--older-than %q is after now %q", providedTime, realTime)
+	}
 
 	bcpType := sdk.ParseBackupType(d.bcpType)
 	backups, err := sdk.ListDeleteBackupBefore(ctx, pbm, ts, bcpType)
@@ -189,6 +194,11 @@ func deletePITR(
 		if err != nil {
 			return nil, errors.Wrap(err, "parse --older-then")
 		}
+		if n := time.Now().UTC(); ts.T > uint32(n.Unix()) {
+			providedTime := time.Unix(int64(ts.T), 0).UTC().Format(time.RFC3339)
+			realTime := n.Format(time.RFC3339)
+			return nil, errors.Errorf("--older-than %q is after now %q", providedTime, realTime)
+		}
 	}
 	cid, err := pbm.DeleteOplogRange(ctx, ts)
 	if err != nil {
@@ -214,6 +224,12 @@ func doCleanup(ctx context.Context, conn connect.Client, pbm sdk.Client, d *clea
 	if err != nil {
 		return nil, errors.Wrap(err, "parse --older-than")
 	}
+	if n := time.Now().UTC(); ts.T > uint32(n.Unix()) {
+		providedTime := time.Unix(int64(ts.T), 0).UTC().Format(time.RFC3339)
+		realTime := n.Format(time.RFC3339)
+		return nil, errors.Errorf("--older-than %q is after now %q", providedTime, realTime)
+	}
+
 	info, err := pbm.CleanupReport(ctx, ts)
 	if err != nil {
 		return nil, errors.Wrap(err, "make cleanup report")

pbm/backup/backup.go

@@ -314,14 +314,6 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 	}
 
 	if inf.IsLeader() {
-		var epch config.Epoch
-		epch, err = config.ResetEpochWithContext(ctx, b.leadConn)
-		if err != nil {
-			l.Error("reset epoch")
-		} else {
-			l.Debug("epoch set to %v", epch)
-		}
-
 		err = b.reconcileStatus(ctx, bcp.Name, opid.String(), defs.StatusDone, nil)
 		if err != nil {
 			return errors.Wrap(err, "check cluster for backup done")

pbm/backup/delete.go

@@ -142,7 +142,7 @@ func CanDeleteIncrementalChain(
 	if base.Status.IsRunning() {
 		return ErrBackupInProgress
 	}
-	if base.Type == defs.IncrementalBackup {
+	if base.Type != defs.IncrementalBackup {
 		return ErrNonIncrementalBackup
 	}
 	if base.SrcBackup != "" {
@@ -269,7 +269,8 @@ func isRequiredForOplogSlicing(ctx context.Context, cc connect.Client, lw primit
 	if err != nil {
 		return false, errors.Wrap(err, "get oplog range from previous backup")
 	}
-	if len(timelines) == 1 && timelines[0].Start <= prevRestoreTime.T {
+	// check if there is a gap (missed ops) in oplog range between previous and following backup restore_to time
+	if len(timelines) != 1 || prevRestoreTime.T >= timelines[0].Start {
 		return false, nil
 	}
 
@@ -326,7 +327,11 @@ func ListDeleteBackupBefore(
 	}
 
 	pred := func(m *BackupMeta) bool { return m.Type == bcpType }
-	if bcpType == SelectiveBackup {
+	if bcpType == defs.LogicalBackup {
+		pred = func(m *BackupMeta) bool {
+			return m.Type == defs.LogicalBackup && !util.IsSelective(m.Namespaces)
+		}
+	} else if bcpType == SelectiveBackup {
 		pred = func(m *BackupMeta) bool { return util.IsSelective(m.Namespaces) }
 	}
 
@@ -345,71 +350,85 @@ func MakeCleanupInfo(ctx context.Context, conn connect.Client, ts primitive.Time
 	if err != nil {
 		return CleanupInfo{}, errors.Wrap(err, "list backups before")
 	}
-
-	exclude := true
-	if l := len(backups) - 1; l != -1 && backups[l].LastWriteTS.T == ts.T {
-		// there is a backup at the `ts`
-		if isValidBaseSnapshot(&backups[l]) {
-			// it can be used to fully restore data to the `ts` state.
-			// no need to exclude any base snapshot and chunks before the `ts`
-			exclude = false
-		}
-		// the backup is not considered to be deleted.
-		// used only for `exclude` value
-		backups = backups[:l]
-	}
-
-	// exclude the last incremental backups if it is required for following (after the `ts`)
-	backups, err = extractLastIncrementalChain(ctx, conn, backups)
-	if err != nil {
-		return CleanupInfo{}, errors.Wrap(err, "extract last incremental chain")
-	}
-
 	chunks, err := listChunksBefore(ctx, conn, ts)
 	if err != nil {
 		return CleanupInfo{}, errors.Wrap(err, "list chunks before")
 	}
-	if !exclude {
-		// all chunks can be deleted. there is a backup to fully restore data
+	if len(backups) == 0 {
 		return CleanupInfo{Backups: backups, Chunks: chunks}, nil
 	}
 
-	// the following check is needed for "delete all" special case.
-	// if there is no base snapshot after `ts` and PITR is running,
-	// the last base snapshot before `ts` should be excluded.
-	// otherwise, it is allowed to delete everything before `ts`
-	required, err := isRequiredForOplogSlicing(ctx, conn, ts)
-	if err != nil {
-		return CleanupInfo{}, err
-	}
-	if !required {
-		return CleanupInfo{Backups: backups, Chunks: chunks}, nil
+	if r := &backups[len(backups)-1]; r.LastWriteTS.T == ts.T {
+		// there is a backup at the `ts`
+		backups = backups[:len(backups)-1]
+
+		if isValidBaseSnapshot(r) {
+			// it can be used to fully restore data to the `ts` state.
+			// no need to exclude any backups and chunks before the `ts`.
+			// except increments that is base for following increment (after `ts`) must be excluded
+			backups, err = extractLastIncrementalChain(ctx, conn, backups)
+			if err != nil {
+				return CleanupInfo{}, errors.Wrap(err, "extract last incremental chain")
+			}
+
+			return CleanupInfo{Backups: backups, Chunks: chunks}, nil
+		}
 	}
 
-	// the `baseIndex` could be the base snapshot index for PITR to the `ts`
-	// or for currently running PITR
-	baseIndex := findLastBaseSnapshotIndex(backups)
+	baseIndex := len(backups) - 1
+	for ; baseIndex != -1; baseIndex-- {
+		if isValidBaseSnapshot(&backups[baseIndex]) {
+			break
+		}
+	}
 	if baseIndex == -1 {
-		// nothing to keep
+		// no valid base snapshot to exclude
 		return CleanupInfo{Backups: backups, Chunks: chunks}, nil
 	}
 
-	excluded := false
-	origin := chunks
-	chunks = []oplog.OplogChunk{}
-	for i := range origin {
-		if backups[baseIndex].LastWriteTS.Compare(origin[i].EndTS) != -1 {
-			chunks = append(chunks, origin[i])
+	beforeChunks := []oplog.OplogChunk{}
+	afterChunks := []oplog.OplogChunk{}
+	for _, chunk := range chunks {
+		if chunk.EndTS.Before(backups[baseIndex].LastWriteTS) {
+			beforeChunks = append(beforeChunks, chunk)
 		} else {
-			excluded = true
+			// keep chunks after the last base snapshot restore time
+			// the backup may be excluded
+			afterChunks = append(afterChunks, chunk)
 		}
 	}
 
-	// if excluded is false, the last found base snapshot is not used for PITR
-	// no need to keep it. otherwise, should be excluded
-	if excluded {
+	if oplog.HasSingleTimelineToCover(afterChunks, backups[baseIndex].LastWriteTS.T, ts.T) {
+		// there is single continuous oplog range between snapshot and ts.
+		// keep the backup and chunks to be able to restore to the ts
 		copy(backups[baseIndex:], backups[baseIndex+1:])
 		backups = backups[:len(backups)-1]
+		chunks = beforeChunks
+	} else {
+		// no chunks yet but if PITR is ON, the backup can be base snapshot for it
+		enabled, oplogOnly, err := config.IsPITREnabled(ctx, conn)
+		if err != nil {
+			return CleanupInfo{}, errors.Wrap(err, "get PITR status")
+		}
+		if enabled && !oplogOnly {
+			nextRestoreTime, err := FindBaseSnapshotLWAfter(ctx, conn, ts)
+			if err != nil {
+				return CleanupInfo{}, errors.Wrap(err, "find next snapshot")
+			}
+
+			if nextRestoreTime.IsZero() {
+				// it is not the last base snapshot for PITR
+				copy(backups[baseIndex:], backups[baseIndex+1:])
+				backups = backups[:len(backups)-1]
+				chunks = beforeChunks
+			}
+		}
+	}
+
+	// exclude increments that is base for following increment (after `ts`)
+	backups, err = extractLastIncrementalChain(ctx, conn, backups)
+	if err != nil {
+		return CleanupInfo{}, errors.Wrap(err, "extract last incremental chain")
 	}
 
 	return CleanupInfo{Backups: backups, Chunks: chunks}, nil
@@ -471,6 +490,10 @@ func extractLastIncrementalChain(
 		return bcps, nil
 	}
 
+	return extractIncrementalChain(bcps, i), nil
+}
+
+func extractIncrementalChain(bcps []BackupMeta, i int) []BackupMeta {
 	for base := bcps[i].Name; i != -1; i-- {
 		if bcps[i].Status != defs.StatusDone {
 			continue
@@ -490,15 +513,5 @@ func extractLastIncrementalChain(
 		}
 	}
 
-	return bcps, nil
-}
-
-func findLastBaseSnapshotIndex(bcps []BackupMeta) int {
-	for i := len(bcps) - 1; i != -1; i-- {
-		if isValidBaseSnapshot(&bcps[i]) {
-			return i
-		}
-	}
-
-	return -1
+	return bcps
 }