REP-5140 Report change events in progress logs. (#25)

Change events can cause confusion due to how the verifier tallies them alongside missing documents. This changeset tries to add clarity by logging the change events seen during a given generation.

A bit of extra debug logging is added here as well.

Author: FGasper

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

@@ -63,6 +63,10 @@ func (verifier *Verifier) HandleChangeStreamEvent(ctx context.Context, changeEve
 	case "replace":
 		fallthrough
 	case "update":
+		if err := verifier.eventRecorder.AddEvent(changeEvent); err != nil {
+			return errors.Wrapf(err, "failed to augment stats with change event: %+v", *changeEvent)
+		}
+
 		return verifier.InsertChangeEventRecheckDoc(ctx, changeEvent)
 	default:
 		return UnknownEventError{Event: changeEvent}

internal/verifier/check.go

@@ -197,6 +197,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.eventRecorder.Reset()
 
 		err := verifier.CheckWorker(ctx)
 		if err != nil {

internal/verifier/event_recorder.go

@@ -0,0 +1,91 @@
+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{}),
+	}
+}
+
+func (er EventRecorder) Reset() {
+	er.guard.Store(func(m eventRecorderMap) eventRecorderMap {
+		return 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

@@ -93,6 +93,8 @@ type Verifier struct {
 	numWorkers         int
 	failureDisplaySize int64
 
+	eventRecorder *EventRecorder
+
 	// Used only with generation 0 to defer the first
 	// progress report until after we’ve finished partitioning
 	// every collection.
@@ -195,6 +197,10 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 		changeStreamErrChan:   make(chan error),
 		changeStreamDoneChan:  make(chan struct{}),
 		readConcernSetting:    readConcern,
+
+		// This will get recreated once gen0 starts, but we want it
+		// here in case the change streams gets an event before then.
+		eventRecorder: NewEventRecorder(),
 	}
 }
 
@@ -223,6 +229,9 @@ func (verifier *Verifier) SetFailureDisplaySize(size int64) {
 }
 
 func (verifier *Verifier) WritesOff(ctx context.Context) {
+	verifier.logger.Debug().
+		Msg("WritesOff called.")
+
 	verifier.mux.Lock()
 	verifier.writesOff = true
 	verifier.mux.Unlock()
@@ -1300,6 +1309,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.eventRecorder.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

@@ -40,6 +40,7 @@ func (suite *MultiMetaVersionTestSuite) TestFailedCompareThenReplace() {
 	)
 
 	event := ParsedEvent{
+		OpType: "insert",
 		DocKey: DocKey{
 			ID: "theDocID",
 		},
@@ -95,7 +96,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 := insertRecheckDocs(ctx, verifier, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{
@@ -156,7 +157,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 := insertRecheckDocs(ctx, verifier, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 	d1 := RecheckDoc{
 		PrimaryKey: RecheckPrimaryKey{
@@ -217,13 +218,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 := insertRecheckDocs(ctx, verifier, "testDB1", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB1", "testColl2", ids, dataSizes)
+	err = insertRecheckDocs(ctx, verifier, "testDB1", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl1", ids, dataSizes)
+	err = insertRecheckDocs(ctx, verifier, "testDB2", "testColl1", ids, dataSizes)
 	suite.Require().NoError(err)
-	err = verifier.insertRecheckDocs(ctx, "testDB2", "testColl2", ids, dataSizes)
+	err = insertRecheckDocs(ctx, verifier, "testDB2", "testColl2", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
@@ -267,17 +268,17 @@ func (suite *MultiMetaVersionTestSuite) TestGenerationalClear() {
 	id2 := "b"
 	ids := []interface{}{id1, id2}
 	dataSizes := []int{1000, 1000}
-	err := verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err := insertRecheckDocs(ctx, verifier, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = insertRecheckDocs(ctx, verifier, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	verifier.generation++
 
-	err = verifier.insertRecheckDocs(ctx, "testDB", "testColl", ids, dataSizes)
+	err = insertRecheckDocs(ctx, verifier, "testDB", "testColl", ids, dataSizes)
 	suite.Require().NoError(err)
 
 	d1 := RecheckDoc{
@@ -326,3 +327,16 @@ func (suite *MultiMetaVersionTestSuite) TestGenerationalClear() {
 	results = suite.fetchRecheckDocs(ctx, verifier)
 	suite.ElementsMatch([]interface{}{}, results)
 }
+
+func insertRecheckDocs(
+	ctx context.Context,
+	verifier *Verifier,
+	dbName, collName string,
+	documentIDs []any,
+	dataSizes []int,
+) error {
+	verifier.mux.Lock()
+	defer verifier.mux.Unlock()
+
+	return verifier.insertRecheckDocsUnderLock(ctx, dbName, collName, documentIDs, dataSizes)
+}

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.eventRecorder.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()
+}

internal/verifier/unit_test_util.go

@@ -153,12 +153,16 @@ func (suite *WithMongodsTestSuite) TearDownSuite() {
 }
 
 func (suite *WithMongodsTestSuite) TearDownTest() {
+	suite.T().Logf("Tearing down test %#q", suite.T().Name())
+
 	ctx := context.Background()
 	for _, client := range []*mongo.Client{suite.srcMongoClient, suite.dstMongoClient, suite.metaMongoClient} {
 		dbNames, err := client.ListDatabaseNames(ctx, bson.D{})
 		suite.Require().NoError(err)
 		for _, dbName := range dbNames {
 			if !suite.initialDbNames[dbName] {
+				suite.T().Logf("Dropping database %#q, which seems to have been created during test %#q.", dbName, suite.T().Name())
+
 				err = client.Database(dbName).Drop(ctx)
 				suite.Require().NoError(err)
 			}