REP-6438 Include currently-running task stats in baseline logs. (#130)

Currently the non-debug logs offer no insight into currently-active tasks’ progress. So if tasks take a long time, non-debug logging will show now progress until one of those tasks finishes.

This amends logging thus:
- Document & byte tallies now incorporate in-progress tasks’ document & byte counts.
- “Active document comparison threads” is now printed.
- Each line of the “Namespaces in progress” table shows that namespace’s count of active threads.

Author: FGasper

internal/verifier/compare.go

@@ -9,7 +9,6 @@ import (
 	"github.com/10gen/migration-verifier/chanutil"
 	"github.com/10gen/migration-verifier/contextplus"
 	"github.com/10gen/migration-verifier/dockey"
-	"github.com/10gen/migration-verifier/internal/reportutils"
 	"github.com/10gen/migration-verifier/internal/retry"
 	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/10gen/migration-verifier/internal/util"
@@ -231,13 +230,10 @@ func (verifier *Verifier) compareDocsFromChannels(
 
 					srcDocCount++
 					srcByteCount += types.ByteCount(len(srcDocWithTs.doc))
-					verifier.workerTracker.SetDetail(
+					verifier.workerTracker.SetSrcCounts(
 						workerNum,
-						fmt.Sprintf(
-							"%s documents (%s)",
-							reportutils.FmtReal(srcDocCount),
-							reportutils.FmtBytes(srcByteCount),
-						),
+						srcDocCount,
+						srcByteCount,
 					)
 				}
 

internal/verifier/migration_verifier_test.go

@@ -471,7 +471,7 @@ func (suite *IntegrationTestSuite) TestVerifierFetchDocuments() {
 	)
 }
 
-func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Recheck() {
+func (suite *IntegrationTestSuite) TestGetPersistedNamespaceStatistics_Recheck() {
 	ctx := suite.Context()
 	verifier := suite.BuildVerifier()
 
@@ -522,7 +522,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Recheck() {
 		suite.Require().NoError(verifier.GenerateRecheckTasksWhileLocked(ctx))
 	}()
 
-	stats, err := verifier.GetNamespaceStatistics(ctx)
+	stats, err := verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -547,7 +547,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 	ctx := suite.Context()
 	verifier := suite.BuildVerifier()
 
-	stats, err := verifier.GetNamespaceStatistics(ctx)
+	stats, err := verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -565,7 +565,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 	task1, err := verifier.InsertCollectionVerificationTask(ctx, "mydb.coll1")
 	suite.Require().NoError(err)
 
-	stats, err = verifier.GetNamespaceStatistics(ctx)
+	stats, err = verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -593,7 +593,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 	err = verifier.UpdateVerificationTask(ctx, task1)
 	suite.Require().NoError(err)
 
-	stats, err = verifier.GetNamespaceStatistics(ctx)
+	stats, err = verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -643,7 +643,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 		task2parts[i] = task2part
 	}
 
-	stats, err = verifier.GetNamespaceStatistics(ctx)
+	stats, err = verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -671,7 +671,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 	err = verifier.UpdateVerificationTask(ctx, task1parts[0])
 	suite.Require().NoError(err)
 
-	stats, err = verifier.GetNamespaceStatistics(ctx)
+	stats, err = verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(
@@ -710,7 +710,7 @@ func (suite *IntegrationTestSuite) TestGetNamespaceStatistics_Gen0() {
 	err = verifier.UpdateVerificationTask(ctx, task2parts[1])
 	suite.Require().NoError(err)
 
-	stats, err = verifier.GetNamespaceStatistics(ctx)
+	stats, err = verifier.GetPersistedNamespaceStatistics(ctx)
 	suite.Require().NoError(err)
 
 	suite.Assert().Equal(

internal/verifier/statistics.go

@@ -168,10 +168,10 @@ const perNsStatsPipelineTemplate = `[
 var templateOnce sync.Once
 var jsonTemplate *template.Template
 
-// GetNamespaceStatistics queries the verifier’s metadata for statistics
+// GetPersistedNamespaceStatistics queries the verifier’s metadata for statistics
 // on progress for each namespace. The returned array is sorted by
 // Namespace and contains one entry for each namespace.
-func (verifier *Verifier) GetNamespaceStatistics(ctx context.Context) ([]NamespaceStats, error) {
+func (verifier *Verifier) GetPersistedNamespaceStatistics(ctx context.Context) ([]NamespaceStats, error) {
 	generation, _ := verifier.getGeneration()
 
 	templateOnce.Do(func() {

internal/verifier/summary.go

@@ -240,7 +240,7 @@ OUTB:
 
 // Boolean returned indicates whether this generation has any tasks.
 func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuilder *strings.Builder, now time.Time) (bool, error) {
-	stats, err := verifier.GetNamespaceStatistics(ctx)
+	stats, err := verifier.GetPersistedNamespaceStatistics(ctx)
 	if err != nil {
 		return false, err
 	}
@@ -282,6 +282,25 @@ func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuild
 
 	elapsed := now.Sub(verifier.generationStartTime)
 
+	var activeWorkers int
+	perNamespaceWorkerStats := verifier.getPerNamespaceWorkerStats()
+	for _, nsWorkerStats := range perNamespaceWorkerStats {
+		for _, workerStats := range nsWorkerStats {
+			activeWorkers++
+			comparedDocs += workerStats.SrcDocCount
+			comparedBytes += workerStats.SrcByteCount
+		}
+	}
+
+	if activeWorkers > 0 {
+		fmt.Fprintf(
+			strBuilder,
+			"Active document comparison threads: %d of %d\n",
+			activeWorkers,
+			verifier.numWorkers,
+		)
+	}
+
 	docsPerSecond := float64(comparedDocs) / elapsed.Seconds()
 	bytesPerSecond := float64(comparedBytes) / elapsed.Seconds()
 	perSecondDataUnit := reportutils.FindBestUnit(bytesPerSecond)
@@ -336,7 +355,7 @@ func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuild
 
 	table := tablewriter.NewWriter(strBuilder)
 
-	headers := []string{"Src Namespace", "Src Docs Compared"}
+	headers := []string{"Src Namespace", "Threads", "Src Docs Compared"}
 	if showDataTotals {
 		headers = append(headers, "Src Data Compared")
 	}
@@ -351,18 +370,32 @@ func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuild
 
 		tableHasRows = true
 
-		row := []string{result.Namespace}
+		var threads int
+
+		docsCompared := result.DocsCompared
+		bytesCompared := result.BytesCompared
+
+		if nsWorkerStats, ok := perNamespaceWorkerStats[result.Namespace]; ok {
+			threads = len(nsWorkerStats)
+
+			for _, workerStats := range nsWorkerStats {
+				docsCompared += workerStats.SrcDocCount
+				bytesCompared += workerStats.SrcByteCount
+			}
+		}
+
+		row := []string{result.Namespace, reportutils.FmtReal(threads)}
 
 		var docsCell string
 
 		if result.TotalDocs > 0 {
 			docsCell = fmt.Sprintf("%s of %s (%s%%)",
-				reportutils.FmtReal(result.DocsCompared),
+				reportutils.FmtReal(docsCompared),
 				reportutils.FmtReal(result.TotalDocs),
-				reportutils.FmtPercent(result.DocsCompared, result.TotalDocs),
+				reportutils.FmtPercent(docsCompared, result.TotalDocs),
 			)
 		} else {
-			docsCell = reportutils.FmtReal(result.DocsCompared)
+			docsCell = reportutils.FmtReal(docsCompared)
 		}
 
 		row = append(row, docsCell)
@@ -374,16 +407,16 @@ func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuild
 				dataUnit := reportutils.FindBestUnit(result.TotalBytes)
 
 				dataCell = fmt.Sprintf("%s of %s %s (%s%%)",
-					reportutils.BytesToUnit(result.BytesCompared, dataUnit),
+					reportutils.BytesToUnit(bytesCompared, dataUnit),
 					reportutils.BytesToUnit(result.TotalBytes, dataUnit),
 					dataUnit,
-					reportutils.FmtPercent(result.BytesCompared, result.TotalBytes),
+					reportutils.FmtPercent(bytesCompared, result.TotalBytes),
 				)
 			} else {
-				dataUnit := reportutils.FindBestUnit(result.BytesCompared)
+				dataUnit := reportutils.FindBestUnit(bytesCompared)
 
 				dataCell = fmt.Sprintf("%s %s",
-					reportutils.BytesToUnit(result.BytesCompared, dataUnit),
+					reportutils.BytesToUnit(bytesCompared, dataUnit),
 					dataUnit,
 				)
 			}
@@ -403,7 +436,7 @@ func (verifier *Verifier) printNamespaceStatistics(ctx context.Context, strBuild
 }
 
 func (verifier *Verifier) printEndOfGenerationStatistics(ctx context.Context, strBuilder *strings.Builder, now time.Time) (bool, error) {
-	stats, err := verifier.GetNamespaceStatistics(ctx)
+	stats, err := verifier.GetPersistedNamespaceStatistics(ctx)
 	if err != nil {
 		return false, err
 	}
@@ -564,6 +597,25 @@ func (verifier *Verifier) printChangeEventStatistics(builder *strings.Builder, n
 	}
 }
 
+func (verifier *Verifier) getPerNamespaceWorkerStats() map[string][]WorkerStatus {
+	wsmap := verifier.workerTracker.Load()
+
+	retMap := map[string][]WorkerStatus{}
+
+	for _, workerStats := range wsmap {
+		if workerStats.TaskID == nil {
+			continue
+		}
+
+		retMap[workerStats.Namespace] = append(
+			retMap[workerStats.Namespace],
+			workerStats,
+		)
+	}
+
+	return retMap
+}
+
 func (verifier *Verifier) printWorkerStatus(builder *strings.Builder, now time.Time) {
 
 	table := tablewriter.NewWriter(builder)
@@ -572,7 +624,7 @@ func (verifier *Verifier) printWorkerStatus(builder *strings.Builder, now time.T
 	wsmap := verifier.workerTracker.Load()
 
 	activeThreadCount := 0
-	for w := 0; w <= verifier.numWorkers; w++ {
+	for w := range verifier.numWorkers {
 		if wsmap[w].TaskID == nil {
 			continue
 		}
@@ -590,23 +642,27 @@ func (verifier *Verifier) printWorkerStatus(builder *strings.Builder, now time.T
 			taskIdStr = fmt.Sprintf("%s", wsmap[w].TaskID)
 		}
 
+		var detail string
+		if wsmap[w].TaskType == verificationTaskVerifyDocuments {
+			detail = fmt.Sprintf(
+				"%s documents (%s)",
+				reportutils.FmtReal(wsmap[w].SrcDocCount),
+				reportutils.FmtBytes(wsmap[w].SrcByteCount),
+			)
+		}
+
 		table.Append(
 			[]string{
 				reportutils.FmtReal(w),
 				wsmap[w].Namespace,
 				taskIdStr,
 				reportutils.DurationToHMS(now.Sub(wsmap[w].StartTime)),
-				wsmap[w].Detail,
+				detail,
 			},
 		)
 	}
 
-	fmt.Fprintf(
-		builder,
-		"\nActive worker threads (%s of %s):\n",
-		reportutils.FmtReal(activeThreadCount),
-		reportutils.FmtReal(verifier.numWorkers),
-	)
+	fmt.Fprintf(builder, "\nWorker thread details:\n")
 
 	table.Render()
 }

internal/verifier/worker_tracker.go

@@ -3,6 +3,7 @@ package verifier
 import (
 	"time"
 
+	"github.com/10gen/migration-verifier/internal/types"
 	"github.com/10gen/migration-verifier/msync"
 	"golang.org/x/exp/maps"
 )
@@ -20,11 +21,12 @@ 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
-	Detail    string
+	TaskID       any
+	StartTime    time.Time
+	TaskType     verificationTaskType
+	Namespace    string
+	SrcDocCount  types.DocumentCount
+	SrcByteCount types.ByteCount
 }
 
 // NewWorkerTracker creates and returns a WorkerTracker.
@@ -52,10 +54,11 @@ func (wt *WorkerTracker) Set(workerNum int, task VerificationTask) {
 	})
 }
 
-func (wt *WorkerTracker) SetDetail(workerNum int, detail string) {
+func (wt *WorkerTracker) SetSrcCounts(workerNum int, docs types.DocumentCount, bytes types.ByteCount) {
 	wt.guard.Store(func(m WorkerStatusMap) WorkerStatusMap {
 		status := m[workerNum]
-		status.Detail = detail
+		status.SrcDocCount = docs
+		status.SrcByteCount = bytes
 		m[workerNum] = status
 
 		return m