REP-5140 Report change events in progress logs.

README.md

@@ -78,7 +78,7 @@ metaURI: mongodb://localhost:28012
 1. After launching the verifier (see above), you can send it requests to get it to start verifying. The verification process is started by using the `check`command. An [optional `filter` parameter](#document-filtering) can be passed within the `check` request body to only check documents within that filter. The verification process will keep running until you tell the verifier to stop. It will keep track of the inconsistencies it has found and will keep checking those inconsistencies hoping that eventually they will resolve.
 
     ```
-    curl -H "Content-Type: application/json" -X POST -d '{}' http://127.0.0.1:27020/api/v1/check
+    curl -H "Content-Type: application/json" -d '{}' http://127.0.0.1:27020/api/v1/check
     ```
 
 

internal/verifier/change_stream.go

@@ -2,11 +2,11 @@ package verifier
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/keystring"
+	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.mongodb.org/mongo-driver/mongo"
@@ -48,6 +48,10 @@ func (verifier *Verifier) HandleChangeStreamEvent(ctx context.Context, changeEve
 	case "replace":
 		fallthrough
 	case "update":
+		if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
+			return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
+		}
+
 		return verifier.InsertChangeEventRecheckDoc(ctx, changeEvent)
 	default:
 		return errors.New(`Not supporting: "` + changeEvent.OpType + `" events`)

internal/verifier/check.go

@@ -157,6 +157,10 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 		verifier.phase = Idle
 	}()
 
+	// A “throwaway” EventRecorder so that any change events that arrive
+	// right away will have somewhere to be recorded.
+	verifier.generationEventRecorder = NewEventRecorder()
+
 	verifier.mux.RLock()
 	csRunning := verifier.changeStreamRunning
 	verifier.mux.RUnlock()
