PCSM-230: Fix filtered sync (#162)

inelpandzic · web-flow · commit b91b67b8d5ae · 2025-12-17T02:13:06.000-06:00
diff --git a/AGENTS.md b/AGENTS.md
@@ -0,0 +1,21 @@
+# AGENTS.md - Percona ClusterSync for MongoDB
+
+## Build/Lint/Test Commands
+- **Build**: `make build` (production) or `make test-build` (with race detection)
+- **Lint**: `make lint` (uses golangci-lint with gofmt, gofumpt, goimports)
+- **Test Go**: `go test -race ./...` or `make test`
+- **Test Python**: `poetry run pytest` or `make pytest`
+- **Single Go Test**: `go test -race -run TestName ./package`
+- **Single Python Test**: `poetry run pytest tests/test_file.py::test_name`
+- **Clean**: `make clean`
+
+## Code Style Guidelines
+- **Imports**: Group stdlib, third-party, local (separated by blank lines). Use `goimports` for formatting.
+- **Error Handling**: Use custom `errors` package (`errors.Wrap`, `errors.New`, `errors.Join`). Always wrap errors with context. Return `nil` for nil errors in Wrap.
+- **Types**: Prefer explicit types. Use generics where appropriate. Document exported types with comments.
+- **Naming**: Use camelCase for unexported, PascalCase for exported. Acronyms stay uppercase (e.g., `URI`, `HTTP`).
+- **Comments**: Document all exported functions/types with godoc format. Start with the function/type name.
+- **Linting**: Follow golangci-lint v2 with most linters enabled (see `.golangci.yml`). Use `//nolint:lintername` sparingly with justification.
+- **Context**: Pass `context.Context` as first parameter. Use `util.CtxWithTimeout` for timeout operations.
+- **Formatting**: Use `gofmt` and `gofumpt`. Max line length flexible but keep readable. Use tabs for indentation.
+- **Testing**: Use `testify` for assertions. Add `-race` flag. Python tests use `pytest` with `pymongo`.
diff --git a/hack/change_stream.py b/hack/change_stream.py
@@ -1,13 +1,13 @@
 #!/usr/bin/env python3
-# poetry run python3 .dev/change_stream.py mongodb://adm:pass@rs00:30000
+# poetry run python3 .dev/change_stream.py mongodb://rs00:30000
 
 import sys
 from signal import SIG_DFL, SIGINT, signal
 
 import bson.json_util as json
 import pymongo
 
-MONGODB_URI = "mongodb://adm:pass@rs00:30000"
+MONGODB_URI = "mongodb://rs00:30000"
 
 ignoreCheckpoints = True  # set to False to see all changes, including checkpoints
 
@@ -26,7 +26,7 @@
         del change["wallTime"]
 
         ns = change["ns"]["db"]
-        
+
         if ignoreCheckpoints and ns == "percona_link_mongodb":
             continue
 
diff --git a/hack/compare-all.py b/hack/compare-all.py
@@ -7,11 +7,11 @@
 from pymongo import ASCENDING, MongoClient
 from pymongo.collection import Collection
 
-SRC_URI = "mongodb://adm:pass@rs00:30000"
-TGT_URI = "mongodb://adm:pass@rs10:30100"
+SRC_URI = "mongodb://rs00:30000"
+TGT_URI = "mongodb://rs10:30100"
 
-SRC_SH_URI = "mongodb://adm:pass@src-mongos:27017"
-TGT_SH_URI = "mongodb://adm:pass@tgt-mongos:29017"
+SRC_SH_URI = "mongodb://src-mongos:27017"
+TGT_SH_URI = "mongodb://tgt-mongos:29017"
 
 src = pymongo.MongoClient(SRC_SH_URI)
 tgt = pymongo.MongoClient(TGT_SH_URI)
diff --git a/hack/drop-all.py b/hack/drop-all.py
@@ -2,11 +2,11 @@
 
 import pymongo
 
-SRC_URI = "mongodb://adm:pass@rs00:30000"
-TGT_URI = "mongodb://adm:pass@rs10:30100"
+SRC_URI = "mongodb://rs00:30000"
+TGT_URI = "mongodb://rs10:30100"
 
-SRC_SH_URI = "mongodb://adm:pass@src-mongos:27017"
-TGT_SH_URI = "mongodb://adm:pass@tgt-mongos:29017"
+SRC_SH_URI = "mongodb://src-mongos:27017"
+TGT_SH_URI = "mongodb://tgt-mongos:29017"
 
 src = pymongo.MongoClient(SRC_SH_URI)
 tgt = pymongo.MongoClient(TGT_SH_URI)
diff --git a/hack/oplog_stream.py b/hack/oplog_stream.py
@@ -4,7 +4,7 @@
 import pymongo
 
 if __name__ == "__main__":
-    MONGODB_URI = "mongodb://adm:pass@rs10:30100,rs11:30101,rs12:30102"
+    MONGODB_URI = "mongodb://rs10:30100,rs11:30101,rs12:30102"
     m = pymongo.MongoClient(MONGODB_URI, readPreference="primary")
     for change in m.local["oplog.rs"].find():
         del change["_id"]
diff --git a/hack/reference.md b/hack/reference.md
@@ -10,7 +10,7 @@ host.docker.internal:2242/metrics
 ## MongoLink CLI
 
 ```shell
-go install . && pml --source="mongodb://adm:pass@rs00:30000" --target="mongodb://adm:pass@rs10:30100" --reset-state --log-level="debug"
+go install . && pcsm --source="mongodb://rs00:30000" --target="mongodb://rs10:30100" --reset-state --log-level="debug"
 ```
 
 
diff --git a/hack/rs/generate-data.py b/hack/rs/generate-data.py
@@ -4,7 +4,7 @@
 import pymongo
 from pymongo import MongoClient
 
-MONGODB_URI = "mongodb://adm:pass@rs00:30000"
+MONGODB_URI = "mongodb://rs00:30000"
 client = MongoClient(MONGODB_URI)
 db = client["stress_test_db"]
 
diff --git a/hack/rs/generate-replication-data.py b/hack/rs/generate-replication-data.py
@@ -7,7 +7,7 @@
 from bson import ObjectId, BSON
 from datetime import datetime
 
-uri = "mongodb://adm:pass@rs00:30000"
+uri = "mongodb://rs00:30000"
 # uri="mongodb://inel:m0sl1RoYDXkSLEse@cluster0.vrfly.mongodb.net/"
 
 
diff --git a/hack/rs/generate-replication-txn.py b/hack/rs/generate-replication-txn.py
@@ -6,7 +6,7 @@
 import threading
 
 # MongoDB connection
-uri = "mongodb://adm:pass@rs00:30000"
+uri = "mongodb://rs00:30000"
 client = MongoClient(uri)  # Ensure you have a replica set
 
 # Databases and collections
diff --git a/hack/rs/load.py b/hack/rs/load.py
@@ -7,7 +7,7 @@
 from bson import ObjectId, BSON
 from datetime import datetime
 
-uri="mongodb://adm:pass@rs00:30000"
+uri="mongodb://rs00:30000"
 # uri="mongodb://inel:m0sl1RoYDXkSLEse@cluster0.vrfly.mongodb.net/"
 
 def log(msg):
diff --git a/hack/sh/scripts/drop-all.py b/hack/sh/scripts/drop-all.py
@@ -2,8 +2,8 @@
 
 import pymongo
 
-SRC_URI = "mongodb://adm:pass@src-mongos:27017"
-TGT_URI = "mongodb://adm:pass@tgt-mongos:29017"
+SRC_URI = "mongodb://src-mongos:27017"
+TGT_URI = "mongodb://tgt-mongos:29017"
 
 src = pymongo.MongoClient(SRC_URI)
 tgt = pymongo.MongoClient(TGT_URI)
diff --git a/hack/sh/scripts/load-hashed.py b/hack/sh/scripts/load-hashed.py
@@ -7,7 +7,7 @@
 from bson import ObjectId, BSON
 from datetime import datetime
 
-uri="mongodb://adm:pass@src-mongos:27017"
+uri="mongodb://src-mongos:27017"
 # uri="mongodb://inel:m0sl1RoYDXkSLEse@cluster0.vrfly.mongodb.net/"
 
 def create_sharded_collection(db, collection_name, shard_key):
@@ -59,4 +59,4 @@ def worker(args):
         for collection_name, shard_key in collections
     ]
     with multiprocessing.Pool(processes=len(collections)) as pool:
