Merge branch 'main' into REP-5329-retry-consistently

Author: FGasper

internal/verifier/check.go

@@ -484,6 +484,8 @@ func (verifier *Verifier) work(ctx context.Context, workerNum int) error {
 			continue
 		}
 
+		verifier.workerTracker.Set(workerNum, *task)
+
 		switch task.Type {
 		case verificationTaskVerifyCollection:
 			err := verifier.ProcessCollectionVerificationTask(ctx, workerNum, &task)

internal/verifier/migration_verifier.go

@@ -142,6 +142,8 @@ type Verifier struct {
 
 	pprofInterval time.Duration
 
+	workerTracker *WorkerTracker
+
 	verificationStatusCheckInterval time.Duration
 }
 
@@ -206,6 +208,8 @@ func NewVerifier(settings VerifierSettings) *Verifier {
 		// here in case the change streams gets an event before then.
 		eventRecorder: NewEventRecorder(),
 
+		workerTracker: NewWorkerTracker(NumWorkers),
+
 		verificationStatusCheckInterval: 15 * time.Second,
 	}
 }
@@ -1518,6 +1522,14 @@ func (verifier *Verifier) PrintVerificationSummary(ctx context.Context, genstatu
 
 	verifier.printChangeEventStatistics(strBuilder)
 
+	// Only print the worker status table if debug logging is enabled.
+	if verifier.logger.Debug().Enabled() {
+		switch genstatus {
+		case Gen0MetadataAnalysisComplete, GenerationInProgress:
+			verifier.printWorkerStatus(strBuilder)
+		}
+	}
+
 	var statusLine string
 
 	if hasTasks {

internal/verifier/summary.go

@@ -15,6 +15,7 @@ import (
 	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/olekukonko/tablewriter"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"golang.org/x/exp/maps"
 )
 
@@ -436,3 +437,48 @@ func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder) {
 	builder.WriteString("\nMost frequently-changing namespaces:\n")
 	table.Render()
 }
+
+func (verifier *Verifier) printWorkerStatus(builder *strings.Builder) {
+
+	table := tablewriter.NewWriter(builder)
+	table.SetHeader([]string{"Thread #", "Namespace", "Task", "Time Elapsed"})
+
+	wsmap := verifier.workerTracker.Load()
+
+	activeThreadCount := 0
+	for w := 0; w <= verifier.numWorkers; w++ {
+		if wsmap[w].TaskID == nil {
+			continue
+		}
+
+		activeThreadCount++
+
+		var taskIdStr string
+
+		switch id := wsmap[w].TaskID.(type) {
+		case primitive.ObjectID:
+			theBytes, _ := id.MarshalText()
+
+			taskIdStr = string(theBytes)
+		default:
+			taskIdStr = fmt.Sprintf("%s", wsmap[w].TaskID)
+		}
+
+		table.Append(
+			[]string{
+				strconv.Itoa(w),
+				wsmap[w].Namespace,
+				taskIdStr,
+				reportutils.DurationToHMS(time.Since(wsmap[w].StartTime)),
+			},
+		)
+	}
+
+	builder.WriteString(fmt.Sprintf(
+		"\nActive worker threads (%d of %d):\n",
+		activeThreadCount,
+		verifier.numWorkers,
+	))
+
+	table.Render()
+}

internal/verifier/worker_tracker.go

@@ -0,0 +1,72 @@
+package verifier
+
+import (
+	"time"
+
+	"github.com/10gen/migration-verifier/msync"
+	"golang.org/x/exp/maps"
+)
+
+// WorkerTracker holds certain data points about each worker thread
+// in a check generation. It is thread-safe.
+type WorkerTracker struct {
+	guard *msync.DataGuard[WorkerStatusMap]
+}
+
+// WorkerStatusMap represents the status of each worker,
+// indexed by worker number (which start at 0).
+type WorkerStatusMap = map[int]WorkerStatus
+
+// WorkerStatus details the work that an individual worker thread
+// is doing.
+type WorkerStatus struct {
+	TaskID    any
+	TaskType  verificationTaskType
+	Namespace string
+	StartTime time.Time
+}
+
+// NewWorkerTracker creates and returns a WorkerTracker.
+func NewWorkerTracker(workersCount int) *WorkerTracker {
+	wsmap := WorkerStatusMap{}
+	for i := 0; i < workersCount; i++ {
+		wsmap[i] = WorkerStatus{}
+	}
+	return &WorkerTracker{
+		guard: msync.NewDataGuard(wsmap),
+	}
+}
+
+// Set updates the worker’s state in the WorkerTracker.
+func (wt *WorkerTracker) Set(workerNum int, task VerificationTask) {
+	wt.guard.Store(func(m WorkerStatusMap) WorkerStatusMap {
+		m[workerNum] = WorkerStatus{
+			TaskID:    task.PrimaryKey,
+			TaskType:  task.Type,
+			Namespace: task.QueryFilter.Namespace,
+			StartTime: time.Now(),
+		}
+
+		return m
+	})
+}
+
+// Unset tells the WorkerTracker that the worker is now inactive.
+func (wt *WorkerTracker) Unset(workerNum int) {
+	wt.guard.Store(func(m WorkerStatusMap) WorkerStatusMap {
+		m[workerNum] = WorkerStatus{}
+
+		return m
+	})
+}
+
+// Load duplicates and returns the WorkerTracker’s internal
+// state map.
+func (wt *WorkerTracker) Load() WorkerStatusMap {
+	var wtmap WorkerStatusMap
+	wt.guard.Load(func(m map[int]WorkerStatus) {
+		wtmap = maps.Clone(m)
+	})
+
+	return wtmap
+}