[PBM-1252] keep list of physical files in storage (#927)

Author: defbin

cmd/pbm-agent/agent.go

@@ -34,6 +34,8 @@ type Agent struct {
 
 	brief topo.NodeBrief
 
+	mongoVersion version.MongoVersion
+
 	dumpConns int
 
 	closeCMD chan struct{}
@@ -51,7 +53,12 @@ func newAgent(ctx context.Context, leadConn connect.Client, uri string, dumpConn
 
 	info, err := topo.GetNodeInfo(ctx, m)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "get node info")
+	}
+
+	mongoVersion, err := version.GetMongoVersion(ctx, m)
+	if err != nil {
+		return nil, errors.Wrap(err, "get mongo version")
 	}
 
 	a := &Agent{
@@ -64,6 +71,7 @@ func newAgent(ctx context.Context, leadConn connect.Client, uri string, dumpConn
 		SetName: info.SetName,
 		Me:      info.Me,
 	}
+	a.mongoVersion = mongoVersion
 	a.dumpConns = dumpConns
 	return a, nil
 }

cmd/pbm-agent/backup.go

@@ -112,6 +112,7 @@ func (a *Agent) Backup(ctx context.Context, cmd *ctrl.BackupCmd, opid ctrl.OPID,
 		return
 	}
 
+	bcp.SetMongoVersion(a.mongoVersion.VersionString)
 	bcp.SetSlicerInterval(cfg.BackupSlicerInterval())
 	bcp.SetTimeouts(cfg.Backup.Timeouts)
 

cmd/pbm/backup.go

@@ -138,6 +138,7 @@ func runBackup(
 			Namespaces:       nss,
 			Compression:      compression,
 			CompressionLevel: level,
+			Filelist:         b.externList,
 		},
 	})
 	if err != nil {
@@ -169,7 +170,7 @@ func runBackup(
 			return nil, errors.Errorf("unexpected backup status %v", str)
 		}
 
-		bcp, err := backup.NewDBManager(conn).GetBackupByName(ctx, b.name)
+		bcp, err := pbm.GetBackupByName(ctx, b.name, sdk.GetBackupByNameOptions{FetchFilelist: b.externList})
 		if err != nil {
 			return nil, errors.Wrap(err, "get backup meta")
 		}

cmd/pbm/main.go

@@ -132,7 +132,7 @@ func main() {
 	backupCmd.Flag("wait", "Wait for the backup to finish").
 		Short('w').
 		BoolVar(&backupOptions.wait)
-	backupCmd.Flag("list-files", "Wait for the backup to finish").
+	backupCmd.Flag("list-files", "Shows the list of files per node (external backup only)").
 		Short('l').
 		BoolVar(&backupOptions.externList)
 

pbm/backup/backup.go

@@ -27,6 +27,7 @@ type Backup struct {
 	leadConn            connect.Client
 	nodeConn            *mongo.Client
 	brief               topo.NodeBrief
+	mongoVersion        string
 	typ                 defs.BackupType
 	incrBase            bool
 	timeouts            *config.BackupTimeouts
@@ -72,6 +73,10 @@ func NewIncremental(leadConn connect.Client, conn *mongo.Client, brief topo.Node
 	}
 }
 
+func (b *Backup) SetMongoVersion(v string) {
+	b.mongoVersion = v
+}
+
 func (b *Backup) SetTimeouts(t *config.BackupTimeouts) {
 	b.timeouts = t
 }
@@ -117,6 +122,7 @@ func (b *Backup) Init(
 		// the driver (mongo?) sets TS to the current wall clock if TS was 0, so have to init with 1
 		FirstWriteTS:   primitive.Timestamp{T: 1, I: 1},
 		PBMVersion:     version.Current().Version,
+		MongoVersion:   b.mongoVersion,
 		Nomination:     []BackupRsNomination{},
 		BalancerStatus: balancer,
 		Hb:             ts,
@@ -132,12 +138,6 @@ func (b *Backup) Init(
 	}
 	meta.Store = cfg.Storage
 
-	ver, err := version.GetMongoVersion(ctx, b.nodeConn)
-	if err != nil {
-		return errors.Wrap(err, "get mongo version")
-	}
-	meta.MongoVersion = ver.VersionString
-
 	fcv, err := version.GetFCV(ctx, b.nodeConn)
 	if err != nil {
 		return errors.Wrap(err, "get featureCompatibilityVersion")
@@ -182,6 +182,8 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 	rsMeta := BackupReplset{
 		Name:         inf.SetName,
 		Node:         inf.Me,
+		PBMVersion:   version.Current().Version,
+		MongoVersion: b.mongoVersion,
 		StartTS:      time.Now().UTC().Unix(),
 		Status:       defs.StatusRunning,
 		Conditions:   []Condition{},

pbm/backup/physical.go

@@ -354,7 +354,7 @@ func (b *Backup) doPhysical(
 	}
 
 	if b.typ == defs.ExternalBackup {
-		return b.handleExternal(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, opid, inf, l)
+		return b.handleExternal(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, opid, inf, stg, l)
 	}
 
 	return b.uploadPhysical(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, stg, l)
@@ -364,16 +364,18 @@ func (b *Backup) handleExternal(
 	ctx context.Context,
 	bcp *ctrl.BackupCmd,
 	rsMeta *BackupReplset,
-	data,
+	data []File,
 	jrnls []File,
 	dbpath string,
 	opid ctrl.OPID,
 	inf *topo.NodeInfo,
+	stg storage.Storage,
 	l log.LogEvent,
 ) error {
+	filelist := make(Filelist, 0, len(data)+len(jrnls)+2) // +2 for metadata and filelist
 	for _, f := range append(data, jrnls...) {
 		f.Name = path.Clean("./" + strings.TrimPrefix(f.Name, dbpath))
-		rsMeta.Files = append(rsMeta.Files, f)
+		filelist = append(filelist, f)
 	}
 
 	// We'll rewrite rs' LastWriteTS with the cluster LastWriteTS for the meta
@@ -389,19 +391,33 @@ func (b *Backup) handleExternal(
 	}
 	// save rs meta along with the data files so it can be used during the restore
 	metaf := fmt.Sprintf(defs.ExternalRsMetaFile, fsMeta.Name)
-	fsMeta.Files = append(fsMeta.Files, File{
-		Name: metaf,
-	})
+	filelist = append(filelist, File{Name: metaf}, File{Name: FilelistName})
 	metadst := filepath.Join(dbpath, metaf)
 	err = writeRSmetaToDisk(metadst, &fsMeta)
 	if err != nil {
 		// we can restore without it
 		l.Warning("failed to save rs meta file <%s>: %v", metadst, err)
 	}
 
-	err = RSSetPhyFiles(ctx, b.leadConn, bcp.Name, rsMeta.Name, rsMeta)
+	filelistPath := filepath.Join(dbpath, FilelistName)
+	err = saveFilelist(filelistPath, filelist)
 	if err != nil {
-		return errors.Wrap(err, "set shard's files list")
+		return errors.Wrap(err, "save filelist to dbpath")
+	}
+
+	if bcp.Filelist {
+		// keep filelist on backup storage for listing files to copy
+		bcpStoragePath := path.Join(bcp.Name, rsMeta.Name, FilelistName)
+		_, err = storage.Upload(ctx, filelist, stg, compress.CompressionTypeNone, nil, bcpStoragePath, -1)
+		if err != nil {
+			return errors.Wrapf(err, "save filelist to storage: %q", bcpStoragePath)
+		}
+
+		defer func() {
+			if err := stg.Delete(bcp.Name); err != nil {
+				l.Warning("remove backup folder <%s>: %v", bcp.Name, err)
+			}
+		}()
 	}
 
 	err = b.toState(ctx, defs.StatusCopyReady, bcp.Name, opid.String(), inf, nil)
@@ -419,10 +435,28 @@ func (b *Backup) handleExternal(
 	if err != nil {
 		l.Warning("remove rs meta file <%s>: %v", metadst, err)
 	}
+	err = os.Remove(filelistPath)
+	if err != nil {
+		l.Warning("remove file <%s>: %v", filelistPath, err)
+	}
 
 	return nil
 }
 
+func saveFilelist(filepath string, fl Filelist) error {
+	fw, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
+	if err != nil {
+		return errors.Wrapf(err, "create/open")
+	}
+	defer fw.Close()
+
+	if _, err = fl.WriteTo(fw); err != nil {
+		return errors.Wrap(err, "write")
+	}
+
+	return errors.Wrap(fw.Sync(), "fsync")
+}
+
 func writeRSmetaToDisk(fname string, rsMeta *BackupReplset) error {
 	fw, err := os.OpenFile(fname, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
 	if err != nil {
@@ -454,35 +488,38 @@ func (b *Backup) uploadPhysical(
 	stg storage.Storage,
 	l log.LogEvent,
 ) error {
-	var err error
 	l.Info("uploading data")
-	rsMeta.Files, err = uploadFiles(ctx, data, bcp.Name+"/"+rsMeta.Name, dbpath,
+	dataFiles, err := uploadFiles(ctx, data, bcp.Name+"/"+rsMeta.Name, dbpath,
 		b.typ == defs.IncrementalBackup, stg, bcp.Compression, bcp.CompressionLevel, l)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "upload data files")
 	}
 	l.Info("uploading data done")
 
 	l.Info("uploading journals")
 	ju, err := uploadFiles(ctx, jrnls, bcp.Name+"/"+rsMeta.Name, dbpath,
 		false, stg, bcp.Compression, bcp.CompressionLevel, l)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "upload journal files")
 	}
 	l.Info("uploading journals done")
-	rsMeta.Files = append(rsMeta.Files, ju...)
 
-	err = RSSetPhyFiles(ctx, b.leadConn, bcp.Name, rsMeta.Name, rsMeta)
-	if err != nil {
-		return errors.Wrap(err, "set shard's files list")
-	}
+	filelist := Filelist(dataFiles)
+	filelist = append(filelist, ju...)
 
 	size := int64(0)
-	for _, f := range rsMeta.Files {
+	for _, f := range filelist {
 		size += f.StgSize
 	}
 
-	err = IncBackupSize(ctx, b.leadConn, bcp.Name, size)
+	filelistPath := path.Join(bcp.Name, rsMeta.Name, FilelistName)
+	flSize, err := storage.Upload(ctx, filelist, stg, compress.CompressionTypeNone, nil, filelistPath, -1)
+	if err != nil {
+		return errors.Wrapf(err, "upload filelist %q", filelistPath)
+	}
+	l.Info("uploaded: %q %s", filelistPath, fmtSize(flSize))
+
+	err = IncBackupSize(ctx, b.leadConn, bcp.Name, size+flSize)
 	if err != nil {
 		return errors.Wrap(err, "inc backup size")
 	}

pbm/backup/query.go

@@ -208,19 +208,6 @@ func IncBackupSize(ctx context.Context, conn connect.Client, bcpName string, siz
 	return err
 }
 
-func RSSetPhyFiles(ctx context.Context, conn connect.Client, bcpName, rsName string, rs *BackupReplset) error {
-	_, err := conn.BcpCollection().UpdateOne(
-		ctx,
-		bson.D{{"name", bcpName}, {"replsets.name", rsName}},
-		bson.D{
-			{"$set", bson.M{"replsets.$.files": rs.Files}},
-			{"$set", bson.M{"replsets.$.journal": rs.Journal}},
-		},
-	)
-
-	return err
-}
-
 func SetRSLastWrite(conn connect.Client, bcpName, rsName string, ts primitive.Timestamp) error {
 	_, err := conn.BcpCollection().UpdateOne(
 		context.Background(),

pbm/backup/storage.go

@@ -209,6 +209,35 @@ func DeleteBackupFiles(meta *BackupMeta, stg storage.Storage) error {
 
 // DeleteBackupFiles removes backup's artifacts from storage
 func deletePhysicalBackupFiles(meta *BackupMeta, stg storage.Storage) error {
+	if version.HasFilelistFile(meta.PBMVersion) {
+		for i := range meta.Replsets {
+			rs := &meta.Replsets[i]
+			if rs.Files != nil {
+				// it is already fetched
+				continue
+			}
+
+			filelistPath := path.Join(meta.Name, rs.Name, FilelistName)
+			rdr, err := stg.SourceReader(filelistPath)
+			if err != nil {
+				if errors.Is(err, storage.ErrNotExist) {
+					continue
+				}
+
+				return errors.Wrapf(err, "open %q", filelistPath)
+			}
+			defer rdr.Close()
+
+			filelist, err := ReadFilelist(rdr)
+			rdr.Close()
+			if err != nil {
+				return errors.Wrapf(err, "parse filelist")
+			}
+
+			rs.Files = filelist
+		}
+	}
+
 	for _, r := range meta.Replsets {
 		for _, f := range r.Files {
 			fname := meta.Name + "/" + r.Name + "/" + f.Name + meta.Compression.Suffix()