PBM-1649 Profile support for delete and cleanup in PBM agent

Author: jcechace

cmd/pbm-agent/delete.go

@@ -96,7 +96,7 @@ func (a *Agent) Delete(ctx context.Context, d *ctrl.DeleteBackupCmd, opid ctrl.O
 		}
 
 		l.Info("deleting backups older than %v", t)
-		err = backup.DeleteBackupBefore(ctx, a.leadConn, t, bcpType, nodeInfo.Me)
+		err = backup.DeleteBackupBefore(ctx, a.leadConn, t, bcpType, d.Profile, nodeInfo.Me)
 		if err != nil {
 			l.Error("deleting: %v", err)
 			return
@@ -254,7 +254,7 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 		return
 	}
 
-	cfg, err := config.GetConfig(ctx, a.leadConn)
+	cfg, err := config.GetProfiledConfig(ctx, a.leadConn, d.Profile)
 	if err != nil {
 		l.Error("get config: %v", err)
 	}
@@ -267,7 +267,7 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 	eg := errgroup.Group{}
 	eg.SetLimit(runtime.NumCPU())
 
-	cr, err := backup.MakeCleanupInfo(ctx, a.leadConn, d.OlderThan)
+	cr, err := backup.MakeCleanupInfo(ctx, a.leadConn, d.OlderThan, d.Profile)
 	if err != nil {
 		l.Error("make cleanup report: " + err.Error())
 		return
@@ -289,15 +289,20 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 		bcp := &cr.Backups[i]
 
 		eg.Go(func() error {
-			err := backup.DeleteBackupFiles(stg, bcp.Name)
-			return errors.Wrapf(err, "delete backup files %q", bcp.Name)
+			err := backup.DeleteBackup(ctx, a.leadConn, bcp.Name, a.brief.Me)
+			return errors.Wrapf(err, "delete backup %q", bcp.Name)
 		})
 	}
 	if err := eg.Wait(); err != nil {
 		l.Error(err.Error())
 	}
 
-	err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me, false)
+	if d.Profile == "" {
+		err = resync.Resync(ctx, a.leadConn, &cfg.Storage, a.brief.Me, false)
+	} else {
+		err = resync.SyncBackupList(ctx, a.leadConn, &cfg.Storage, d.Profile, a.brief.Me)
+	}
+
 	if err != nil {
 		l.Error("storage resync: " + err.Error())
 	}
@@ -306,7 +311,7 @@ func (a *Agent) Cleanup(ctx context.Context, d *ctrl.CleanupCmd, opid ctrl.OPID,
 func (a *Agent) deletePITRImpl(ctx context.Context, ts primitive.Timestamp) error {
 	l := log.LogEventFromContext(ctx)
 
-	r, err := backup.MakeCleanupInfo(ctx, a.leadConn, ts)
+	r, err := backup.MakeCleanupInfo(ctx, a.leadConn, ts, "")
 	if err != nil {
 		return errors.Wrap(err, "get pitr chunks")
 	}

pbm/backup/delete.go

@@ -15,6 +15,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
 )
 
@@ -321,17 +322,23 @@ func DeleteBackupBefore(
 	conn connect.Client,
 	t time.Time,
 	bcpType defs.BackupType,
+	profile string,
 	node string,
 ) error {
-	backups, err := ListDeleteBackupBefore(ctx, conn, primitive.Timestamp{T: uint32(t.Unix())}, bcpType)
+	backups, err := ListDeleteBackupBefore(ctx, conn, primitive.Timestamp{T: uint32(t.Unix())}, bcpType, profile)
 	if err != nil {
 		return err
 	}
 	if len(backups) == 0 {
 		return nil
 	}
 
-	stg, err := util.GetStorage(ctx, conn, node, log.LogEventFromContext(ctx))
+	var stg storage.Storage
+	if profile == "" {
+		stg, err = util.GetStorage(ctx, conn, node, log.LogEventFromContext(ctx))
+	} else {
+		stg, err = util.GetProfiledStorage(ctx, conn, profile, node, log.LogEventFromContext(ctx))
+	}
 	if err != nil {
 		return errors.Wrap(err, "get storage")
 	}
@@ -360,7 +367,7 @@ func ListDeleteBackupBefore(
 	bcpType defs.BackupType,
 	profile string,
 ) ([]BackupMeta, error) {
-	info, err := MakeCleanupInfo(ctx, conn, ts)
+	info, err := MakeCleanupInfo(ctx, conn, ts, profile)
 	if err != nil {
 		return nil, err
 	}
@@ -391,8 +398,8 @@ func ListDeleteBackupBefore(
 	return rv, nil
 }
 
-func MakeCleanupInfo(ctx context.Context, conn connect.Client, ts primitive.Timestamp) (CleanupInfo, error) {
-	backups, err := listBackupsBefore(ctx, conn, primitive.Timestamp{T: ts.T + 1})
+func MakeCleanupInfo(ctx context.Context, conn connect.Client, ts primitive.Timestamp, profile string) (CleanupInfo, error) {
+	backups, err := listBackupsBefore(ctx, conn, primitive.Timestamp{T: ts.T + 1}, profile)
 	if err != nil {
 		return CleanupInfo{}, errors.Wrap(err, "list backups before")
 	}
@@ -500,9 +507,8 @@ func MakeCleanupInfo(ctx context.Context, conn connect.Client, ts primitive.Time
 // listBackupsBefore returns backups with restore cluster time less than or equals to ts.
 //
 // It does not include backups stored on an external storages.
-func listBackupsBefore(ctx context.Context, conn connect.Client, ts primitive.Timestamp) ([]BackupMeta, error) {
+func listBackupsBefore(ctx context.Context, conn connect.Client, ts primitive.Timestamp, profile string) ([]BackupMeta, error) {
 	f := bson.D{
-		{"store.profile", nil},
 		{"last_write_ts", bson.M{"$lt": ts}},
 		{"status", bson.M{"$in": bson.A{
 			defs.StatusDone,

sdk/impl.go

@@ -332,9 +332,8 @@ func (c *Client) DeleteOplogRange(ctx context.Context, until Timestamp) (Command
 	return CommandID(opid.String()), err
 }
 
-func (c *Client) CleanupReport(ctx context.Context, beforeTS Timestamp) (CleanupReport, error) {
-	return backup.MakeCleanupInfo(ctx, c.conn, beforeTS)
 func (c *Client) CleanupReport(ctx context.Context, beforeTS Timestamp, profile string) (CleanupReport, error) {
+	return backup.MakeCleanupInfo(ctx, c.conn, beforeTS, profile)
 }
 
 func (c *Client) RunCleanup(ctx context.Context, beforeTS Timestamp, profile string) (CommandID, error) {

sdk/sdk.go

@@ -225,7 +225,7 @@ func ListDeleteChunksBefore(
 	client *Client,
 	ts primitive.Timestamp,
 ) ([]OplogChunk, error) {
-	r, err := backup.MakeCleanupInfo(ctx, client.conn, ts)
+	r, err := backup.MakeCleanupInfo(ctx, client.conn, ts, "")
 	return r.Chunks, err
 }