mongo version check

Author: jgao54

flow/connectors/mongo/mongo.go

@@ -21,6 +21,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
+	peerdb_mongo "github.com/PeerDB-io/peerdb/flow/shared/mongo"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
@@ -80,47 +81,11 @@ func (c *MongoConnector) ConnectionActive(ctx context.Context) error {
 }
 
 func (c *MongoConnector) GetVersion(ctx context.Context) (string, error) {
-	db := c.client.Database("admin")
-
-	var buildInfoDoc bson.M
-	if err := db.RunCommand(ctx, bson.D{bson.E{Key: "buildInfo", Value: 1}}).Decode(&buildInfoDoc); err != nil {
-		return "", fmt.Errorf("failed to run buildInfo command: %w", err)
-	}
-
-	version, ok := buildInfoDoc["version"].(string)
-	if !ok {
-		return "", fmt.Errorf("buildInfo.version is not a string, but %T", buildInfoDoc["version"])
-	}
-	return version, nil
-}
-
-func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigs) error {
-	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
-		return nil
-	}
-
-	// Check if MongoDB is configured as a replica set
-	var result bson.M
-	if err := c.client.Database("admin").RunCommand(ctx, bson.D{
-		{Key: "replSetGetStatus", Value: 1},
-	}).Decode(&result); err != nil {
-		return fmt.Errorf("failed to get replica set status: %w", err)
-	}
-
-	// Check if this is a replica set
-	if _, ok := result["set"]; !ok {
-		return errors.New("MongoDB is not configured as a replica set, which is required for CDC")
-	}
-
-	if myState, ok := result["myState"]; !ok {
-		return errors.New("myState not found in response")
-	} else if myStateInt, ok := myState.(int32); !ok {
-		return fmt.Errorf("failed to convert myState %v to int32", myState)
-	} else if myStateInt != 1 {
-		return fmt.Errorf("MongoDB is not the primary node in the replica set, current state: %d", myState)
+	buildInfo, err := peerdb_mongo.GetBuildInfo(ctx, c.client)
+	if err != nil {
+		return "", err
 	}
-
-	return nil
+	return buildInfo.Version, nil
 }
 
 func (c *MongoConnector) GetTableSchema(

flow/connectors/mongo/validate.go

@@ -0,0 +1,34 @@
+package connmongo
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	shared_mongo "github.com/PeerDB-io/peerdb/flow/shared/mongo"
+)
+
+const MinSupportedVersion = "5.1.0"
+
+func (c *MongoConnector) ValidateCheck(ctx context.Context) error {
+	cmp, err := shared_mongo.CompareServerVersion(ctx, c.client, MinSupportedVersion)
+	if err != nil {
+		return err
+	}
+	if cmp == -1 {
+		return fmt.Errorf("require minimum mongo version %s", MinSupportedVersion)
+	}
+	return nil
+}
+
+func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.FlowConnectionConfigs) error {
+	if cfg.DoInitialSnapshot && cfg.InitialSnapshotOnly {
+		return nil
+	}
+
+	_, err := shared_mongo.GetReplSetGetStatus(ctx, c.client)
+	if err != nil {
+		return fmt.Errorf("failed to get replica set status: %w", err)
+	}
+	return nil
+}

flow/e2eshared/e2eshared.go

@@ -33,7 +33,7 @@ func RunSuite[T Suite](t *testing.T, setup func(t *testing.T) T) {
 					subtest.Parallel()
 					suite := setup(subtest)
 					subtest.Cleanup(func() {
-						ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+						ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
 						defer cancel()
 						suite.Teardown(ctx)
 					})
@@ -56,7 +56,7 @@ func RunSuiteNoParallel[T Suite](t *testing.T, setup func(t *testing.T) T) {
 				t.Run(m.Name, func(subtest *testing.T) {
 					suite := setup(subtest)
 					subtest.Cleanup(func() {
-						ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+						ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
 						defer cancel()
 						suite.Teardown(ctx)
 					})

flow/go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/eventhub/armeventhub v1.3.0
 	github.com/ClickHouse/ch-go v0.66.0
 	github.com/ClickHouse/clickhouse-go/v2 v2.37.1
+	github.com/Masterminds/semver v1.5.0
 	github.com/PeerDB-io/glua64 v1.0.1
 	github.com/PeerDB-io/gluabit32 v1.0.2
 	github.com/PeerDB-io/gluajson v1.0.2

flow/go.sum

@@ -82,6 +82,8 @@ github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.53.0/go.mod h1:jUZ5LYlw40WMd07qxcQJD5M40aUxrfwqQX1g7zxYnrQ=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0 h1:Ron4zCA/yk6U7WOBXhTJcDpsUBG9npumK6xw2auFltQ=
 github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.53.0/go.mod h1:cSgYe11MCNYunTnRXrKiR/tHc0eoKjICUuWpNZoVCOo=
+github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
+github.com/Masterminds/semver v1.5.0/go.mod h1:MB6lktGJrhw8PrUyiEoblNEGEQ+RzHPF078ddwwvV3Y=
 github.com/PeerDB-io/glua64 v1.0.1 h1:biXLlFF/L5pnJCwDon7hkWkuQPozC8NjKS3J7Wzi69I=
 github.com/PeerDB-io/glua64 v1.0.1/go.mod h1:UHmAhniv61bJPMhQvxkpC7jXbn353dSbQviu83bgQVg=
 github.com/PeerDB-io/gluabit32 v1.0.2 h1:AGI1Z7dwDVotakpuOOuyTX4/QGi5HUYsipL/VfodmO4=

flow/shared/mongo/validation.go

@@ -0,0 +1,59 @@
+package mongo
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/Masterminds/semver"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+type BuildInfo struct {
+	Version string `bson:"version"`
+}
+
+type ReplSetGetStatus struct {
+	Set     string `bson:"set"`
+	MyState int    `bson:"myState"`
+}
+
+func GetBuildInfo(ctx context.Context, client *mongo.Client) (*BuildInfo, error) {
+	singleResult := client.Database("admin").RunCommand(ctx, bson.D{bson.E{Key: "buildInfo", Value: 1}})
+	if singleResult.Err() != nil {
+		return nil, fmt.Errorf("failed to run 'buildInfo' command: %w", singleResult.Err())
+	}
+	var info BuildInfo
+	if err := singleResult.Decode(&info); err != nil {
+		return nil, fmt.Errorf("failed to decode BuildInfo: %w", err)
+	}
+	return &info, nil
+}
+
+func CompareServerVersion(ctx context.Context, client *mongo.Client, versionString string) (int, error) {
+	buildInfo, err := GetBuildInfo(ctx, client)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get mongo version: %w", err)
+	}
+	parsedVersion, err := semver.NewVersion(buildInfo.Version)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse mongo version %w", err)
+	}
+	compareVersion := semver.MustParse(versionString)
+	return parsedVersion.Compare(compareVersion), nil
+}
+
+func GetReplSetGetStatus(ctx context.Context, client *mongo.Client) (*ReplSetGetStatus, error) {
+	db := client.Database("admin")
+	singleResult := db.RunCommand(ctx, bson.D{
+		bson.E{Key: "replSetGetStatus", Value: 1},
+	})
+	if singleResult.Err() != nil {
+		return nil, fmt.Errorf("failed to run 'replSetGetStatus' command: %w", singleResult.Err())
+	}
+	var status ReplSetGetStatus
+	if err := singleResult.Decode(&status); err != nil {
+		return nil, fmt.Errorf("failed to decode ReplSetGetStatus: %w", err)
+	}
+	return &status, nil
+}