REP-2943 / REP-2944: Add filter parameter to /check endpoint and enable query filtering (#4)

The `/check` endpoint takes a JSON filter. The verifier parses the JSON filter to `map[string]any`. The verifier starts a change stream without query filtering. The verifier generates a recheck document for a change event of a filtered document, but the document should be filtered out during fetching and thus not compared by the verifier.

Author: tdq45gj

README.md

@@ -63,7 +63,7 @@ To set a port, use `--serverPort <port number>`. The default is 27020.
 
 
 
-1. After launching the verifier (see above), you can send it requests to get it to start verifying. The verification process is started by using the `check`command. The verification process will keep running until you tell the verifier to stop. It will keep track of the inconsistencies it has found and will keep checking those inconsistencies hoping that eventually they will resolve.
+1. After launching the verifier (see above), you can send it requests to get it to start verifying. The verification process is started by using the `check`command. An [optional `filter` parameter](#document-filtering) can be passed within the `check` request body to only check documents within that filter. The verification process will keep running until you tell the verifier to stop. It will keep track of the inconsistencies it has found and will keep checking those inconsistencies hoping that eventually they will resolve.
 
     ```
     curl -H "Content-Type: application/json" -X POST -d '{}' http://127.0.0.1:27020/api/v1/check
@@ -98,7 +98,6 @@ The verifier will now check to completion to make sure that there are no inconsi
 
 The verifier’s iterative process can handle data changes while it is running, until you hit the writesOff endpoint.  However, it cannot handle DDL commands.  If the verifier receives a DDL change stream event (drop, dropDatabase, rename), the verification will fail.  If an untracked DDL event (create, createIndexes, dropIndexes, modify) occurs, the verifier may miss the change. 
 
-
 # Benchmarking Results
 
 Ran on m6id.metal + M40 with 3 replica sets
@@ -131,6 +130,22 @@ The migration-verifier has two steps:
     5. Every document to check is written with a generation number. A checking round checks documents for a specific generation. When a check round begins, we start writing new documents with a new generation number
     6. The verifier fetches all collection/index/view information on the source and destination and confirms they are identical in every generation. This is duplicated work, but it's fast and convenient for the code.
 
+# Document Filtering
+
+Document filtering can be enabled by passing a `filter` parameter in the `check` request body when starting a check. The filter takes a JSON query. The query syntax is identical to the [read operation query syntax](https://www.mongodb.com/docs/manual/tutorial/query-documents/#std-label-read-operations-query-argument). For example, running the following command makes the verifier check to only check documents within the filter `{"inFilter": {"$ne": false}}` for _all_ namespaces:
+
+```
+curl -H "Content-Type: application/json" -X POST -d '{{"filter": {"inFilter": {"$ne": false}}}}' http://127.0.0.1:27020/api/v1/check
+```
+If a checking is started with the above filter, the table below summarizes the verifier's behavior: 
+
+| Source Document                                   | Destination Document                              | Verifier's Behavior                         |
+|---------------------------------------------------|---------------------------------------------------|---------------------------------------------|
+| `{"_id": <id>, "inFilter": true, "diff": "src"}`  | `{"_id": <id>, "inFilter": true, "diff": "dst"}`  | ❗ (Finds a document with differing content) |
+| `{"_id": <id>, "inFilter": false, "diff": "src"}` | `{"_id": <id>, "inFilter": false, "diff": "dst"}` | ✅ (Skips a document)                        |
+| `{"_id": <id>, "inFilter": true, "diff": "src"}`  | `{"_id": <id>, "inFilter": false, "diff": "dst"}` | ❗ (Finds a document missing on Destination) |
+| `{"_id": <id>, "inFilter": false, "diff": "src"}` | `{"_id": <id>, "inFilter": true, "diff": "dst"}`  | ❗ (Finds a document missing on Source)      |
+
 # Checking Failures
 
 Because the algorithm is generational, the only failures we care about are in the last generation to run. The first goal is to find the last generation:

internal/partitions/partition.go

@@ -137,7 +137,7 @@ func (p *Partition) FindCmd(
 // (e.g. use the partitions on the source to read the destination for verification)
 // If the passed-in buildinfo indicates a mongodb version < 5.0, type bracketing is not used.
 // filterAndPredicates is a slice of filter criteria that's used to construct the "filter" field in the find option.
-func (p *Partition) GetFindOptions(buildInfo *bson.M, filterAndPredicates []bson.D) bson.D {
+func (p *Partition) GetFindOptions(buildInfo *bson.M, filterAndPredicates bson.A) bson.D {
 	if p == nil {
 		if len(filterAndPredicates) > 0 {
 			return bson.D{{"filter", bson.D{{"$and", filterAndPredicates}}}}
@@ -190,16 +190,16 @@ func (p *Partition) GetFindOptions(buildInfo *bson.M, filterAndPredicates []bson
 func (p *Partition) filterWithNoTypeBracketing() bson.D {
 	// We use $expr to avoid type bracketing and allow comparison of different _id types,
 	// and $literal to avoid MQL injection from an _id's value.
-	return bson.D{{"$and", bson.A{
+	return bson.D{{"$and", []bson.D{
 		// All _id values >= lower bound.
-		bson.D{{"$expr", bson.D{
+		{{"$expr", bson.D{
 			{"$gte", bson.A{
 				"$_id",
 				bson.D{{"$literal", p.Key.Lower}},
 			}},
 		}}},
 		// All _id values <= upper bound.
-		bson.D{{"$expr", bson.D{
+		{{"$expr", bson.D{
 			{"$lte", bson.A{
 				"$_id",
 				bson.D{{"$literal", p.Upper}},
@@ -212,10 +212,10 @@ func (p *Partition) filterWithNoTypeBracketing() bson.D {
 // partition.  This filter will not properly handle mixed-type _ids -- if the upper and lower
 // bounds are of different types (except minkey/maxkey), nothing will be returned.
 func (p *Partition) filterWithTypeBracketing() bson.D {
-	return bson.D{{"$and", bson.A{
+	return bson.D{{"$and", []bson.D{
 		// All _id values >= lower bound.
-		bson.D{{"_id", bson.D{{"$gte", p.Key.Lower}}}},
+		{{"_id", bson.D{{"$gte", p.Key.Lower}}}},
 		// All _id values <= upper bound.
-		bson.D{{"_id", bson.D{{"$lte", p.Upper}}}},
+		{{"_id", bson.D{{"$lte", p.Upper}}}},
 	}}}
 }

internal/partitions/partition_test.go

@@ -134,14 +134,14 @@ func makeExpectedFilter(lower, upper interface{}) bson.D {
 	return bson.D{{"$and", bson.A{
 		bson.D{{"$and", []bson.D{
 			// All _id values >= lower bound.
-			bson.D{{"$expr", bson.D{
+			{{"$expr", bson.D{
 				{"$gte", bson.A{
 					"$_id",
 					bson.D{{"$literal", lower}},
 				}},
 			}}},
 			// All _id values <= upper bound.
-			bson.D{{"$expr", bson.D{
+			{{"$expr", bson.D{
 				{"$lte", bson.A{
 					"$_id",
 					bson.D{{"$literal", upper}},
@@ -155,9 +155,9 @@ func makeExpectedFilterWithTypeBracketing(lower, upper interface{}) bson.D {
 	return bson.D{{"$and", bson.A{
 		bson.D{{"$and", []bson.D{
 			// All _id values >= lower bound.
-			bson.D{{"_id", bson.D{{"$gte", lower}}}},
+			{{"_id", bson.D{{"$gte", lower}}}},
 			// All _id values <= upper bound.
-			bson.D{{"_id", bson.D{{"$lte", upper}}}},
+			{{"_id", bson.D{{"$lte", upper}}}},
 		}}},
 	}}}
 }

internal/verifier/check.go

@@ -33,9 +33,9 @@ var verificationStatusCheckInterval time.Duration = 15 * time.Second
 // testChan is a pair of channels for coordinating generations in tests.
 // testChan[0] is a channel signalled when when a generation is complete
 // testChan[1] is a channel signalled when Check should continue with the next generation.
-func (verifier *Verifier) Check(ctx context.Context) {
+func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 	go func() {
-		err := verifier.CheckDriver(ctx, nil)
+		err := verifier.CheckDriver(ctx, filter)
 		if err != nil {
 			verifier.logger.Fatal().Err(err).Msgf("Fatal error in generation %d", verifier.generation)
 		}
@@ -122,7 +122,7 @@ func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 	return nil
 }
 
-func (verifier *Verifier) CheckDriver(ctx context.Context, filter bson.D, testChan ...chan struct{}) error {
+func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any, testChan ...chan struct{}) error {
 	verifier.mux.Lock()
 	if verifier.running {
 		verifier.mux.Unlock()

internal/verifier/migration_verifier.go

@@ -135,7 +135,7 @@ type Verifier struct {
 	// A user-defined $match-compatible document-level query filter.
 	// The filter is applied to all namespaces in both initial checking and iterative checking.
 	// The verifier only checks documents within the filter.
-	globalFilter bson.D
+	globalFilter map[string]any
 }
 
 // VerificationStatus holds the Verification Status
@@ -397,18 +397,20 @@ func (verifier *Verifier) getGenerationWhileLocked() (int, bool) {
 	return verifier.generation, verifier.lastGeneration
 }
 
-func (verifier *Verifier) maybeAppendGlobalFilterToPredicates(predicates []bson.D) []bson.D {
+func (verifier *Verifier) maybeAppendGlobalFilterToPredicates(predicates bson.A) bson.A {
 	if verifier.globalFilter == nil {
+		verifier.logger.Debug().Msg("No filter to append; globalFilter is nil")
 		return predicates
 	}
+	verifier.logger.Debug().Str("filter", fmt.Sprintf("%v", verifier.globalFilter)).Msg("Appending filter to find query")
 	return append(predicates, verifier.globalFilter)
 }
 
 func (verifier *Verifier) getDocumentsCursor(ctx context.Context, collection *mongo.Collection, buildInfo *bson.M,
 	startAtTs *primitive.Timestamp, task *VerificationTask) (*mongo.Cursor, error) {
 	var findOptions bson.D
 	runCommandOptions := options.RunCmd()
-	var andPredicates []bson.D
+	var andPredicates bson.A
 
 	if len(task.Ids) > 0 {
 		andPredicates = append(andPredicates, bson.D{{"_id", bson.M{"$in": task.Ids}}})
@@ -642,6 +644,9 @@ func (verifier *Verifier) ProcessVerifyTask(workerNum int, task *VerificationTas
 					ids = append(ids, v.ID)
 					dataSizes = append(dataSizes, v.dataSize)
 				}
+				// Update ids of the failed task so that only ids from mismatches are reported.
+				// Ids of matching documents are discarded and hidden from the mismatching documents report.
+				task.Ids = ids
 				err := verifier.InsertFailedCompareRecheckDocs(task.QueryFilter.Namespace, ids, dataSizes)
 				if err != nil {
 					verifier.logger.Error().Msgf("[Worker %d] Error inserting document mismatch into Recheck queue: %+v", workerNum, err)

internal/verifier/migration_verifier_test.go

@@ -210,7 +210,7 @@ func (suite *MultiDataVersionTestSuite) TestVerifierFetchDocuments() {
 	expectTwoCommonDocs(srcDocumentMap, dstDocumentMap)
 
 	// Test fetchDocuments for ids with a global filter.
-	verifier.globalFilter = bson.D{{"num", bson.D{{"$lt", 100}}}}
+	verifier.globalFilter = map[string]any{"num": map[string]any{"$lt": 100}}
 	srcDocumentMap, dstDocumentMap, err = verifier.fetchDocuments(task)
 	suite.Require().NoError(err)
 	expectOneCommonDoc(srcDocumentMap, dstDocumentMap)
@@ -220,7 +220,7 @@ func (suite *MultiDataVersionTestSuite) TestVerifierFetchDocuments() {
 		Ns:       &partitions.Namespace{DB: "keyhole", Coll: "dealers"},
 		IsCapped: false,
 	}
-	verifier.globalFilter = bson.D{{"num", bson.D{{"$lt", 100}}}}
+	verifier.globalFilter = map[string]any{"num": map[string]any{"$lt": 100}}
 	srcDocumentMap, dstDocumentMap, err = verifier.fetchDocuments(task)
 	suite.Require().NoError(err)
 	expectOneCommonDoc(srcDocumentMap, dstDocumentMap)
@@ -1428,3 +1428,97 @@ func (suite *MultiDataVersionTestSuite) TestGenerationalRechecking() {
 	// there should be a failure from the src insert
 	suite.Require().Equal(VerificationStatus{TotalTasks: 1, FailedTasks: 1}, *status)
 }
+
+func (suite *MultiDataVersionTestSuite) TestVerifierWithFilter() {
+	zerolog.SetGlobalLevel(zerolog.DebugLevel)
+
+	filter := map[string]any{"inFilter": map[string]any{"$ne": false}}
+	verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
+	verifier.SetSrcNamespaces([]string{"testDb1.testColl1"})
+	verifier.SetDstNamespaces([]string{"testDb2.testColl3"})
+	verifier.SetNamespaceMap()
+	verifier.SetIgnoreBSONFieldOrder(true)
+
+	ctx := context.Background()
+
+	srcColl := suite.srcMongoClient.Database("testDb1").Collection("testColl1")
+	dstColl := suite.dstMongoClient.Database("testDb2").Collection("testColl3")
+
+	// Documents {_id: 1} should match, {_id: 2} should be ignored because it's not in the filter.
+	_, err := srcColl.InsertOne(ctx, bson.M{"_id": 1, "x": 42, "inFilter": true})
+	suite.Require().NoError(err)
+	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 2, "x": 53, "inFilter": false})
+	suite.Require().NoError(err)
+	_, err = dstColl.InsertOne(ctx, bson.M{"_id": 1, "x": 42, "inFilter": true})
+	suite.Require().NoError(err)
+
+	checkDoneChan := make(chan struct{})
+	checkContinueChan := make(chan struct{})
+	go func() {
+		err := verifier.CheckDriver(ctx, filter, checkDoneChan, checkContinueChan)
+		suite.Require().NoError(err)
+	}()
+
+	waitForTasks := func() *VerificationStatus {
+		status, err := verifier.GetVerificationStatus()
+		suite.Require().NoError(err)
+
+		for status.TotalTasks == 0 && verifier.generation < 10 {
+			suite.T().Logf("TotalTasks is 0 (generation=%d); waiting another generation …", verifier.generation)
+			checkContinueChan <- struct{}{}
+			<-checkDoneChan
+			status, err = verifier.GetVerificationStatus()
+			suite.Require().NoError(err)
+		}
+		return status
+	}
+
+	// Wait for one generation to finish.
+	<-checkDoneChan
+	status := waitForTasks()
+	suite.Require().Equal(VerificationStatus{TotalTasks: 2, FailedTasks: 0, CompletedTasks: 2}, *status)
+
+	// Insert another document that is not in the filter.
+	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 3, "x": 43, "inFilter": false})
+	suite.Require().NoError(err)
+
+	// Tell check to start the next generation.
+	checkContinueChan <- struct{}{}
+
+	// Wait for generation to finish.
+	<-checkDoneChan
+	status = waitForTasks()
+	// There should be no failures, since the inserted document is not in the filter.
+	suite.Require().Equal(VerificationStatus{TotalTasks: 1, CompletedTasks: 1}, *status)
+
+	// Now insert in the source, this should come up next generation.
+	_, err = srcColl.InsertOne(ctx, bson.M{"_id": 4, "x": 44, "inFilter": true})
+	suite.Require().NoError(err)
+
+	// Tell check to start the next generation.
+	checkContinueChan <- struct{}{}
+
+	// Wait for one generation to finish.
+	<-checkDoneChan
+	status = waitForTasks()
+
+	// There should be a failure from the src insert of a document in the filter.
+	suite.Require().Equal(VerificationStatus{TotalTasks: 1, FailedTasks: 1}, *status)
+
+	// Now patch up the destination.
+	_, err = dstColl.InsertOne(ctx, bson.M{"_id": 4, "x": 44, "inFilter": true})
+	suite.Require().NoError(err)
+
+	// Continue.
+	checkContinueChan <- struct{}{}
+
+	// Wait for it to finish again, this should be a clean run.
+	<-checkDoneChan
+	status = waitForTasks()
+
+	// There should be no failures now, since they are equivalent at this point in time.
+	suite.Require().Equal(VerificationStatus{TotalTasks: 1, CompletedTasks: 1}, *status)
+
+	// Turn writes off.
+	verifier.WritesOff(ctx)
+}

internal/verifier/web_server.go

@@ -21,7 +21,7 @@ const RequestInProgressErrorDescription = "Another request is currently in progr
 
 // MigrationVerifierAPI represents the interaction webserver with mongosync
 type MigrationVerifierAPI interface {
-	Check(ctx context.Context)
+	Check(ctx context.Context, filter map[string]any)
 	WritesOff(ctx context.Context)
 	WritesOn(ctx context.Context)
 	GetProgress(ctx context.Context) (Progress, error)
@@ -121,10 +121,12 @@ func (server *WebServer) RequestAndResponseLogger() gin.HandlerFunc {
 	}
 }
 
-// Run checks the web server. This is a blocking call.
-// This function should only be called once during each Webserver's life time.
-func (server *WebServer) Run(ctx context.Context) error {
+func (server *WebServer) setupRouter() *gin.Engine {
 	gin.SetMode(gin.ReleaseMode)
+	// This is set so that gin does not decode JSON numbers into float64.
+	// The actual BSON type conversion for JSON numbers is deferred to go-driver.
+	gin.EnableJsonDecoderUseNumber()
+
 	router := gin.New()
 	router.Use(server.RequestAndResponseLogger(), gin.Recovery())
 
@@ -139,10 +141,15 @@ func (server *WebServer) Run(ctx context.Context) error {
 	}
 
 	router.HandleMethodNotAllowed = true
+	return router
+}
 
+// Run checks the web server. This is a blocking call.
+// This function should only be called once during each Webserver's life time.
+func (server *WebServer) Run(ctx context.Context) error {
 	server.srv = &http.Server{
 		Addr:    "0.0.0.0:" + strconv.Itoa(server.port),
-		Handler: router,
+		Handler: server.setupRouter(),
 	}
 
 	webServerCtx, shutDownWebServer := context.WithCancel(ctx)
@@ -177,15 +184,21 @@ func (server *WebServer) Run(ctx context.Context) error {
 // EmptyRequest is for request with empty body
 type EmptyRequest struct{}
 
+// CheckRequest is for requests to the /check endpoint.
+type CheckRequest struct {
+	Filter map[string]any `json:"filter"`
+}
+
 func (server *WebServer) checkEndPoint(c *gin.Context) {
-	var json EmptyRequest
+	var req CheckRequest
 
-	if err := c.ShouldBindJSON(&json); err != nil {
+	if err := c.ShouldBindJSON(&req); err != nil {
+		err = errors.Wrap(err, "filter is not valid JSON")
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
 	}
 
-	server.Mapi.Check(context.Background())
+	server.Mapi.Check(context.Background(), req.Filter)
 	//if err != nil {
 	//	server.operationalErrorResponse(c, err)
 	//	return