-        pool.map(worker, args_list)
+        pool.map(worker, args_list)
diff --git a/hack/sh/scripts/load-ranged.py b/hack/sh/scripts/load-ranged.py
@@ -7,7 +7,7 @@
 from bson import ObjectId, BSON
 from datetime import datetime
 
-uri = "mongodb://adm:pass@src-mongos:27017"
+uri = "mongodb://src-mongos:27017"
 # uri="mongodb://inel:m0sl1RoYDXkSLEse@cluster0.vrfly.mongodb.net/"
 
 
diff --git a/pcsm/clone.go b/pcsm/clone.go
@@ -618,6 +618,8 @@ func (c *Clone) collectSizeMap(ctx context.Context) error {
 					continue
 				}
 
+				lg.With(log.NS(db, spec.Name)).Infof("Namespace %q included", db+"."+spec.Name)
+
 				collGrp.Go(func() error {
 					ns := db + "." + spec.Name
 					if spec.Type == topo.TypeView {
diff --git a/sel/sel.go b/sel/sel.go
@@ -3,6 +3,8 @@ package sel
 import (
 	"slices"
 	"strings"
+
+	"github.com/percona/percona-clustersync-mongodb/log"
 )
 
 // NSFilter returns true if a namespace is allowed.
@@ -13,47 +15,46 @@ func AllowAllFilter(string, string) bool {
 }
 
 func MakeFilter(include, exclude []string) NSFilter {
+	lg := log.New("filter")
+
 	if len(include) == 0 && len(exclude) == 0 {
 		return AllowAllFilter
 	}
 
-	includeFilter := doMakeFitler(include)
-	excludeFilter := doMakeFitler(exclude)
+	includeFilter := doMakeFilter(include)
+	excludeFilter := doMakeFilter(exclude)
+
+	lg.Infof("Include filter: %v", strings.Join(include, ", "))
+	lg.Infof("Exclude filter: %v", strings.Join(exclude, ", "))
 
 	return func(db, coll string) bool {
 		_, dbIncluded := includeFilter[db]
-		_, dbExcluded := excludeFilter[db]
 
 		nsIncluded := len(includeFilter) > 0 && includeFilter.Has(db, coll)
 		nsExcluded := len(excludeFilter) > 0 && excludeFilter.Has(db, coll)
 
-		if nsIncluded && dbIncluded && !nsExcluded {
-			// If the namespace is included, it is allowed.
-			// Also make sure that the namespace is not excluded,
-			// because exclusion takes precedence.
-			return true
-		}
-
-		if dbIncluded && !nsExcluded {
-			// If the database is included in the filter,
-			// but the namespace is not included, it is not allowed.
-			// Also make sure that the namespace is not excluded,
-			// because exclusion takes precedence.
+		// Exclusion takes precedence - if explicitly excluded, deny immediately.
+		if nsExcluded {
 			return false
 		}
 
-		if nsExcluded && dbExcluded {
-			// If the namespace is excluded, it is not allowed.
+		// If include filter exists, use whitelist logic (deny by default).
+		if len(includeFilter) > 0 {
+			// Allow if namespace is explicitly included.
+			if nsIncluded {
+				return true
+			}
+			// DB is in include filter but collection is not - deny.
+			if dbIncluded {
+				return false
+			}
+			// DB not in include filter at all - deny.
 			return false
 		}
 
-		if dbExcluded {
-			// If the database is included in the filter,
-			// but the namespace is not excluded, it is allowed.
-			return true
-		}
+		// No include filter (exclude-only or no filters).
+		// Allow by default since exclusions are already handled above.
 
-		// If the namespace is not present in either filter, it is allowed by default.
 		return true
 	}
 }
@@ -73,7 +74,7 @@ func (f filterMap) Has(db, coll string) bool {
 	return slices.Contains(list, coll) // only if explcitly listed
 }
 
-func doMakeFitler(filter []string) filterMap {
+func doMakeFilter(filter []string) filterMap {
 	// keys are database names. values are list collections that belong to the db.
 	// if a key contains empty/nil, whole db is included (all its collections).
 	filterMap := make(map[string][]string)
diff --git a/sel/sel_test.go b/sel/sel_test.go
diff --git a/tests/test_selective.py b/tests/test_selective.py
diff --git a/tests/test_selective_sharded.py b/tests/test_selective_sharded.py
