[PBM-1246] improve delete-pitr command (#910)

Author: defbin

cmd/pbm-agent/delete.go

@@ -5,17 +5,20 @@ import (
 	"runtime"
 	"time"
 
+	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/percona/percona-backup-mongodb/pbm/backup"
 	"github.com/percona/percona-backup-mongodb/pbm/config"
+	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/ctrl"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/lock"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
-	"github.com/percona/percona-backup-mongodb/pbm/oplog/oplogtmp"
+	"github.com/percona/percona-backup-mongodb/pbm/oplog"
 	"github.com/percona/percona-backup-mongodb/pbm/resync"
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
 )
@@ -164,34 +167,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")
+	ct, err := topo.ClusterTimeFromNodeInfo(nodeInfo)
+	if err != nil {
+		l.Error("get cluster time: %v", err)
+		return
+	}
 
-		l = logger.NewEvent(string(ctrl.CmdDeletePITR), obj, opid.String(), ep.TS())
-		ctx := log.SetLogEventToContext(ctx, l)
+	t := time.Unix(d.OlderThan, 0).UTC()
+	obj := t.Format("2006-01-02T15:04:05Z")
 
-		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 = logger.NewEvent(string(ctrl.CmdDeletePITR), obj, opid.String(), ep.TS())
+	ctx = log.SetLogEventToContext(ctx, l)
 
-		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)
+	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
 	}
+
+	ts := primitive.Timestamp{T: uint32(t.Unix())}
+	l.Info("deleting pitr chunks older than %v", t)
+	err = deletePITRImpl(ctx, a.leadConn, ts)
 	if err != nil {
 		l.Error("deleting: %v", err)
 		return
@@ -301,3 +298,73 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 		l.Error("storage resync: " + err.Error())
 	}
 }
+
+func deletePITRImpl(ctx context.Context, conn connect.Client, ts primitive.Timestamp) error {
+	l := log.LogEventFromContext(ctx)
+
+	cfg, err := config.GetConfig(ctx, conn)
+	if err != nil {
+		return errors.Wrap(err, "get config")
+	}
+
+	if cfg.PITR.Enabled && !cfg.PITR.OplogOnly {
+		ct, err := topo.GetClusterTime(ctx, conn)
+		if err != nil {
+			return errors.Wrap(err, "get cluster time")
+		}
+
+		lw, err := backup.FindBaseSnapshotLWBefore(ctx, conn, ct, primitive.Timestamp{})
+		if err != nil {
+			return errors.Wrap(err, "find previous snapshot")
+		}
+		if !lw.IsZero() {
+			if lw.T < ts.T || (lw.T == ts.T && (ts.I == 0 || lw.I < ts.I)) {
+				ts = lw
+			}
+		}
+	}
+
+	chunks, err := oplog.ListDeleteChunksBefore(ctx, conn, ts)
+	if err != nil {
+		return errors.Wrap(err, "get pitr chunks")
+	}
+	if len(chunks) == 0 {
+		l.Debug("nothing to delete")
+		return nil
+	}
+
+	stg, err := util.StorageFromConfig(cfg.Storage, l)
+	if err != nil {
+		return errors.Wrap(err, "get storage")
+	}
+
+	return deleteChunks(ctx, conn, stg, chunks)
+}
+
+func deleteChunks(ctx context.Context, m connect.Client, stg storage.Storage, chunks []oplog.OplogChunk) error {
+	l := log.LogEventFromContext(ctx)
+
+	for _, chnk := range chunks {
+		err := stg.Delete(chnk.FName)
+		if err != nil && !errors.Is(err, storage.ErrNotExist) {
+			return errors.Wrapf(err, "delete pitr chunk '%s' (%v) from storage", chnk.FName, chnk)
+		}
+
+		_, err = m.PITRChunksCollection().DeleteOne(
+			ctx,
+			bson.D{
+				{"rs", chnk.RS},
+				{"start_ts", chnk.StartTS},
+				{"end_ts", chnk.EndTS},
+			},
+		)
+
+		if err != nil {
+			return errors.Wrap(err, "delete pitr chunk metadata")
+		}
+
+		l.Debug("deleted %s", chnk.FName)
+	}
+
+	return nil
+}

cmd/pbm/delete.go