@@ -190,6 +194,7 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 	// Now enter the multi-generational steady check state
 	for {
 		verifier.generationStartTime = time.Now()
+		verifier.generationEventRecorder = NewEventRecorder()
 
 		err := verifier.CheckWorker(ctx)
 		if err != nil {
@@ -355,8 +360,15 @@ func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.Wait
 		default:
 			task, err := verifier.FindNextVerifyTaskAndUpdate()
 			if errors.Is(err, mongo.ErrNoDocuments) {
-				verifier.logger.Debug().Msgf("[Worker %d] No tasks found, sleeping...", workerNum)
-				time.Sleep(verifier.workerSleepDelayMillis * time.Millisecond)
+				delay := verifier.workerSleepDelayMillis * time.Millisecond
+
+				verifier.logger.Debug().
+					Int("workerNum", workerNum).
+					Stringer("duration", delay).
+					Msg("No tasks found. Worker sleeping ...")
+
+				time.Sleep(delay)
+
 				continue
 			} else if err != nil {
 				panic(err)

internal/verifier/event_recorder.go

@@ -0,0 +1,85 @@
+package verifier
+
+import (
+	"github.com/10gen/migration-verifier/msync"
+	"github.com/pkg/errors"
+	"golang.org/x/exp/maps"
+)
+
+type eventRecorderMap = map[string]PerNamespaceStats
+
+// EventRecorder maintains statistics on change events.
+type EventRecorder struct {
+	guard *msync.DataGuard[eventRecorderMap]
+}
+
+// PerNamespaceStats records a given namespace’s event totals by optype.
+type PerNamespaceStats struct {
+	Insert  int
+	Update  int
+	Replace int
+	Delete  int
+}
+
+func (pns PerNamespaceStats) Total() int {
+	return pns.Insert + pns.Update + pns.Replace + pns.Delete
+}
+
+// NewEventRecorder creates and returns a new EventRecorder.
+func NewEventRecorder() *EventRecorder {
+	return &EventRecorder{
+		guard: msync.NewDataGuard(eventRecorderMap{}),
+	}
+}
+
+// AddEvent adds a ParsedEvent to the EventRecorder’s statistics.
+func (er EventRecorder) AddEvent(changeEvent *ParsedEvent) error {
+	// This shouldn’t happen, but just in case:
+	if changeEvent.Ns == nil {
+		return errors.Errorf("Change event lacks a namespace: %+v", changeEvent)
+	}
+
+	nsStr := changeEvent.Ns.DB + "." + changeEvent.Ns.Coll
+
+	var err error
+
+	er.guard.Store(func(m eventRecorderMap) eventRecorderMap {
+		if _, exists := m[nsStr]; !exists {
+			m[nsStr] = PerNamespaceStats{}
+		}
+
+		nsStats := m[nsStr]
+
+		switch changeEvent.OpType {
+		case "insert":
+			nsStats.Insert++
+		case "update":
+			nsStats.Update++
+		case "replace":
+			nsStats.Replace++
+		case "delete":
+			nsStats.Delete++
+		default:
+			err = errors.Errorf("Event recorder received event with unknown optype: %+v", *changeEvent)
+		}
+
+		m[nsStr] = nsStats
+
+		return m
+	})
+
+	return err
+}
+
+// Read returns a map of the tracked change events. The map
+// indexes on namespace then event optype. Each namespace will
+// have `insert`, `update`
+func (er EventRecorder) Read() eventRecorderMap {
+	var theCopy eventRecorderMap
+
+	er.guard.Load(func(m eventRecorderMap) {
+		theCopy = maps.Clone(m)
+	})
+
+	return theCopy
+}

internal/verifier/migration_verifier.go

@@ -95,6 +95,8 @@ type Verifier struct {
 	numWorkers         int
 	failureDisplaySize int64
 
+	generationEventRecorder *EventRecorder
+
 	// Used only with generation 0 to defer the first
 	// progress report until after we’ve finished partitioning
 	// every collection.
@@ -1399,6 +1401,8 @@ func (verifier *Verifier) PrintVerificationSummary(ctx context.Context, genstatu
 		verifier.logger.Err(err).Msgf("Failed to report per-namespace statistics")
 	}
 
+	verifier.printChangeEventStatistics(strBuilder)
+
 	var statusLine string
 
 	if hasTasks {

internal/verifier/recheck.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/10gen/migration-verifier/internal/types"
+	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
@@ -34,7 +35,11 @@ type RecheckDoc struct {
 func (verifier *Verifier) InsertFailedCompareRecheckDocs(
 	namespace string, documentIDs []interface{}, dataSizes []int) error {
 	dbName, collName := SplitNamespace(namespace)
-	return verifier.insertRecheckDocs(context.Background(),
+
+	verifier.mux.Lock()
+	defer verifier.mux.Unlock()
+
+	return verifier.insertRecheckDocsUnderLock(context.Background(),
 		dbName, collName, documentIDs, dataSizes)
 }
 
@@ -48,15 +53,20 @@ func (verifier *Verifier) InsertChangeEventRecheckDoc(ctx context.Context, chang
 	// total data size for noninitial generations in the log.
 	dataSizes := []int{maxBSONObjSize}
 
-	return verifier.insertRecheckDocs(
+	verifier.mux.Lock()
+	defer verifier.mux.Unlock()
+
+	if err := verifier.generationEventRecorder.AddEvent(changeEvent); err != nil {
+		return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
+	}
+
+	return verifier.insertRecheckDocsUnderLock(
 		ctx, changeEvent.Ns.DB, changeEvent.Ns.Coll, documentIDs, dataSizes)
 }
 
-func (verifier *Verifier) insertRecheckDocs(
+func (verifier *Verifier) insertRecheckDocsUnderLock(
 	ctx context.Context,
 	dbName, collName string, documentIDs []interface{}, dataSizes []int) error {
-	verifier.mux.Lock()
-	defer verifier.mux.Unlock()
 
 	generation, _ := verifier.getGenerationWhileLocked()
 

internal/verifier/recheck_test.go

@@ -95,7 +95,7 @@ func (suite *MultiMetaVersionTestSuite) TestLargeIDInsertions() {
 	id3 := strings.Repeat("c", overlyLarge)
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{overlyLarge, overlyLarge, overlyLarge}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{
@@ -156,7 +156,7 @@ func (suite *MultiMetaVersionTestSuite) TestLargeDataInsertions() {
 	id3 := "c"
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{400 * 1024, 700 * 1024, 1024}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 	d1 := RecheckDoc{
 		PrimaryKey: RecheckPrimaryKey{
@@ -217,13 +217,13 @@ func (suite *MultiMetaVersionTestSuite) TestMultipleNamespaces() {
 	id3 := "c"
 	ids := []interface{}{id1, id2, id3}
 	dataSizes := []int{1000, 1000, 1000}
-	err := verifier.insertRecheckDocs(ctx, "testDB1", "testColl1", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB1", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB1", "testColl2", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB1", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl1", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB2", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl2", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB2", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
@@ -267,17 +267,17 @@ func (suite *MultiMetaVersionTestSuite) TestGenerationalClear() {
 	id2 := "b"
 	ids := []interface{}{id1, id2}
 	dataSizes := []int{1000, 1000}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = verifier.insertRecheckDocsUnderLock(ctx, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{

internal/verifier/summary.go

@@ -7,14 +7,19 @@ package verifier
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/olekukonko/tablewriter"
+	"golang.org/x/exp/maps"
 )
 
+const changeEventsTableMaxSize = 10
+
 // NOTE: Each of the following should print one trailing and one final
 // newline.
 
@@ -362,3 +367,61 @@ func (verifier *Verifier) printMismatchInvestigationNotes(strBuilder *strings.Bu
 		strBuilder.WriteString(line + "\n")
 	}
 }
+
+func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder) {
+	nsStats := verifier.generationEventRecorder.Read()
+
+	activeNamespacesCount := len(nsStats)
+
+	totalEvents := 0
+	nsTotals := map[string]int{}
+	for ns, events := range nsStats {
+		nsTotals[ns] = events.Total()
+		totalEvents += nsTotals[ns]
+	}
+
+	eventsDescr := "none"
+	if totalEvents > 0 {
+		eventsDescr = fmt.Sprintf("%d total, across %d namespace(s)", totalEvents, activeNamespacesCount)
+	}
+
+	builder.WriteString(fmt.Sprintf("\nChange events this generation: %s\n", eventsDescr))
+
+	if totalEvents == 0 {
+		return
+	}
+
+	reverseSortedNamespaces := maps.Keys(nsTotals)
+	sort.Slice(
+		reverseSortedNamespaces,
+		func(i, j int) bool {
+			return nsTotals[reverseSortedNamespaces[i]] > nsTotals[reverseSortedNamespaces[j]]
+		},
+	)
+
+	// Only report the busiest namespaces.
+	if len(reverseSortedNamespaces) > changeEventsTableMaxSize {
+		reverseSortedNamespaces = reverseSortedNamespaces[:changeEventsTableMaxSize]
+	}
+
+	table := tablewriter.NewWriter(builder)
+	table.SetHeader([]string{"Namespace", "Insert", "Update", "Replace", "Delete", "Total"})
+
+	for _, ns := range reverseSortedNamespaces {
+		curNsStats := nsStats[ns]
+
+		table.Append(
+			append(
+				[]string{ns},
+				strconv.Itoa(curNsStats.Insert),
+				strconv.Itoa(curNsStats.Update),
+				strconv.Itoa(curNsStats.Replace),
+				strconv.Itoa(curNsStats.Delete),
+				strconv.Itoa(curNsStats.Total()),
+			),
+		)
+	}
+
+	builder.WriteString("\nMost frequently-changing namespaces:\n")
+	table.Render()
+}

msync/dataguard.go

@@ -0,0 +1,34 @@
+package msync
+
+import "sync"
+
+// DataGuard encapsulates a value with a mutex to ensure that anything that
+// accesses it does so in a race-safe way.
+type DataGuard[T any] struct {
+	mutex sync.RWMutex
+	value T
+}
+
+// NewDataGuard returns a new DataGuard that wraps the given value.
+func NewDataGuard[T any](val T) *DataGuard[T] {
+	return &DataGuard[T]{
+		value: val,
+	}
+}
+
+// Load runs the given callback, passing it the DataGuard’s stored value.
+func (l *DataGuard[T]) Load(cb func(T)) {
+	l.mutex.RLock()
+	defer l.mutex.RUnlock()
+
+	cb(l.value)
+}
+
+// Store is like Load but will replace the DataGuard’s stored value with the
+// callback’s return.
+func (l *DataGuard[T]) Store(cb func(T) T) {
+	l.mutex.Lock()
+	defer l.mutex.Unlock()
+
+	l.value = cb(l.value)
+}