mongo: add validation for (1) storage engine (2) oplog retention (#3168)

>You can specify the minimum number of hours to preserve an oplog entry
where
[mongod](https://www.mongodb.com/docs/manual/reference/program/mongod/#mongodb-binary-bin.mongod)
only removes an oplog entry if both of the following criteria are met:
-The oplog has reached the [maximum configured
size.](https://www.mongodb.com/docs/manual/core/replica-set-oplog/#std-label-replica-set-oplog-sizing)
-The oplog entry is older than the configured number of hours based on
the host system clock.
By default MongoDB does not set a minimum oplog retention period and
automatically truncates the oplog starting with the oldest entries to
maintain the configured maximum oplog size.

Seems like a good idea to require a min oplog retention (instead of just
relying on max oplog size).

Also add a validation for storage engne, as 'wiredTiger' is a
prerequisite for changestream to work.

Test: tested e2e locally

Author: jgao54

flow/connectors/mongo/validate.go

@@ -2,6 +2,7 @@ package connmongo
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
@@ -30,7 +31,20 @@ func (c *MongoConnector) ValidateMirrorSource(ctx context.Context, cfg *protos.F
 
 	_, err := shared_mongo.GetReplSetGetStatus(ctx, c.client)
 	if err != nil {
-		return fmt.Errorf("failed to get replica set status: %w", err)
+		return err
+	}
+
+	serverStatus, err := shared_mongo.GetServerStatus(ctx, c.client)
+	if err != nil {
+		return err
+	}
+	if serverStatus.StorageEngine.Name != "wiredTiger" {
+		return errors.New("storage engine must be 'wiredTiger'")
 	}
+	if serverStatus.OplogTruncation.OplogMinRetentionHours == 0 ||
+		serverStatus.OplogTruncation.OplogMinRetentionHours < shared_mongo.MinOplogRetentionHours {
+		return errors.New("oplog retention must be set to >= 24 hours")
+	}
+
 	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/shared/mongo/validation.go

@@ -8,7 +8,10 @@ import (
 	"go.mongodb.org/mongo-driver/v2/mongo"
 )
 
-const MinSupportedVersion = "5.1.0"
+const (
+	MinSupportedVersion    = "5.1.0"
+	MinOplogRetentionHours = 24
+)
 
 type BuildInfo struct {
 	Version string `bson:"version"`
@@ -19,6 +22,19 @@ type ReplSetGetStatus struct {
 	MyState int    `bson:"myState"`
 }
 
+type OplogTruncation struct {
+	OplogMinRetentionHours float64 `bson:"oplogMinRetentionHours"`
+}
+
+type StorageEngine struct {
+	Name string `bson:"name"`
+}
+
+type ServerStatus struct {
+	StorageEngine   StorageEngine   `bson:"storageEngine"`
+	OplogTruncation OplogTruncation `bson:"oplogTruncation"`
+}
+
 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 {
@@ -44,3 +60,17 @@ func GetReplSetGetStatus(ctx context.Context, client *mongo.Client) (*ReplSetGet
 	}
 	return &status, nil
 }
+
+func GetServerStatus(ctx context.Context, client *mongo.Client) (*ServerStatus, error) {
+	singleResult := client.Database("admin").RunCommand(ctx, bson.D{
+		bson.E{Key: "serverStatus", Value: 1},
+	})
+	if singleResult.Err() != nil {
+		return nil, fmt.Errorf("failed to run 'serverStatus' command: %w", singleResult.Err())
+	}
+	var status ServerStatus
+	if err := singleResult.Decode(&status); err != nil {
+		return nil, fmt.Errorf("failed to decode ServerStatus: %w", err)
+	}
+	return &status, nil
+}