@@ -12,6 +12,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson/primitive"
 
 	"github.com/percona/percona-backup-mongodb/pbm/backup"
+	"github.com/percona/percona-backup-mongodb/pbm/config"
 	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/ctrl"
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
@@ -33,7 +34,6 @@ func deleteBackup(
 	conn connect.Client,
 	pbm sdk.Client,
 	d *deleteBcpOpts,
-	outf outFormat,
 ) (fmt.Stringer, error) {
 	if d.name == "" && d.olderThan == "" {
 		return nil, errors.New("either --name or --older-than should be set")
@@ -59,7 +59,7 @@ func deleteBackup(
 		return nil, err
 	}
 
-	if outf != outText || d.dryRun {
+	if d.dryRun {
 		return nil, nil
 	}
 
@@ -153,22 +153,72 @@ type deletePitrOpts struct {
 	olderThan string
 	yes       bool
 	all       bool
+	wait      bool
+	dryRun    bool
 }
 
 func deletePITR(
 	ctx context.Context,
 	conn connect.Client,
 	pbm sdk.Client,
 	d *deletePitrOpts,
-	outf outFormat,
 ) (fmt.Stringer, error) {
+	if d.olderThan == "" && !d.all {
+		return nil, errors.New("either --older-than or --all should be set")
+	}
 	if d.olderThan != "" && d.all {
 		return nil, errors.New("cannot use --older-then and --all at the same command")
 	}
-	if !d.all && d.olderThan == "" {
-		return nil, errors.New("either --older-than or --all should be set")
+
+	var until primitive.Timestamp
+	if d.all {
+		until = primitive.Timestamp{T: uint32(time.Now().UTC().Unix())}
+	} else {
+		var err error
+		until, err = parseOlderThan(d.olderThan)
+		if err != nil {
+			return nil, errors.Wrap(err, "parse --older-then")
+		}
+
+		now := time.Now().UTC()
+		if until.T > uint32(now.Unix()) {
+			providedTime := time.Unix(int64(until.T), 0).UTC().Format(time.RFC3339)
+			realTime := now.Format(time.RFC3339)
+			return nil, errors.Errorf("--older-than %q is after now %q", providedTime, realTime)
+		}
+	}
+
+	enabled, oplogOnly, err := config.IsPITREnabled(ctx, conn)
+	if err != nil {
+		return nil, errors.Wrap(err, "check pitr status")
 	}
 
+	if enabled && !oplogOnly {
+		lw, err := backup.FindBaseSnapshotLWBefore(ctx,
+			conn, primitive.Timestamp{T: uint32(time.Now().UTC().Unix())}, primitive.Timestamp{})
+		if err != nil {
+			return nil, errors.Wrap(err, "find previous snapshot")
+		}
+		if !lw.IsZero() {
+			if lw.T < until.T || (lw.T == until.T && (until.I == 0 || lw.I < until.I)) {
+				until = lw
+			}
+		}
+	}
+
+	chunks, err := sdk.ListDeleteChunksBefore(ctx, pbm, until)
+	if err != nil {
+		return nil, errors.Wrap(err, "list chunks")
+	}
+	if len(chunks) == 0 {
+		return outMsg{"nothing to delete"}, nil
+	}
+
+	printDeleteInfoTo(os.Stdout, nil, chunks)
+
+	if d.dryRun {
+		return nil, nil
+	}
 	if !d.yes {
 		q := "Are you sure you want to delete chunks?"
 		if d.all {
@@ -182,26 +232,13 @@ func deletePITR(
 		}
 	}
 
-	var ts primitive.Timestamp
-	if d.olderThan != "" {
-		var err error
-		ts, err = parseOlderThan(d.olderThan)
-		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)
+	cid, err := pbm.DeleteOplogRange(ctx, until)
 	if err != nil {
 		return nil, errors.Wrap(err, "schedule pitr delete")
 	}
 
-	if outf != outText {
-		return nil, nil
+	if !d.wait {
+		return outMsg{"Processing by agents. Please check status later"}, nil
 	}
 
 	return waitForDelete(ctx, conn, pbm, cid)
@@ -314,7 +351,7 @@ func printDeleteInfoTo(w io.Writer, backups []backup.BackupMeta, chunks []oplog.
 			}
 
 			restoreTime := time.Unix(int64(bcp.LastWriteTS.T), 0).UTC().Format(time.RFC3339)
-			fmt.Fprintf(w, " - %q [size: %s type: <%s>, restore time: %s]",
+			fmt.Fprintf(w, " - %q [size: %s type: <%s>, restore time: %s]\n",
 				bcp.Name, fmtSize(bcp.Size), t, restoreTime)
 		}
 	}

cmd/pbm/main.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	stdlog "log"
 	"os"
 	"strings"
 	"time"
@@ -264,7 +263,7 @@ func main() {
 			datetimeFormat,
 			dateFormat)).
 		StringVar(&deletePitr.olderThan)
-	deletePitrCmd.Flag("all", "Delete all chunks").
+	deletePitrCmd.Flag("all", `Delete all chunks (deprecated). Use --older-than="0d"`).
 		Short('a').
 		BoolVar(&deletePitr.all)
 	deletePitrCmd.Flag("yes", "Don't ask for confirmation").
@@ -273,6 +272,11 @@ func main() {
 	deletePitrCmd.Flag("force", "Don't ask for confirmation (deprecated)").
 		Short('f').
 		BoolVar(&deletePitr.yes)
+	deletePitrCmd.Flag("wait", "Wait for deletion done").
+		Short('w').
+		BoolVar(&deletePitr.wait)
+	deletePitrCmd.Flag("dry-run", "Report but do not delete").
+		BoolVar(&deletePitr.dryRun)
 
 	cleanupCmd := pbmCmd.Command("cleanup", "Delete Backups and PITR chunks")
 	cleanupOpts := cleanupOptions{}
@@ -324,7 +328,7 @@ func main() {
 		BoolVar(&logs.extr)
 
 	statusOpts := statusOptions{}
-	statusCmd := pbmCmd.Command("status", "Show PBM status")
+	statusCmd := pbmCmd.Command("status", "Show PBM status").Alias("s")
 	statusCmd.Flag(RSMappingFlag, RSMappingDoc).
 		Envar(RSMappingEnvVar).
 		StringVar(&statusOpts.rsMap)
@@ -386,7 +390,8 @@ func main() {
 
 		ver, err := version.GetMongoVersion(ctx, conn.MongoClient())
 		if err != nil {
-			stdlog.Fatalf("get mongo version: %v", err)
+			fmt.Fprintf(os.Stderr, "get mongo version: %v", err)
+			os.Exit(1)
 		}
 		if err := version.FeatureSupport(ver).PBMSupport(); err != nil {
 			fmt.Fprintf(os.Stderr, "WARNING: %v\n", err)
@@ -420,9 +425,9 @@ func main() {
 	case listCmd.FullCommand():
 		out, err = runList(ctx, conn, pbm, &list)
 	case deleteBcpCmd.FullCommand():
-		out, err = deleteBackup(ctx, conn, pbm, &deleteBcp, pbmOutF)
+		out, err = deleteBackup(ctx, conn, pbm, &deleteBcp)
 	case deletePitrCmd.FullCommand():
-		out, err = deletePITR(ctx, conn, pbm, &deletePitr, pbmOutF)
+		out, err = deletePITR(ctx, conn, pbm, &deletePitr)
 	case cleanupCmd.FullCommand():
 		out, err = doCleanup(ctx, conn, pbm, &cleanupOpts)
 	case logsCmd.FullCommand():

pbm/oplog/chunk.go

@@ -491,3 +491,16 @@ func HasSingleTimelineToCover(chunks []OplogChunk, from, till uint32) bool {
 
 	return false
 }
+
+func ListDeleteChunksBefore(ctx context.Context, conn connect.Client, ts primitive.Timestamp) ([]OplogChunk, error) {
+	f := bson.D{{"end_ts", bson.M{"$lt": ts}}}
+	o := options.Find().SetSort(bson.D{{"start_ts", 1}})
+	cur, err := conn.PITRChunksCollection().Find(ctx, f, o)
+	if err != nil {
+		return nil, errors.Wrap(err, "query")
+	}
+
+	rv := []OplogChunk{}
+	err = cur.All(ctx, &rv)
+	return rv, errors.Wrap(err, "cursor: all")
+}