REP-6829 Add 4.0 support (#176)

This adds support for MongoDB 4.0 sources to Migration Verifier.

- The support is limited to replica sets because 4.0 didn’t support change streams from a mongos.
- The only tested destination version is 8.0.
- This _downgrades_ the Go driver to v2.3.1 because that’s the last version that supported 4.0.
- CI also now uses an outdated PyMongo for the same reason.

Also added is CI testing of 4.2->8.0.

Author: FGasper

.github/workflows/all.yml

@@ -7,20 +7,26 @@ on:
   pull_request:
   workflow_dispatch:
 
+env:
+    replsetSrcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
+    replsetDstConnStr: mongodb://localhost:27030,localhost:27031,localhost:27032
+    shardedSrcConnStr: mongodb://localhost:27020
+    shardedDstConnStr: mongodb://localhost:27030
+
 jobs:
   basics:
     strategy:
       fail-fast: false
       matrix:
         include:
+          # NB: This is the only setup we’re supporting with 4.0.
+          - mongodb_versions: [ '4.0', '8.0' ]
+            topology: replset
 
           # Testing fallback when `hello` isn’t implemented
           # (but appendOplogNote is).
           - mongodb_versions: [ '4.2.5', '6.0' ]
-            topology:
-              name: replset
-              srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
-              dstConnStr: mongodb://localhost:27030,localhost:27031,localhost:27032
+            topology: replset
 
         exclude:
           - mongodb_versions: [ '4.2', '4.2' ]
@@ -31,13 +37,16 @@ jobs:
             toHashedIndexKey: true
           - mongodb_versions: [ '4.2', '6.0' ]
             toHashedIndexKey: true
+          - mongodb_versions: [ '4.2', '8.0' ]
+            toHashedIndexKey: true
 
         # versions are: source, destination
         mongodb_versions:
           - [ '4.2', '4.2' ]
           - [ '4.2', '4.4' ]
           - [ '4.2', '5.0' ]
           - [ '4.2', '6.0' ]
+          - [ '4.2', '8.0' ]
 
           - [ '4.4', '4.4' ]
           - [ '4.4', '5.0' ]
@@ -61,26 +70,15 @@ jobs:
         toHashedIndexKey: [true, false]
 
         topology:
-          - name: replset
-            srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
-            dstConnStr: mongodb://localhost:27030,localhost:27031,localhost:27032
-
-          - name: replset-to-sharded
-            dstArgs: --sharded 2
-            srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
-            dstConnStr: mongodb://localhost:27030
-
-          - name: sharded
-            srcArgs: --sharded 2
-            dstArgs: --sharded 2
-            srcConnStr: mongodb://localhost:27020
-            dstConnStr: mongodb://localhost:27030
+          - replset
+          - replset-to-sharded
+          - sharded
 
     # Ubuntu 24 lacks OpenSSL 1.1.1’s libcrypto, which pre-v6 MongoDB
     # versions need.
     runs-on: ubuntu-22.04
 
-    name: ${{ matrix.mongodb_versions[0] }} to ${{ matrix.mongodb_versions[1] }}, ${{ matrix.topology.name }}${{ matrix.toHashedIndexKey && ', hashed doc compare' || '' }}
+    name: ${{ matrix.mongodb_versions[0] }} to ${{ matrix.mongodb_versions[1] }}, ${{ matrix.topology }}${{ matrix.toHashedIndexKey && ', hashed doc compare' || '' }}
 
     steps:
       - run: uname -a
@@ -95,11 +93,12 @@ jobs:
         with:
           go-version: stable
 
+      # Use an old pymongo so we can install MongoDB 4.0.
       - name: Install m and mtools
         run: |-
             {
                 echo npm install -g m
-                echo pipx install 'mtools[all]'
+                echo pip install pymongo==4.10.1 'mtools[all]'
             } | parallel
 
       - name: Install MongoDB ${{ matrix.mongodb_versions[0] }} (source)
@@ -115,15 +114,17 @@ jobs:
         run: |-
             {
                 echo ./build.sh
-                echo mlaunch init --binarypath $(cat .srcpath) --port 27020 --dir src --replicaset ${{ matrix.topology.srcArgs }}
-                echo mlaunch init --binarypath $(cat .dstpath) --port 27030 --dir dst --replicaset ${{ matrix.topology.dstArgs }}
+                echo mlaunch init --binarypath $(cat .srcpath) --port 27020 --dir src --replicaset ${{ (matrix.topology == 'sharded') && '--sharded 2' || '' }}
+                echo mlaunch init --binarypath $(cat .dstpath) --port 27030 --dir dst --replicaset ${{ (matrix.topology == 'sharded' || matrix.topology == 'replset-to-sharded') && '--sharded 2' || '' }}
                 echo mlaunch init --binarypath $(cat .metapath) --port 27040 --dir meta --replicaset --nodes 1
             } | parallel
 
       - name: Test
         run: go test -v ./... -race
         env:
           MVTEST_DOC_COMPARE_METHOD: ${{matrix.toHashedIndexKey && 'toHashedIndexKey' || ''}}
-          MVTEST_SRC: ${{matrix.topology.srcConnStr}}
-          MVTEST_DST: ${{matrix.topology.dstConnStr}}
+
+          MVTEST_SRC: ${{ (matrix.topology == 'sharded') && env.shardedSrcConnStr || env.replsetSrcConnStr }}
+          MVTEST_DST: ${{ (matrix.topology == 'sharded' || matrix.topology == 'replset-to-sharded') && env.shardedDstConnStr || env.replsetDstConnStr }}
+
           MVTEST_META: mongodb://localhost:27040

go.mod

@@ -8,8 +8,10 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/gin-contrib/pprof v1.5.3
 	github.com/gin-gonic/gin v1.10.0
+	github.com/goaux/timer v1.1.0
 	github.com/google/uuid v1.3.0
 	github.com/huandu/go-clone/generic v1.7.2
+	github.com/libp2p/go-buffer-pool v0.1.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pkg/errors v0.9.1
 	github.com/quasilyte/go-ruleguard/dsl v0.3.22
@@ -18,7 +20,9 @@ require (
 	github.com/samber/mo v1.13.0
 	github.com/stretchr/testify v1.10.0
 	github.com/urfave/cli v1.22.9
-	go.mongodb.org/mongo-driver/v2 v2.4.0
+
+	// NB: This is the last driver version that supported 4.0.
+	go.mongodb.org/mongo-driver/v2 v2.3.1
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56
 	golang.org/x/sync v0.13.0
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
@@ -36,15 +40,13 @@ require (
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.26.0 // indirect
-	github.com/goaux/timer v1.1.0 // indirect
 	github.com/goccy/go-json v0.10.5 // indirect
 	github.com/golang/snappy v1.0.0 // indirect
 	github.com/huandu/go-clone v1.6.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.16.7 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
-	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-runewidth v0.0.9 // indirect

go.sum

@@ -129,8 +129,8 @@ github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gi
 github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM=
 github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
-go.mongodb.org/mongo-driver/v2 v2.4.0 h1:Oq6BmUAAFTzMeh6AonuDlgZMuAuEiUxoAD1koK5MuFo=
-go.mongodb.org/mongo-driver/v2 v2.4.0/go.mod h1:jHeEDJHJq7tm6ZF45Issun9dbogjfnPySb1vXA7EeAI=
+go.mongodb.org/mongo-driver/v2 v2.3.1 h1:WrCgSzO7dh1/FrePud9dK5fKNZOE97q5EQimGkos7Wo=
+go.mongodb.org/mongo-driver/v2 v2.3.1/go.mod h1:jHeEDJHJq7tm6ZF45Issun9dbogjfnPySb1vXA7EeAI=
 golang.org/x/arch v0.16.0 h1:foMtLTdyOmIniqWCHjY6+JxuC54XP1fDwx4N0ASyW+U=
 golang.org/x/arch v0.16.0/go.mod h1:JmwW7aLIoRUKgaTzhkiEFxvcEiQGyOg9BMonBJUS7EE=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

internal/testutil/testutil.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"testing"
 
+	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/pkg/errors"
 	"github.com/samber/lo"
 	"go.mongodb.org/mongo-driver/v2/bson"
@@ -93,8 +94,18 @@ func KillApplicationChangeStreams(
 	ctx context.Context,
 	t *testing.T,
 	client *mongo.Client,
+	versionArray []int,
 	appName string,
 ) error {
+	currentOpArg := bson.D{}
+
+	if util.ClusterHasCurrentOpIdleCursors([2]int(versionArray)) {
+		currentOpArg = append(
+			currentOpArg,
+			bson.E{"idleCursors", true},
+		)
+	}
+
 	// Kill verifier’s change stream.
 	cursor, err := client.Database(
 		"admin",
@@ -103,9 +114,7 @@ func KillApplicationChangeStreams(
 		ctx,
 		mongo.Pipeline{
 			{
-				{"$currentOp", bson.D{
-					{"idleCursors", true},
-				}},
+				{"$currentOp", currentOpArg},
 			},
 			{
 				{"$match", bson.D{

internal/util/clusterinfo.go

@@ -31,6 +31,18 @@ func ClusterHasBSONSize(va [2]int) bool {
 	return major > 4
 }
 
+func ClusterHasCurrentOpIdleCursors(va [2]int) bool {
+	major := va[0]
+
+	if major == 4 {
+		return va[1] >= 2
+	}
+
+	return major > 4
+}
+
+var ClusterHasChangeStreamStartAfter = ClusterHasCurrentOpIdleCursors
+
 const (
 	TopologySharded ClusterTopology = "sharded"
 	TopologyReplset ClusterTopology = "replset"

internal/verifier/change_stream.go

@@ -374,7 +374,11 @@ func (csr *ChangeStreamReader) createChangeStream(
 
 		logEvent.Msg("Starting change stream from persisted resume token.")
 
-		opts = opts.SetStartAfter(token)
+		if util.ClusterHasChangeStreamStartAfter([2]int(csr.clusterInfo.VersionArray)) {
+			opts = opts.SetStartAfter(token)
+		} else {
+			opts = opts.SetResumeAfter(token)
+		}
 	} else {
 		csStartLogEvent.Msgf("Starting change stream from current %s cluster time.", csr.readerType)
 	}
@@ -386,7 +390,7 @@ func (csr *ChangeStreamReader) createChangeStream(
 	sctx := mongo.NewSessionContext(ctx, sess)
 	changeStream, err := csr.watcherClient.Watch(sctx, pipeline, opts)
 	if err != nil {
-		return nil, nil, bson.Timestamp{}, errors.Wrap(err, "failed to open change stream")
+		return nil, nil, bson.Timestamp{}, errors.Wrap(err, "opening change stream")
 	}
 
 	err = csr.persistResumeToken(ctx, changeStream.ResumeToken())

internal/verifier/change_stream_test.go

@@ -803,6 +803,7 @@ func (suite *IntegrationTestSuite) TestWritesOffCursorKilledResilience() {
 				suite.Context(),
 				suite.T(),
 				suite.srcMongoClient,
+				verifier.srcClusterInfo.VersionArray,
 				clientAppName,
 			),
 		)
@@ -831,6 +832,7 @@ func (suite *IntegrationTestSuite) TestCursorKilledResilience() {
 			suite.Context(),
 			suite.T(),
 			suite.srcMongoClient,
+			verifier.srcClusterInfo.VersionArray,
 			clientAppName,
 		),
 	)

internal/verifier/dockey_agg_test.go

@@ -66,7 +66,7 @@ func (suite *IntegrationTestSuite) testExtractTrueDocKeyAgg(reverseYN bool) {
 	cursor, err := coll.Aggregate(
 		ctx,
 		mongo.Pipeline{
-			{{"$replaceWith", computedDocKeyAgg}},
+			{{"$replaceRoot", bson.D{{"newRoot", computedDocKeyAgg}}}},
 		},
 	)
 	require.NoError(err, "should open cursor to agg")

internal/verifier/uri.go

@@ -79,14 +79,7 @@ func (verifier *Verifier) maybeSuggestHashedComparisonOptimization() {
 }
 
 func isVersionSupported(version []int) bool {
-	if version[0] >= 5 {
-		return true
-	}
-	if version[0] < 4 {
-		return false
-	}
-
-	return version[1] >= 2
+	return version[0] >= 4
 }
 
 func (verifier *Verifier) SetDstURI(ctx context.Context, uri string) error {