Skip to content

Commit 6beb9a5

Browse files
FGasperFGasper
committed
Merge branch 'main' into REP-5329-retry-consistently
2 parents 979fb74 + 8992698 commit 6beb9a5

File tree

13 files changed

+592
-116
lines changed

13 files changed

+592
-116
lines changed

‎.github/workflows/all.yml‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,14 @@ jobs:
4141
srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
4242
dstConnStr: mongodb://localhost:27030,localhost:27031,localhost:27032
4343

44+
- name: replset-to-sharded
45+
dstArgs: --sharded 2
46+
srcConnStr: mongodb://localhost:27020,localhost:27021,localhost:27022
47+
dstConnStr: mongodb://localhost:27030
48+
4449
- name: sharded
45-
args: --sharded 2
50+
srcArgs: --sharded 2
51+
dstArgs: --sharded 2
4652
srcConnStr: mongodb://localhost:27020
4753
dstConnStr: mongodb://localhost:27030
4854

@@ -82,8 +88,8 @@ jobs:
8288
- name: Start clusters
8389
run: |-
8490
{
85-
echo mlaunch init --binarypath $(cat .srcpath) --port 27020 --dir src --replicaset ${{ matrix.topology.args }}
86-
echo mlaunch init --binarypath $(cat .dstpath) --port 27030 --dir dst --replicaset ${{ matrix.topology.args }}
91+
echo mlaunch init --binarypath $(cat .srcpath) --port 27020 --dir src --replicaset ${{ matrix.topology.srcArgs }}
92+
echo mlaunch init --binarypath $(cat .dstpath) --port 27030 --dir dst --replicaset ${{ matrix.topology.dstArgs }}
8793
echo mlaunch init --binarypath $(cat .metapath) --port 27040 --dir meta --replicaset --nodes 1
8894
} | parallel
8995

‎internal/util/askserver.go‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
package util
2+
3+
import (
4+
"context"
5+
"slices"
6+
7+
"github.com/10gen/migration-verifier/mslices"
8+
"github.com/10gen/migration-verifier/option"
9+
"github.com/pkg/errors"
10+
"go.mongodb.org/mongo-driver/bson"
11+
"go.mongodb.org/mongo-driver/mongo"
12+
)
13+
14+
// ServerThinksTheseMatch runs an aggregation on the server that determines
15+
// whether the server thinks a & b are equal. This allows you, e.g., to
16+
// ignore BSON type differences for equivalent numbers.
17+
//
18+
// tinker is an optional pipeline that operates on the documents in the
19+
// pipeline (`a` and `b`, respectively) before they're compared.
20+
//
21+
// In migration-verifier the client is generally expected to be for
22+
// the metadata cluster.
23+
func ServerThinksTheseMatch(
24+
ctx context.Context,
25+
client *mongo.Client,
26+
a, b any,
27+
tinker option.Option[mongo.Pipeline],
28+
) (bool, error) {
29+
pipeline := mongo.Pipeline{
30+
{{"$documents", []bson.D{
31+
{
32+
{"a", bson.D{{"$literal", a}}},
33+
{"b", bson.D{{"$literal", b}}},
34+
},
35+
}}},
36+
37+
// Now check to be sure that those specs match.
38+
{{"$match", bson.D{
39+
{"$expr", bson.D{
40+
{"$eq", mslices.Of("$a", "$b")},
41+
}},
42+
}}},
43+
}
44+
45+
if extra, hasExtra := tinker.Get(); hasExtra {
46+
pipeline = slices.Insert(
47+
pipeline,
48+
1,
49+
extra...,
50+
)
51+
}
52+
53+
cursor, err := client.Database("admin").Aggregate(ctx, pipeline)
54+
55+
if err == nil {
56+
defer cursor.Close(ctx)
57+
58+
if cursor.Next(ctx) {
59+
return true, nil
60+
}
61+
62+
err = cursor.Err()
63+
}
64+
65+
return false, errors.Wrapf(err, "failed to ask server if a (%v) matches b (%v)", a, b)
66+
}

