rework cluster-time-getting

Author: FGasper

internal/verifier/check.go

@@ -49,7 +49,7 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context) error {
 	if csRunning {
 		verifier.logger.Debug().Msg("Changestream still running, signalling that writes are done and waiting for change stream to exit")
 
-		finalTs, err := GetClusterTime(
+		finalTs, err := GetNewClusterTime(
 			ctx,
 			verifier.logger,
 			verifier.srcClient,

internal/verifier/clustertime.go

@@ -2,11 +2,9 @@ package verifier
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/10gen/migration-verifier/internal/logger"
 	"github.com/10gen/migration-verifier/internal/retry"
-	"github.com/10gen/migration-verifier/internal/util"
 	"github.com/10gen/migration-verifier/mbson"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
@@ -16,11 +14,11 @@ import (
 
 const opTimeKeyInServerResponse = "operationTime"
 
-// GetClusterTime returns the remote cluster’s cluster time.
-// In so doing it creates a new oplog entry across all shards.
-// The “note” will go into the cluster’s oplog, so keep it
-// short but meaningful.
-func GetClusterTime(
+// GetNewClusterTime advances the remote cluster’s cluster time an returns
+// that time. In sharded clusters this advancement happens across all shards,
+// which (usefully!) equalizes the shards’ cluster time and triggers them to
+// output all events before then.
+func GetNewClusterTime(
 	ctx context.Context,
 	logger *logger.Logger,
 	client *mongo.Client,
@@ -29,125 +27,47 @@ func GetClusterTime(
 
 	var optime primitive.Timestamp
 
-	// To get the cluster time, we submit a request to append an oplog note
-	// but set an unreasonably-early maxClusterTime. All the shards will fail
-	// the request but still send their current clusterTime to the mongos,
-	// which will return the most recent of those. Thus we can fetch the
-	// most recent cluster time without altering the oplog.
 	err := retryer.RunForTransientErrorsOnly(
 		ctx,
 		logger,
 		func(_ *retry.Info) error {
 			var err error
 			optime, err = syncClusterTimeAcrossShards(ctx, client)
-			if err != nil {
-				err = fmt.Errorf("%w: %w", retry.CustomTransientErr, err)
-			}
 			return err
 		},
 	)
 
-	if err != nil {
-		return primitive.Timestamp{}, err
-	}
-
-	// OPTIMIZATION FOR SHARDED CLUSTERS: We append another oplog entry to
-	// bring all shards to a cluster time that’s *after* the optime that we’ll
-	// return. That way any events *at*
-	//
-	// Since this is just an optimization, failures here are nonfatal.
-	err = retryer.RunForTransientErrorsOnly(
-		ctx,
-		logger,
-		func(_ *retry.Info) error {
-			var err error
-			optime, err = syncClusterTimeAcrossShards(ctx, client)
-			if err != nil {
-				err = fmt.Errorf("%w: %w", retry.CustomTransientErr, err)
-			}
-			return err
-		},
-	)
-	if err != nil {
-		// This isn't serious enough even to warn on, so leave it at info-level.
-		logger.Info().Err(err).
-			Msg("Failed to append oplog note; change stream may need extra time to finish.")
-	}
-
-	return optime, nil
+	return optime, err
 }
 
-// Use this when we just need the correct cluster time without
-// actually changing any shards’ oplogs.
-func fetchClusterTime(
+func syncClusterTimeAcrossShards(
 	ctx context.Context,
 	client *mongo.Client,
 ) (primitive.Timestamp, error) {
-	cmd, rawResponse, err := runAppendOplogNote(
-		ctx,
-		client,
-		"expect StaleClusterTime error",
-		bson.E{"maxClusterTime", primitive.Timestamp{1, 0}},
-	)
-
-	// We expect an error here; if we didn't get one then something is
-	// amiss on the server.
-	if err == nil {
-		return primitive.Timestamp{}, errors.Errorf("server request unexpectedly succeeded: %v", cmd)
+	cmd := bson.D{
+		{"appendOplogNote", 1},
+		{"data", bson.D{
+			{"migration-verifier", "syncing cluster time"},
+		}},
 	}
 
-	if !util.IsStaleClusterTimeError(err) {
-		return primitive.Timestamp{}, errors.Wrap(
-			err,
-			"unexpected error (expected StaleClusterTime) from request",
-		)
-	}
+	resp := client.Database("admin").RunCommand(ctx, cmd)
 
-	return getOpTimeFromRawResponse(rawResponse)
-}
-
-func syncClusterTimeAcrossShards(
-	ctx context.Context,
-	client *mongo.Client,
-) (primitive.Timestamp, error) {
-	_, rawResponse, err := runAppendOplogNote(ctx, client, "syncing cluster time")
+	rawResponse, err := resp.Raw()
 
 	if err != nil {
-		return primitive.Timestamp{}, err
+		return primitive.Timestamp{}, errors.Wrapf(
+			err,
+			"command (%v) failed unexpectedly",
+			cmd,
+		)
 	}
 
 	return getOpTimeFromRawResponse(rawResponse)
 }
 
-func runAppendOplogNote(
-	ctx context.Context,
-	client *mongo.Client,
-	note string,
-	extraPieces ...bson.E,
-) (bson.D, bson.Raw, error) {
-	cmd := append(
-		bson.D{
-			{"appendOplogNote", 1},
-			{"data", bson.D{
-				{"migration-verifier", note},
-			}},
-		},
-		extraPieces...,
-	)
-
-	resp := client.Database("admin").RunCommand(ctx, cmd)
-
-	raw, err := resp.Raw()
-
-	return cmd, raw, errors.Wrapf(
-		err,
-		"command (%v) failed unexpectedly",
-		cmd,
-	)
-}
-
 func getOpTimeFromRawResponse(rawResponse bson.Raw) (primitive.Timestamp, error) {
-	// Get the `operationTime` from the response and return it.
+	// Return the response’s `operationTime`.
 	var optime primitive.Timestamp
 
 	found, err := mbson.RawLookup(rawResponse, &optime, opTimeKeyInServerResponse)

internal/verifier/clustertime_test.go

@@ -6,7 +6,7 @@ func (suite *IntegrationTestSuite) TestGetClusterTime() {
 	ctx := context.Background()
 	logger, _ := getLoggerAndWriter("stdout")
 
-	ts, err := GetClusterTime(ctx, logger, suite.srcMongoClient)
+	ts, err := GetNewClusterTime(ctx, logger, suite.srcMongoClient)
 	suite.Require().NoError(err)
 
 	suite.Assert().NotZero(ts, "timestamp should be nonzero")