remove lag computation

Author: FGasper

internal/verifier/change_stream.go

@@ -8,7 +8,6 @@ import (
 	"github.com/10gen/migration-verifier/internal/keystring"
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/util"
-	"github.com/10gen/migration-verifier/option"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/samber/mo"
@@ -139,15 +138,15 @@ func (verifier *Verifier) GetChangeStreamFilter() []bson.D {
 // is unideal but shouldn’t impede correctness since post-writesOff events
 // shouldn’t really happen anyway by definition.
 func (verifier *Verifier) readAndHandleOneChangeEventBatch(
-	sctx mongo.SessionContext,
+	ctx context.Context,
 	ri *retry.Info,
 	cs *mongo.ChangeStream,
 ) error {
 	eventsRead := 0
 	var changeEventBatch []ParsedEvent
 
 	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
-		gotEvent := cs.TryNext(sctx)
+		gotEvent := cs.TryNext(ctx)
 
 		if cs.Err() != nil {
 			return errors.Wrap(cs.Err(), "change stream iteration failed")
@@ -174,15 +173,7 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 		return nil
 	}
 
-	// Update the change stream lag, which we’ll display in the logs.
-	lag, err := computeChangeStreamLag(sctx, cs)
-	if err == nil {
-		verifier.changeStreamLag.Store(option.Some(lag))
-	} else {
-		verifier.logger.Debug().Err(err).Msg("Failed to compute change stream lag.")
-	}
-
-	err = verifier.HandleChangeStreamEvents(sctx, changeEventBatch)
+	err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
 	if err != nil {
 		return errors.Wrap(err, "failed to handle change events")
 	}
@@ -191,7 +182,7 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 }
 
 func (verifier *Verifier) iterateChangeStream(
-	sctx mongo.SessionContext,
+	ctx context.Context,
 	ri *retry.Info,
 	cs *mongo.ChangeStream,
 ) error {
@@ -202,7 +193,7 @@ func (verifier *Verifier) iterateChangeStream(
 			return nil
 		}
 
-		err := verifier.persistChangeStreamResumeToken(sctx, cs)
+		err := verifier.persistChangeStreamResumeToken(ctx, cs)
 		if err == nil {
 			lastPersistedTime = time.Now()
 		}
@@ -217,8 +208,8 @@ func (verifier *Verifier) iterateChangeStream(
 		select {
 
 		// If the context is canceled, return immmediately.
-		case <-sctx.Done():
-			return sctx.Err()
+		case <-ctx.Done():
+			return ctx.Err()
 
 		// If the changeStreamEnderChan has a message, the user has indicated that
 		// source writes are ended. This means we should exit rather than continue
@@ -250,15 +241,15 @@ func (verifier *Verifier) iterateChangeStream(
 					break
 				}
 
-				err = verifier.readAndHandleOneChangeEventBatch(sctx, ri, cs)
+				err = verifier.readAndHandleOneChangeEventBatch(ctx, ri, cs)
 
 				if err != nil {
 					return err
 				}
 			}
 
 		default:
-			err = verifier.readAndHandleOneChangeEventBatch(sctx, ri, cs)
+			err = verifier.readAndHandleOneChangeEventBatch(ctx, ri, cs)
 
 			if err == nil {
 				err = persistResumeTokenIfNeeded()
@@ -296,7 +287,7 @@ func (verifier *Verifier) iterateChangeStream(
 }
 
 func (verifier *Verifier) createChangeStream(
-	sctx mongo.SessionContext,
+	ctx context.Context,
 ) (*mongo.ChangeStream, primitive.Timestamp, error) {
 	pipeline := verifier.GetChangeStreamFilter()
 	opts := options.ChangeStream().
@@ -307,7 +298,7 @@ func (verifier *Verifier) createChangeStream(
 		opts = opts.SetCustomPipeline(bson.M{"showExpandedEvents": true})
 	}
 
-	savedResumeToken, err := verifier.loadChangeStreamResumeToken(sctx)
+	savedResumeToken, err := verifier.loadChangeStreamResumeToken(ctx)
 	if err != nil {
 		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to load persisted change stream resume token")
 	}
@@ -334,12 +325,17 @@ func (verifier *Verifier) createChangeStream(
 		csStartLogEvent.Msg("Starting change stream from current source cluster time.")
 	}
 
+	sess, err := verifier.srcClient.StartSession()
+	if err != nil {
+		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to start session")
+	}
+	sctx := mongo.NewSessionContext(ctx, sess)
 	srcChangeStream, err := verifier.srcClient.Watch(sctx, pipeline, opts)
 	if err != nil {
 		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to open change stream")
 	}
 
-	err = verifier.persistChangeStreamResumeToken(sctx, srcChangeStream)
+	err = verifier.persistChangeStreamResumeToken(ctx, srcChangeStream)
 	if err != nil {
 		return nil, primitive.Timestamp{}, err
 	}
@@ -352,7 +348,7 @@ func (verifier *Verifier) createChangeStream(
 	// With sharded clusters the resume token might lead the cluster time
 	// by 1 increment. In that case we need the actual cluster time;
 	// otherwise we will get errors.
-	clusterTime, err := getClusterTimeFromSession(sctx)
+	clusterTime, err := getClusterTimeFromSession(sess)
 	if err != nil {
 		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to read cluster time from session")
 	}
@@ -382,14 +378,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 			ctx,
 			verifier.logger,
 			func(ri *retry.Info) error {
-				sess, err := verifier.srcClient.StartSession()
-				if err != nil {
-					return errors.Wrap(err, "failed to start change stream session")
-				}
-
-				sctx := mongo.NewSessionContext(ctx, sess)
-
-				srcChangeStream, startTs, err := verifier.createChangeStream(sctx)
+				srcChangeStream, startTs, err := verifier.createChangeStream(ctx)
 				if err != nil {
 					if parentThreadWaiting {
 						initialCreateResultChan <- mo.Err[primitive.Timestamp](err)
@@ -407,7 +396,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 					parentThreadWaiting = false
 				}
 
-				return verifier.iterateChangeStream(sctx, ri, srcChangeStream)
+				return verifier.iterateChangeStream(ctx, ri, srcChangeStream)
 			},
 		)
 
@@ -460,26 +449,12 @@ func (verifier *Verifier) loadChangeStreamResumeToken(ctx context.Context) (bson
 	return token, err
 }
 
-func computeChangeStreamLag(
-	sctx mongo.SessionContext,
-	cs *mongo.ChangeStream,
-) (time.Duration, error) {
-	token := cs.ResumeToken()
-	ts, err := extractTimestampFromResumeToken(token)
-	if err != nil {
-		return 0, errors.Wrap(err, "failed to extract timestamp from change stream resume token")
-	}
-
-	optime := sctx.OperationTime()
-	return time.Second * time.Duration(optime.T-ts.T), nil
-}
-
-func (verifier *Verifier) persistChangeStreamResumeToken(sctx mongo.SessionContext, cs *mongo.ChangeStream) error {
+func (verifier *Verifier) persistChangeStreamResumeToken(ctx context.Context, cs *mongo.ChangeStream) error {
 	token := cs.ResumeToken()
 
 	coll := verifier.getChangeStreamMetadataCollection()
 	_, err := coll.ReplaceOne(
-		sctx,
+		ctx,
 		bson.D{{"_id", "resumeToken"}},
 		token,
 		options.Replace().SetUpsert(true),

internal/verifier/migration_verifier.go

@@ -24,7 +24,6 @@ import (
 	"github.com/10gen/migration-verifier/internal/uuidutil"
 	"github.com/10gen/migration-verifier/mbson"
 	"github.com/10gen/migration-verifier/mslices"
-	"github.com/10gen/migration-verifier/msync"
 	"github.com/10gen/migration-verifier/option"
 	"github.com/olekukonko/tablewriter"
 	"github.com/pkg/errors"
@@ -130,7 +129,6 @@ type Verifier struct {
 	changeStreamWritesOffTsChan chan primitive.Timestamp
 	changeStreamErrChan         chan error
 	changeStreamDoneChan        chan struct{}
-	changeStreamLag             *msync.TypedAtomic[option.Option[time.Duration]]
 	lastChangeEventTime         *primitive.Timestamp
 	writesOffTimestamp          *primitive.Timestamp
 
@@ -202,7 +200,6 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 		changeStreamWritesOffTsChan: make(chan primitive.Timestamp),
 		changeStreamErrChan:         make(chan error),
 		changeStreamDoneChan:        make(chan struct{}),
-		changeStreamLag:             msync.NewTypedAtomic(option.None[time.Duration]()),
 		readConcernSetting:          readConcern,
 
 		// This will get recreated once gen0 starts, but we want it

internal/verifier/summary.go

@@ -424,10 +424,4 @@ func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder) {
 
 	builder.WriteString("\nMost frequently-changing namespaces:\n")
 	table.Render()
-
-	if lag, hasLag := verifier.changeStreamLag.Load().Get(); hasLag {
-		builder.WriteString(
-			fmt.Sprintf("\nChange stream lag: %s\n", reportutils.DurationToHMS(lag)),
-		)
-	}
 }