Put lag in log.

Author: FGasper

internal/verifier/change_stream.go

@@ -8,6 +8,7 @@ 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"
@@ -138,14 +139,14 @@ 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(
-	ctx context.Context,
+	sctx mongo.SessionContext,
 	cs *mongo.ChangeStream,
 ) error {
 	eventsRead := 0
 	var changeEventBatch []ParsedEvent
 
 	for hasEventInBatch := true; hasEventInBatch; hasEventInBatch = cs.RemainingBatchLength() > 0 {
-		gotEvent := cs.TryNext(ctx)
+		gotEvent := cs.TryNext(sctx)
 
 		if cs.Err() != nil {
 			return errors.Wrap(cs.Err(), "change stream iteration failed")
@@ -170,7 +171,15 @@ func (verifier *Verifier) readAndHandleOneChangeEventBatch(
 		return nil
 	}
 
-	err := verifier.HandleChangeStreamEvents(ctx, changeEventBatch)
+	// 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)
 	if err != nil {
 		return errors.Wrap(err, "failed to handle change events")
 	}
@@ -435,6 +444,20 @@ 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 {
 	token := cs.ResumeToken()
 
@@ -453,10 +476,6 @@ func (verifier *Verifier) persistChangeStreamResumeToken(sctx mongo.SessionConte
 
 		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")

internal/verifier/migration_verifier.go

@@ -22,6 +22,8 @@ import (
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/10gen/migration-verifier/internal/uuidutil"
+	"github.com/10gen/migration-verifier/msync"
+	"github.com/10gen/migration-verifier/option"
 	"github.com/olekukonko/tablewriter"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
@@ -126,6 +128,7 @@ 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
 
@@ -197,6 +200,7 @@ 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,4 +424,10 @@ 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)),
+		)
+	}
 }

msync/typed_atomic.go

@@ -0,0 +1,50 @@
+package msync
+
+import "sync/atomic"
+
+// TypedAtomic is a type-safe wrapper around the standard-library atomic.Value.
+// TypedAtomic serves largely the same purpose as atomic.Pointer but stores
+// the value itself rather than a pointer to it. This is often more ergonomic
+// than an atomic.Pointer: it can be used to store constants directly (where
+// taking a pointer is inconvenient), and it defaults to the type's zero value
+// rather than a nil pointer.
+type TypedAtomic[T any] struct {
+	v atomic.Value
+}
+
+// NewTypedAtomic returns a new TypedAtomic, initialized to val.
+func NewTypedAtomic[T any](val T) *TypedAtomic[T] {
+	var v atomic.Value
+	v.Store(val)
+	return &TypedAtomic[T]{v}
+}
+
+// Load returns the value set by the most recent Store. It returns the zero
+// value for the type if there has been no call to Store.
+func (ta *TypedAtomic[T]) Load() T {
+	return orZero[T](ta.v.Load())
+}
+
+// Store sets the value TypedAtomic to val. Store(nil) panics.
+func (ta *TypedAtomic[T]) Store(val T) {
+	ta.v.Store(val)
+}
+
+// Swap stores newVal into the TypedAtomic and returns the previous value. It
+// returns the zero value for the type if the value is empty.
+func (ta *TypedAtomic[T]) Swap(newVal T) T {
+	return orZero[T](ta.v.Swap(newVal))
+}
+
+// CompareAndSwap executes the compare-and-swap operation for the TypedAtomic.
+func (ta *TypedAtomic[T]) CompareAndSwap(oldVal, newVal T) bool {
+	return ta.v.CompareAndSwap(oldVal, newVal)
+}
+
+func orZero[T any](val any) T {
+	if val == nil {
+		return *new(T)
+	}
+
+	return val.(T)
+}

msync/typed_atomic_test.go

@@ -0,0 +1,59 @@
+package msync
+
+import (
+	"sync"
+)
+
+func (s *unitTestSuite) TestTypedAtomic() {
+	ta := NewTypedAtomic(42)
+
+	s.Require().Equal(42, ta.Load())
+	s.Require().False(ta.CompareAndSwap(17, 99))
+	s.Require().True(ta.CompareAndSwap(42, 99))
+	s.Require().Equal(99, ta.Load())
+	s.Require().Equal(99, ta.Swap(42))
+	s.Require().Equal(42, ta.Load())
+
+	ta.Store(17)
+	s.Require().Equal(17, ta.Load())
+
+	// This block is for race detection under -race.
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ta.Load()
+			ta.Store(i)
+		}()
+	}
+	wg.Wait()
+}
+
+func (s *unitTestSuite) TestAtomicZeroValues() {
+	s.Run("string", func() {
+		var ta TypedAtomic[string]
+		s.Require().Equal("", ta.Load())
+		s.Require().Equal("", ta.Swap("foo"))
+		s.Require().Equal("foo", ta.Load())
+	})
+
+	s.Run("int", func() {
+		var ta TypedAtomic[int]
+		s.Require().Equal(0, ta.Load())
+		s.Require().Equal(0, ta.Swap(42))
+		s.Require().Equal(42, ta.Load())
+	})
+
+	s.Run("arbitrary data", func() {
+		type data struct {
+			I int
+			S string
+		}
+
+		var ta TypedAtomic[data]
+		s.Require().Equal(data{}, ta.Load())
+		s.Require().Equal(data{}, ta.Swap(data{76, "trombones"}))
+		s.Require().Equal(data{76, "trombones"}, ta.Load())
+	})
+}