PBM-1358: support selective backup and restore with configshard (#1037)

Author: defbin

cmd/pbm/backup.go

@@ -348,6 +348,7 @@ type bcpReplDesc struct {
 	LastWriteTime      string              `json:"last_write_time" yaml:"last_write_time"`
 	LastTransitionTime string              `json:"last_transition_time" yaml:"last_transition_time"`
 	IsConfigSvr        *bool               `json:"configsvr,omitempty" yaml:"configsvr,omitempty"`
+	IsConfigShard      *bool               `json:"configshard,omitempty" yaml:"configshard,omitempty"`
 	SecurityOpts       *topo.MongodOptsSec `json:"security,omitempty" yaml:"security,omitempty"`
 	Error              *string             `json:"error,omitempty" yaml:"error,omitempty"`
 	Collections        []string            `json:"collections,omitempty" yaml:"collections,omitempty"`
@@ -442,6 +443,7 @@ func describeBackup(
 			Name:               r.Name,
 			Node:               r.Node,
 			IsConfigSvr:        r.IsConfigSvr,
+			IsConfigShard:      r.IsConfigShard,
 			Status:             r.Status,
 			LastWriteTS:        int64(r.LastWriteTS.T),
 			LastTransitionTS:   r.LastTransitionTS,

pbm/archive/backup.go

@@ -19,7 +19,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/version"
 )
 
-var configCollectionsToKeep = []string{
+var RouterConfigCollections = []string{
 	"chunks",
 	"collections",
 	"databases",
@@ -191,7 +191,7 @@ func (bcp *backupImpl) listDBNamespaces(ctx context.Context, db string) ([]*Name
 	case "admin":
 		filter = bson.D{{"name", bson.M{"$ne": "system.keys"}}}
 	case "config":
-		filter = bson.D{{"name", bson.M{"$in": configCollectionsToKeep}}}
+		filter = bson.D{{"name", bson.M{"$in": RouterConfigCollections}}}
 	}
 
 	cur, err := bcp.conn.Database(db).ListCollections(ctx, filter)
@@ -279,6 +279,14 @@ func (bcp *backupImpl) dumpAllCollections(ctx context.Context, nss []*NamespaceV
 }
 
 func (bcp *backupImpl) dumpCollection(ctx context.Context, ns *NamespaceV2) error {
+	count, err := bcp.conn.Database(ns.DB).Collection(ns.Name).EstimatedDocumentCount(ctx)
+	if err != nil {
+		return errors.Wrap(err, "estimate document count")
+	}
+	if count == 0 {
+		return nil
+	}
+
 	file, err := bcp.newFile(ns.NS())
 	if err != nil {
 		return errors.Wrap(err, "new file")

pbm/archive/conv.go

@@ -68,6 +68,7 @@ func convertMetaToV1(metaV2 *ArchiveMetaV2) (*archiveMeta, error) {
 			ServerVersion: metaV2.ServerVersion,
 			ToolVersion:   metaV2.Version,
 		},
+		Namespaces: []*Namespace{},
 	}
 
 	for _, coll := range metaV2.Namespaces {

pbm/backup/backup.go

@@ -166,18 +166,6 @@ func (b *Backup) Init(
 //
 //nolint:nonamedreturns
 func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l log.LogEvent) (err error) {
-	if b.brief.Sharded &&
-		b.brief.Version.IsConfigShardSupported() &&
-		util.IsSelective(bcp.Namespaces) {
-		hasConfigShard, err := topo.HasConfigShard(ctx, b.leadConn)
-		if err != nil {
-			return errors.Wrap(err, "check for Config Shard")
-		}
-		if hasConfigShard {
-			return errors.New("selective backup is not supported with Config Shard")
-		}
-	}
-
 	inf, err := topo.GetNodeInfoExt(ctx, b.nodeConn)
 	if err != nil {
 		return errors.Wrap(err, "get cluster info")
@@ -199,6 +187,14 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 	}
 	if v := inf.IsConfigSrv(); v {
 		rsMeta.IsConfigSvr = &v
+
+		isConfigShard, err := topo.HasConfigShard(ctx, b.leadConn)
+		if err != nil {
+			return errors.Wrap(err, "has configshard")
+		}
+		if isConfigShard {
+			rsMeta.IsConfigShard = &isConfigShard
+		}
 	}
 
 	stg, err := util.StorageFromConfig(&b.config.Storage, inf.Me, l)

pbm/backup/logical.go

@@ -41,19 +41,7 @@ func (b *Backup) doLogical(
 			return errors.Wrap(err, "check for timeseries")
 		}
 	}
-
-	var db, coll string
-	if util.IsSelective(bcp.Namespaces) {
-		// for selective backup, configsvr does not hold any data.
-		// only some collections from config db is required to restore cluster state
-		if inf.IsConfigSrv() {
-			db = "config"
-		} else {
-			db, coll = util.ParseNS(bcp.Namespaces[0])
-		}
-	}
-
-	nssSize, err := getNamespacesSize(ctx, b.nodeConn, db, coll)
+	nssSize, err := getNamespacesSize(ctx, b.nodeConn, bcp.Namespaces)
 	if err != nil {
 		return errors.Wrap(err, "get namespaces size")
 	}
@@ -153,8 +141,9 @@ func (b *Backup) doLogical(
 				return errors.Wrap(err, "fetch uuids")
 			}
 
-			nsFilter = makeConfigsvrNSFilter()
 			docFilter = makeConfigsvrDocFilter(bcp.Namespaces, chunkSelector)
+			nsFilter = makeConfigsvrNSFilter(bcp.Namespaces)
+
 		} else {
 			nsFilter = util.MakeSelectedPred(bcp.Namespaces)
 		}
@@ -358,17 +347,16 @@ func createBackupChunkSelector(ctx context.Context, m connect.Client, nss []stri
 	return chunkSelector, nil
 }
 
-func makeConfigsvrNSFilter() archive.NSFilterFn {
+func makeConfigsvrNSFilter(nss []string) archive.NSFilterFn {
+	selected := make([]string, len(nss)+len(archive.RouterConfigCollections))
+	n := copy(selected, nss)
+
 	// list of required namespaces for further selective restore
-	allowed := map[string]bool{
-		"config.databases":   true,
-		"config.collections": true,
-		"config.chunks":      true,
+	for i, name := range archive.RouterConfigCollections {
+		selected[n+i] = "config." + name
 	}
 
-	return func(ns string) bool {
-		return allowed[ns]
-	}
+	return util.MakeSelectedPred(selected)
 }
 
 func makeConfigsvrDocFilter(nss []string, selector util.ChunkSelector) archive.DocFilterFn {
@@ -395,33 +383,27 @@ func makeConfigsvrDocFilter(nss []string, selector util.ChunkSelector) archive.D
 	}
 }
 
-func getNamespacesSize(ctx context.Context, m *mongo.Client, db, coll string) (map[string]int64, error) {
+func getNamespacesSize(ctx context.Context, m *mongo.Client, nss []string) (map[string]int64, error) {
 	rv := make(map[string]int64)
 
-	q := bson.D{}
-	if db != "" {
-		q = append(q, bson.E{"name", db})
-	}
-	dbs, err := m.ListDatabaseNames(ctx, q)
+	dbs, err := m.ListDatabaseNames(ctx, bson.D{})
 	if err != nil {
 		return nil, errors.Wrap(err, "list databases")
 	}
 	if len(dbs) == 0 {
 		return rv, nil
 	}
 
+	isSelected := util.MakeSelectedPred(nss)
+
 	mu := sync.Mutex{}
 	eg, ctx := errgroup.WithContext(ctx)
 
 	for _, db := range dbs {
 		db := db
 
 		eg.Go(func() error {
-			q := bson.D{}
-			if coll != "" {
-				q = append(q, bson.E{"name", coll})
-			}
-			res, err := m.Database(db).ListCollectionSpecifications(ctx, q)
+			res, err := m.Database(db).ListCollectionSpecifications(ctx, bson.D{})
 			if err != nil {
 				return errors.Wrapf(err, "list collections for %q", db)
 			}
@@ -432,14 +414,16 @@ func getNamespacesSize(ctx context.Context, m *mongo.Client, db, coll string) (m
 			eg, ctx := errgroup.WithContext(ctx)
 
 			for _, coll := range res {
-				coll := coll
-
 				if coll.Type == "view" || strings.HasPrefix(coll.Name, "system.buckets.") {
 					continue
 				}
 
+				ns := db + "." + coll.Name
+				if !isSelected(ns) {
+					continue
+				}
+
 				eg.Go(func() error {
-					ns := db + "." + coll.Name
 					res := m.Database(db).RunCommand(ctx, bson.D{{"collStats", coll.Name}})
 					if err := res.Err(); err != nil {
 						return errors.Wrapf(err, "collStats %q", ns)

pbm/backup/types.go

@@ -118,6 +118,7 @@ type BackupReplset struct {
 	StartTS          int64               `bson:"start_ts" json:"start_ts"`
 	Status           defs.Status         `bson:"status" json:"status"`
 	IsConfigSvr      *bool               `bson:"iscs,omitempty" json:"iscs,omitempty"`
+	IsConfigShard    *bool               `bson:"configshard,omitempty" json:"configshard,omitempty"`
 	LastTransitionTS int64               `bson:"last_transition_ts" json:"last_transition_ts"`
 	FirstWriteTS     primitive.Timestamp `bson:"first_write_ts" json:"first_write_ts"`
 	LastWriteTS      primitive.Timestamp `bson:"last_write_ts" json:"last_write_ts"`