initial code

Author: FGasper

go.mod

@@ -3,24 +3,25 @@ module github.com/10gen/migration-verifier
 go 1.18
 
 require (
+	github.com/cespare/permute/v2 v2.0.0-beta2
 	github.com/deckarep/golang-set/v2 v2.3.0
 	github.com/dustin/go-humanize v1.0.1
 	github.com/gin-gonic/gin v1.8.1
 	github.com/google/uuid v1.3.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pkg/errors v0.9.1
 	github.com/rs/zerolog v1.28.0
+	github.com/samber/lo v1.47.0
 	github.com/stretchr/testify v1.8.0
 	github.com/urfave/cli v1.22.9
 	go.mongodb.org/mongo-driver v1.10.2
-	golang.org/x/exp v0.0.0-20230321023759-10a507213a29
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c
+	golang.org/x/sync v0.8.0
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 )
 
 require (
 	github.com/BurntSushi/toml v0.3.1 // indirect
-	github.com/cespare/permute/v2 v2.0.0-beta2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
@@ -49,8 +50,7 @@ require (
 	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect
 	golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 // indirect
 	golang.org/x/sys v0.1.0 // indirect
-	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	golang.org/x/text v0.16.0 // indirect
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

internal/verifier/change_stream.go

@@ -2,10 +2,11 @@ package verifier
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/10gen/migration-verifier/internal/keystring"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
@@ -48,6 +49,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

@@ -190,6 +190,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 {
@@ -227,6 +228,7 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 			verifier.lastGeneration = true
 		}
 		verifier.generation++
+
 		verifier.phase = Recheck
 		err = verifier.GenerateRecheckTasks(ctx)
 		if err != nil {

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 (et 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
+
+	et.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 (et EventRecorder) Read() eventRecorderMap {
+	var theCopy eventRecorderMap
+
+	et.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.
@@ -1400,6 +1402,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/summary.go

@@ -7,12 +7,16 @@ package verifier
 import (
 	"context"
 	"fmt"
+	"slices"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/olekukonko/tablewriter"
+	"github.com/samber/lo"
+	"golang.org/x/exp/maps"
 )
 
 // NOTE: Each of the following should print one trailing and one final
@@ -362,3 +366,67 @@ 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 := lo.Ternary(
+		totalEvents == 0,
+		"0",
+		fmt.Sprintf("%d total, across %d namespace(s)", totalEvents, activeNamespacesCount),
+	)
+	builder.WriteString(fmt.Sprintf("Change events this generation: %s\n", eventsDescr))
+
+	if totalEvents == 0 {
+		return
+	}
+
+	sortedNamespaces := maps.Keys(nsTotals)
+	slices.SortFunc(
+		sortedNamespaces,
+		func(ns1, ns2 string) int {
+			if nsTotals[ns1] < nsTotals[ns2] {
+				return 1
+			}
+
+			if nsTotals[ns1] > nsTotals[ns2] {
+				return -1
+			}
+
+			return 0
+		},
+	)
+
+	// Only report the busiest namespaces.
+	sortedNamespaces = sortedNamespaces[:10]
+
+	table := tablewriter.NewWriter(builder)
+	table.SetHeader([]string{"Namespace", "Insert", "Update", "Replace", "Delete", "Total"})
+
+	for _, ns := range sortedNamespaces {
+		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)
+}

msync/dataguard_test.go

@@ -0,0 +1,43 @@
+package msync
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type unitTestSuite struct {
+	suite.Suite
+}
+
+func TestUnitTestSuite(t *testing.T) {
+	suite.Run(t, &unitTestSuite{})
+}
+
+func (s *unitTestSuite) TestDataGuard() {
+	l := NewDataGuard(42)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		var delta int
+		if i%2 == 0 {
+			delta = 2
+		} else {
+			delta = -1
+		}
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			l.Store(func(v int) int {
+				return v + delta
+			})
+		}()
+	}
+	wg.Wait()
+
+	l.Load(func(v int) {
+		s.Require().Equal(92, v)
+	})
+}