Merge pull request #1161 from percona/PBM-1497-applyOps-stale-config

PBM-1497: Exclude balancer ops during PITR

Author: boris-ilijic

pbm/oplog/restore.go

@@ -100,6 +100,24 @@ var dontPreserveUUID = []string{
 	"*.system.views",     // timeseries
 }
 
+// ConfigCollToKeep defines a list of collections in the `config`
+// database that PBM will apply oplog events.
+var configCollToKeep = []string{
+	"chunks",
+	"collections",
+	"databases",
+	"shards",
+	"tags",
+	"version",
+}
+
+const settingsColl = "settings"
+
+var settingsToSkip = []string{
+	"balancer",
+	"automerge",
+}
+
 var ErrNoCloningNamespace = errors.New("cloning namespace desn't exist")
 
 // cloneNS has all data related to cloning namespace within oplog
@@ -316,16 +334,23 @@ func (o *OplogRestore) SetCloneNS(ctx context.Context, ns snapshot.CloneNS) erro
 	return nil
 }
 
+// isOpAllowed inspects whether op is allowed from config database point of view.
+// It allows/disallows only specific collections for config database.
+// It disallows balancer settings that would cause the balancer to work during PITR.
 func isOpAllowed(oe *Record) bool {
 	coll, ok := strings.CutPrefix(oe.Namespace, "config.")
 	if !ok {
 		return true // OK: not a "config" database. allow any ops
 	}
 
-	if slices.Contains(dumprestore.ConfigCollectionsToKeep, coll) {
+	if slices.Contains(configCollToKeep, coll) {
 		return true // OK: create/update/delete a doc
 	}
 
+	if coll == settingsColl {
+		return isConfigSettingAllowed(oe)
+	}
+
 	if coll != "$cmd" || len(oe.Object) == 0 {
 		return false // other collection is not allowed
 	}
@@ -342,6 +367,20 @@ func isOpAllowed(oe *Record) bool {
 	return false
 }
 
+// isConfigSettingAllowed filter out entries from config.settings collection
+func isConfigSettingAllowed(oe *Record) bool {
+	for _, e := range oe.Query {
+		if e.Key != "_id" {
+			continue
+		}
+		setting, _ := e.Value.(string)
+		if slices.Contains(settingsToSkip, setting) {
+			return false
+		}
+	}
+	return true // allow setting from config.settings
+}
+
 func (o *OplogRestore) isOpSelected(oe *Record) bool {
 	if o.includeNS == nil || o.includeNS[""] != nil {
 		return true
@@ -412,6 +451,8 @@ func (o *OplogRestore) isOpForCloning(oe *db.Oplog) bool {
 	return false
 }
 
+// isOpExcluded check if collection is excluded.
+// Mostly refers PBM's control collections, but also some config and admin ones.
 func (o *OplogRestore) isOpExcluded(oe *Record) bool {
 	if o.excludeNS == nil {
 		return false

pbm/oplog/restore_test.go

@@ -141,6 +141,85 @@ func TestIsOpForCloning(t *testing.T) {
 	}
 }
 
+func TestIsOpAllowed(t *testing.T) {
+	testCases := []struct {
+		desc      string
+		entry     *db.Oplog
+		opAllowed bool
+	}{
+		{
+			desc:      "any other than config",
+			entry:     createInsertOp(t, "db1.c1"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.chunks",
+			entry:     createInsertOp(t, "config.chunks"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.collections",
+			entry:     createInsertOp(t, "config.collections"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.databases",
+			entry:     createInsertOp(t, "config.databases"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.shards",
+			entry:     createInsertOp(t, "config.shards"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.tags",
+			entry:     createUpdateOp(t, "config.tags"),
+			opAllowed: true,
+		},
+		{
+			desc:      "config.version",
+			entry:     createDeleteOp(t, "config.version"),
+			opAllowed: true,
+		},
+		{
+			desc:      "dissalowed config collection",
+			entry:     createInsertOp(t, "config.mongos"),
+			opAllowed: false,
+		},
+		{
+			desc:      "wrong config.settings doc will not break",
+			entry:     createDeleteOp(t, "config.settings"),
+			opAllowed: true,
+		},
+		{
+			desc:      "settings doc for balancer",
+			entry:     configSettingsBalancerEntry(),
+			opAllowed: false,
+		},
+		{
+			desc:      "settings doc for automerge",
+			entry:     configSettingsAutomergeEntry(),
+			opAllowed: false,
+		},
+		{
+			desc:      "allowed settings doc",
+			entry:     configSettingsAnyOtherEntry(),
+			opAllowed: true,
+		},
+	}
+
+	for _, tC := range testCases {
+		t.Run(tC.desc, func(t *testing.T) {
+			res := isOpAllowed(tC.entry)
+			if res != tC.opAllowed {
+				t.Errorf("%s: for entry: %+v isOpAllowed is: %t, but it should be opposite",
+					tC.desc, tC.entry, tC.opAllowed)
+			}
+		})
+	}
+}
+
 func TestApply(t *testing.T) {
 	t.Run("collection restore", func(t *testing.T) {
 		testCases := []struct {
@@ -638,3 +717,29 @@ func replaceNsWithinOpEntry(t *testing.T, jsonEntry, ns string) *db.Oplog {
 	}
 	return &oe
 }
+
+func configSettingsBalancerEntry() *db.Oplog {
+	return &db.Oplog{
+		Operation: "i",
+		Namespace: "config.settings",
+		Object:    bson.D{{"_id", "balancer"}, {"mode", "off"}, {"stopped", true}},
+		Query:     bson.D{{"_id", "balancer"}},
+	}
+}
+
+func configSettingsAutomergeEntry() *db.Oplog {
+	return &db.Oplog{
+		Operation: "u",
+		Namespace: "config.settings",
+		Query:     bson.D{{"_id", "automerge"}},
+	}
+}
+
+func configSettingsAnyOtherEntry() *db.Oplog {
+	return &db.Oplog{
+		Operation: "i",
+		Namespace: "config.settings",
+		Object:    bson.D{{"_id", "chunksize"}, {"value", 128}},
+		Query:     bson.D{{"_id", "chunksize"}},
+	}
+}