‎internal/util/sharding.go‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
package util
2+
3+
import (
4+
"context"
5+
6+
"github.com/10gen/migration-verifier/option"
7+
"github.com/pkg/errors"
8+
"go.mongodb.org/mongo-driver/bson"
9+
"go.mongodb.org/mongo-driver/mongo"
10+
)
11+
12+
const (
13+
configDBName = "config"
14+
collsCollName = "collections"
15+
)
16+
17+
// GetShardKey returns the collection's shard key, or an empty option
18+
// if the collection is unsharded.
19+
func GetShardKey(
20+
ctx context.Context,
21+
coll *mongo.Collection,
22+
) (option.Option[bson.Raw], error) {
23+
namespace := coll.Database().Name() + "." + coll.Name()
24+
25+
configCollectionsColl := coll.Database().Client().
26+
Database(configDBName).
27+
Collection(collsCollName)
28+
29+
decoded := struct {
30+
Key option.Option[bson.Raw]
31+
}{}
32+
33+
err := configCollectionsColl.
34+
FindOne(ctx, bson.D{{"_id", namespace}}).
35+
Decode(&decoded)
36+
37+
if errors.Is(err, mongo.ErrNoDocuments) {
38+
return option.None[bson.Raw](), nil
39+
} else if err != nil {
40+
return option.None[bson.Raw](), errors.Wrapf(
41+
err,
42+
"failed to find sharding info for %#q",
43+
namespace,
44+
)
45+
}
46+
47+
key, hasKey := decoded.Key.Get()
48+
49+
if !hasKey {
50+
return option.None[bson.Raw](), nil
51+
}
52+
53+
return option.Some(key), nil
54+
}

‎internal/verifier/check.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ const (
2222
GenerationComplete GenerationStatus = "complete"
2323
)
2424

