detect sharding mismatches

Author: FGasper

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 *util.BuildInfo, filterAndPredicates bson.A) bson.D {
+func (p *Partition) GetFindOptions(buildInfo *util.ClusterInfo, filterAndPredicates bson.A) bson.D {
 	if p == nil {
 		if len(filterAndPredicates) > 0 {
 			return bson.D{{"filter", bson.D{{"$and", filterAndPredicates}}}}

internal/partitions/partition_test.go

@@ -78,37 +78,37 @@ func (suite *UnitTestSuite) TestVersioning() {
 	suite.Require().Equal(expectedFilter, filter)
 
 	// 6.0
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{6, 0, 0}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{6, 0, 0}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilter, filter)
 
 	// 5.3.0.9
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{5, 3, 0, 9}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{5, 3, 0, 9}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilter, filter)
 
 	// 7.1.3.5
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{7, 1, 3, 5}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{7, 1, 3, 5}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilter, filter)
 
 	// 4.4 (int64)
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{4, 4, 0, 0}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{4, 4, 0, 0}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilterWithTypeBracketing, filter)
 
 	// 4.4
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{4, 4, 0, 0}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{4, 4, 0, 0}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilterWithTypeBracketing, filter)
 
 	// 4.2
-	findOptions = partition.GetFindOptions(&util.BuildInfo{VersionArray: []int{4, 2, 0, 0}}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{VersionArray: []int{4, 2, 0, 0}}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilterWithTypeBracketing, filter)
 
 	// No version array -- assume old, require type bracketing.
-	findOptions = partition.GetFindOptions(&util.BuildInfo{}, nil)
+	findOptions = partition.GetFindOptions(&util.ClusterInfo{}, nil)
 	filter = getFilterFromFindOptions(findOptions)
 	suite.Require().Equal(expectedFilterWithTypeBracketing, filter)
 }

internal/retry/retry.go

@@ -11,7 +11,12 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 )
 
-// RunForUUIDAndTransientErrors retries f() for the CollectionUUIDMismatch error and for transient errors.
+// RunForUUIDAndTransientErrors should never be used. Its presence here
+// is a vestige of the retryer's origin in mongosync. Since migration-verifier
+// forbids DDL operations, a namespace should be just as stable as its UUID; thus,
+// this function is needless complexity.
+//
+// This function retries f() for the CollectionUUIDMismatch error and for transient errors.
 // This should be used to run a driver operation that optionally specifies the `collectionUUID` parameter
 // for a collection that may have been:
 //
