Add a calculation of change stream lag to the log.

This provides an easy, at-a-glace indication of the amount of time
between the change stream’s resume token and the cluster time.

Author: FGasper

internal/verifier/change_stream.go

@@ -178,15 +178,15 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 	return nil
 }
 
-func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.ChangeStream) error {
+func (verifier *Verifier) iterateChangeStream(sctx mongo.SessionContext, cs *mongo.ChangeStream) error {
 	var lastPersistedTime time.Time
 
 	persistResumeTokenIfNeeded := func() error {
 		if time.Since(lastPersistedTime) <= minChangeStreamPersistInterval {
 			return nil
 		}
 
-		err := verifier.persistChangeStreamResumeToken(ctx, cs)
+		err := verifier.persistChangeStreamResumeToken(sctx, cs)
 		if err == nil {
 			lastPersistedTime = time.Now()
 		}
@@ -201,8 +201,8 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 		select {
 
 		// If the context is canceled, return immmediately.
-		case <-ctx.Done():
-			return ctx.Err()
+		case <-sctx.Done():
+			return sctx.Err()
 
 		// If the changeStreamEnderChan has a message, the user has indicated that
 		// source writes are ended. This means we should exit rather than continue
@@ -234,15 +234,15 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 					break
 				}
 
-				err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
+				err = verifier.readAndHandleOneChangeEventBatch(sctx, cs)
 
 				if err != nil {
 					return err
 				}
 			}
 
 		default:
-			err = verifier.readAndHandleOneChangeEventBatch(ctx, cs)
+			err = verifier.readAndHandleOneChangeEventBatch(sctx, cs)
 
 			if err == nil {
 				err = persistResumeTokenIfNeeded()
@@ -280,14 +280,14 @@ func (verifier *Verifier) iterateChangeStream(ctx context.Context, cs *mongo.Cha
 }
 
 func (verifier *Verifier) createChangeStream(
-	ctx context.Context,
+	sctx mongo.SessionContext,
 ) (*mongo.ChangeStream, primitive.Timestamp, error) {
 	pipeline := verifier.GetChangeStreamFilter()
 	opts := options.ChangeStream().
 		SetMaxAwaitTime(1 * time.Second).
 		SetFullDocument(options.UpdateLookup)
 
-	savedResumeToken, err := verifier.loadChangeStreamResumeToken(ctx)
+	savedResumeToken, err := verifier.loadChangeStreamResumeToken(sctx)
 	if err != nil {
 		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to load persisted change stream resume token")
 	}
@@ -314,17 +314,12 @@ 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(ctx, srcChangeStream)
+	err = verifier.persistChangeStreamResumeToken(sctx, srcChangeStream)
 	if err != nil {
 		return nil, primitive.Timestamp{}, err
 	}
@@ -337,7 +332,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(sess)
+	clusterTime, err := getClusterTimeFromSession(sctx)
 	if err != nil {
 		return nil, primitive.Timestamp{}, errors.Wrap(err, "failed to read cluster time from session")
 	}
@@ -365,7 +360,14 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 				ctx,
 				verifier.logger,
 				func(i *retry.Info) error {
-					srcChangeStream, startTs, err := verifier.createChangeStream(ctx)
+					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)
 					if err != nil {
 						return err
 					}
@@ -378,7 +380,7 @@ func (verifier *Verifier) StartChangeStream(ctx context.Context) error {
 						parentThreadWaiting = false
 					}
 
-					return verifier.iterateChangeStream(ctx, srcChangeStream)
+					return verifier.iterateChangeStream(sctx, srcChangeStream)
 				},
 			)
 
@@ -433,12 +435,12 @@ func (verifier *Verifier) loadChangeStreamResumeToken(ctx context.Context) (bson
 	return token, err
 }
 
-func (verifier *Verifier) persistChangeStreamResumeToken(ctx context.Context, cs *mongo.ChangeStream) error {
+func (verifier *Verifier) persistChangeStreamResumeToken(sctx mongo.SessionContext, cs *mongo.ChangeStream) error {
 	token := cs.ResumeToken()
 
 	coll := verifier.getChangeStreamMetadataCollection()
 	_, err := coll.ReplaceOne(
-		ctx,
+		sctx,
 		bson.D{{"_id", "resumeToken"}},
 		token,
 		options.Replace().SetUpsert(true),
@@ -451,6 +453,10 @@ func (verifier *Verifier) persistChangeStreamResumeToken(ctx context.Context, cs
 
 		if err == nil {
 			logEvent = addTimestampToLogEvent(ts, logEvent)
+
+			optime := sctx.OperationTime()
+			lag := time.Second * time.Duration(optime.T-ts.T)
+			logEvent = logEvent.Stringer("lagAmount", lag)
 		} else {
 			verifier.logger.Warn().Err(err).
 				Msg("failed to extract resume token timestamp")