25-
var failedStatus = mapset.NewSet(
25+
var failedStatuses = mapset.NewSet(
2626
verificationTaskFailed,
2727
verificationTaskMetadataMismatch,
2828
)
@@ -443,7 +443,7 @@ func FetchFailedAndIncompleteTasks(
443443
return err
444444
}
445445
for _, t := range allTasks {
446-
if failedStatus.Contains(t.Status) {
446+
if failedStatuses.Contains(t.Status) {
447447
FailedTasks = append(FailedTasks, t)
448448
} else if t.Status != verificationTaskCompleted {
449449
IncompleteTasks = append(IncompleteTasks, t)

‎internal/verifier/integration_test_suite.go‎

Lines changed: 7 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@ import (
55
"strings"
66
"time"
77

8+
"github.com/10gen/migration-verifier/internal/util"
89
mapset "github.com/deckarep/golang-set/v2"
910
"github.com/pkg/errors"
10-
"github.com/samber/lo"
1111
"github.com/stretchr/testify/suite"
1212
"go.mongodb.org/mongo-driver/bson"
1313
"go.mongodb.org/mongo-driver/mongo"
@@ -16,13 +16,8 @@ import (
1616
"go.mongodb.org/mongo-driver/mongo/writeconcern"
1717
)
1818

19-
type TestTopology string
20-
2119
const (
22-
metaDBName = "VERIFIER_TEST_META"
23-
topologyEnvVar = "MVTEST_TOPOLOGY"
24-
TopologyReplset TestTopology = "replset"
25-
TopologySharded TestTopology = "sharded"
20+
metaDBName = "VERIFIER_TEST_META"
2621
)
2722

2823
type IntegrationTestSuite struct {
@@ -139,22 +134,14 @@ func (suite *IntegrationTestSuite) TearDownTest() {
139134
}
140135
}
141136

142-
func (suite *IntegrationTestSuite) GetSrcTopology() TestTopology {
143-
hello := struct {
144-
Msg string
145-
}{}
146-
147-
resp := suite.srcMongoClient.Database("admin").RunCommand(
137+
func (suite *IntegrationTestSuite) GetTopology(client *mongo.Client) util.ClusterTopology {
138+
clusterInfo, err := util.GetClusterInfo(
148139
suite.Context(),
149-
bson.D{{"hello", 1}},
150-
)
151-
152-
suite.Require().NoError(
153-
resp.Decode(&hello),
154-
"should fetch & decode hello",
140+
client,
155141
)
142+
suite.Require().NoError(err, "should fetch src cluster info")
156143

157-
return lo.Ternary(hello.Msg == "isdbgrid", TopologySharded, "")
144+
return clusterInfo.Topology
158145
}
159146

160147
func (suite *IntegrationTestSuite) BuildVerifier() *Verifier {

‎internal/verifier/metadata.go‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package verifier
2+
3+
import (
4+
"bytes"
5+
"context"
6+
"fmt"
7+
8+
"github.com/10gen/migration-verifier/internal/util"
9+
"github.com/10gen/migration-verifier/option"
10+
"github.com/pkg/errors"
11+
"github.com/samber/lo"
12+
"go.mongodb.org/mongo-driver/mongo"
13+
)
14+
15+
// This is the Field for a VerificationResult for shard key mismatches.
16+
const ShardKeyField = "Shard Key"
17+
18+
func (verifier *Verifier) verifyShardingIfNeeded(
19+
ctx context.Context,
20+
srcColl, dstColl *mongo.Collection,
21+
) ([]VerificationResult, error) {
22+
23+
// We only need to compare if both clusters are sharded
24+
srcSharded := verifier.srcClusterInfo.Topology == util.TopologySharded
25+
dstSharded := verifier.dstClusterInfo.Topology == util.TopologySharded
26+
27+
if !srcSharded || !dstSharded {
28+
return nil, nil
29+
}
30+
31+
srcShardOpt, err := util.GetShardKey(ctx, srcColl)
32+
if err != nil {
33+
return nil, errors.Wrapf(
34+
err,
35+
"failed to fetch %#q's shard key on source",
36+
FullName(srcColl),
37+
)
38+
}
39+
40+
dstShardOpt, err := util.GetShardKey(ctx, dstColl)
41+
if err != nil {
42+
return nil, errors.Wrapf(
43+
err,
44+
"failed to fetch %#q's shard key on destination",
45+
FullName(dstColl),
46+
)
47+
}
48+
49+
srcKey, srcIsSharded := srcShardOpt.Get()
50+
dstKey, dstIsSharded := dstShardOpt.Get()
51+
52+
if !srcIsSharded && !dstIsSharded {
53+
return nil, nil
54+
}
55+
56+
if srcIsSharded != dstIsSharded {
57+
return []VerificationResult{{
58+
Field: ShardKeyField,
59+
Cluster: lo.Ternary(srcIsSharded, ClusterTarget, ClusterSource),
60+
Details: Missing,
61+
NameSpace: FullName(srcColl),
62+
}}, nil
63+
}
64+
65+
if bytes.Equal(srcKey, dstKey) {
66+
return nil, nil
67+
}
68+
69+
areEqual, err := util.ServerThinksTheseMatch(
70+
ctx,
71+
verifier.metaClient,
72+
srcKey, dstKey,
73+
option.None[mongo.Pipeline](),
74+
)
75+
if err != nil {
76+
return nil, errors.Wrapf(
77+
err,
78+
"failed to ask server if shard keys (src %v; dst: %v) match",
79+
srcKey,
80+
dstKey,
81+
)
82+
}
83+
84+
if !areEqual {
85+
return []VerificationResult{{
86+
Field: ShardKeyField,
87+
Details: fmt.Sprintf("%s: src=%v; dst=%v", Mismatch, srcKey, dstKey),
88+
NameSpace: FullName(srcColl),
89+
}}, nil
90+
}
91+
92+
return nil, nil
93+
}

0 commit comments

Comments
 (0)