@@ -34,15 +39,20 @@ import (
 // f() is provided with a collection name string, which is the one that should be used in the body
 // of f() when a collection name is needed. The initial value of this string is expectedCollName.
 //
-// RunForUUIDAndTransientErrors always returns the collection's current name. It returns
+// This function always returns the collection's current name. It returns
 // an error if the duration limit is reached, or if f() returns a non-transient error.
 func (r *Retryer) RunForUUIDAndTransientErrors(
 	ctx context.Context, logger *logger.Logger, expectedCollName string, f func(*Info, string) error,
 ) (string, error) {
 	return r.runRetryLoop(ctx, logger, expectedCollName, f, true, true)
 }
 
-// RunForUUIDErrorOnly retries f() for the CollectionUUIDMismatch error only. This should primarily
+// RunForUUIDErrorOnly should never be used. Its presence here
+// is a vestige of the retryer's origin in mongosync. Since migration-verifier
+// forbids DDL operations, a namespace should be just as stable as its UUID; thus,
+// this function is needless complexity.
+//
+// This function retries f() for the CollectionUUIDMismatch error only. This should primarily
 // be used to wrap a transaction callback containing an operation that specifies the `collectionUUID`
 // parameter for a collection that may have been:
 //
@@ -57,7 +67,7 @@ func (r *Retryer) RunForUUIDAndTransientErrors(
 // f() is provided with a collection name string, which is the one that should be used in the body
 // of f() where a collection name is needed. The initial value of this string is expectedCollName.
 //
-// RunForUUIDErrorOnly returns the collection's current name in all cases.
+// This function returns the collection's current name in all cases.
 func (r *Retryer) RunForUUIDErrorOnly(
 	ctx context.Context, logger *logger.Logger, expectedCollName string, f func(*Info, string) error,
 ) (string, error) {

internal/util/buildinfo.go

@@ -0,0 +1,78 @@
+package util
+
+import (
+	"context"
+
+	"github.com/10gen/migration-verifier/mbson"
+	"github.com/pkg/errors"
+	"github.com/samber/lo"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type ClusterTopology string
+
+type ClusterInfo struct {
+	VersionArray []int
+	Topology     ClusterTopology
+}
+
+const (
+	TopologySharded ClusterTopology = "sharded"
+	TopologyReplset ClusterTopology = "replset"
+)
+
+func GetClusterInfo(ctx context.Context, client *mongo.Client) (ClusterInfo, error) {
+	va, err := getVersionArray(ctx, client)
+	if err != nil {
+		return ClusterInfo{}, errors.Wrap(err, "failed to fetch version array")
+	}
+
+	topology, err := getTopology(ctx, client)
+	if err != nil {
+		return ClusterInfo{}, errors.Wrap(err, "failed to determine topology")
+	}
+
+	return ClusterInfo{
+		VersionArray: va,
+		Topology:     topology,
+	}, nil
+}
+
+func getVersionArray(ctx context.Context, client *mongo.Client) ([]int, error) {
+	commandResult := client.Database("admin").RunCommand(ctx, bson.D{{"buildinfo", 1}})
+
+	rawResp, err := commandResult.Raw()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to run %#q", "buildinfo")
+	}
+
+	var va []int
+	_, err = mbson.RawLookup(rawResp, &va, "versionArray")
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to decode build info version array")
+	}
+
+	return va, nil
+}
+
+func getTopology(ctx context.Context, client *mongo.Client) (ClusterTopology, error) {
+	resp := client.Database("admin").RunCommand(
+		ctx,
+		bson.D{{"hello", 1}},
+	)
+
+	hello := struct {
+		Msg string
+	}{}
+
+	if err := resp.Decode(&hello); err != nil {
+		return "", errors.Wrapf(
+			err,
+			"failed to decode %#q response",
+			"hello",
+		)
+	}
+
+	return lo.Ternary(hello.Msg == "isdbgrid", TopologySharded, TopologyReplset), nil
+}

internal/util/clusterinfo.go

@@ -0,0 +1,66 @@
+package util
+
+import (
+	"context"
+
+	"github.com/10gen/migration-verifier/option"
+	"github.com/pkg/errors"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
+)
+
+const (
+	configDBName  = "config"
+	collsCollName = "collections"
+)
+
+// GetShardKey returns the collection's shard key, or an empty option
+// if the collection is unsharded.
+func GetShardKey(
+	ctx context.Context,
+	coll *mongo.Collection,
+) (option.Option[bson.Raw], error) {
+	namespace := coll.Database().Name() + "." + coll.Name()
+
+	configCollectionsColl := coll.Database().Client().
+		Database(configDBName).
+		Collection(collsCollName)
+
+	rawResult, err := configCollectionsColl.FindOne(ctx, bson.D{{"_id", namespace}}).Raw()
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return option.None[bson.Raw](), nil
+	} else if err != nil {
+		return option.None[bson.Raw](), errors.Wrapf(
+			err,
+			"failed to find sharding info for %#q",
+			namespace,
+		)
+	}
+
+	keyAsVal, err := rawResult.LookupErr("key")
+	if errors.Is(err, bsoncore.ErrElementNotFound) {
+		return option.None[bson.Raw](), nil
+	} else if err != nil {
+		return option.None[bson.Raw](), errors.Wrapf(
+			err,
+			"failed to find %#q in %#q's %#q entry",
+			"key",
+			namespace,
+			FullName(configCollectionsColl),
+		)
+	}
+
+	keyAsRaw, isDoc := keyAsVal.DocumentOK()
+	if !isDoc {
+		return option.None[bson.Raw](), errors.Errorf(
+			"%#q in %#q's %#q entry is of type %#q, not an object",
+			"key",
+			namespace,
+			FullName(configCollectionsColl),
+			keyAsVal.Type,
+		)
+	}
+
+	return option.Some(keyAsRaw), nil
+}

internal/util/sharding.go

@@ -315,7 +315,7 @@ func (verifier *Verifier) createChangeStream(
 		SetMaxAwaitTime(1 * time.Second).
 		SetFullDocument(options.UpdateLookup)
 
-	if verifier.srcBuildInfo.VersionArray[0] >= 6 {
+	if verifier.srcClusterInfo.VersionArray[0] >= 6 {
 		opts = opts.SetCustomPipeline(bson.M{"showExpandedEvents": true})
 	}
 

internal/verifier/change_stream.go

@@ -416,7 +416,7 @@ func (suite *IntegrationTestSuite) testInsertsBeforeWritesOff(docsCount int) {
 
 func (suite *IntegrationTestSuite) TestCreateForbidden() {
 	ctx := suite.Context()
-	buildInfo, err := util.GetBuildInfo(ctx, suite.srcMongoClient)
+	buildInfo, err := util.GetClusterInfo(ctx, suite.srcMongoClient)
 	suite.Require().NoError(err)
 
 	if buildInfo.VersionArray[0] < 6 {

internal/verifier/change_stream_test.go

@@ -266,7 +266,7 @@ func (verifier *Verifier) getFetcherChannels(
 		cursor, err := verifier.getDocumentsCursor(
 			ctx,
 			verifier.srcClientCollection(task),
-			verifier.srcBuildInfo,
+			verifier.srcClusterInfo,
 			verifier.srcStartAtTs,
 			task,
 		)
@@ -290,7 +290,7 @@ func (verifier *Verifier) getFetcherChannels(
 		cursor, err := verifier.getDocumentsCursor(
 			ctx,
 			verifier.dstClientCollection(task),
-			verifier.dstBuildInfo,
+			verifier.dstClusterInfo,
 			nil, //startAtTs
 			task,
 		)