[stats only] Fix broken scorch stats (#2380)

- When introducing a merge, we always incremented `File` based stats
even when the merge could have originated from in-memory segment merge.
Fix this by tracking the origin of the merge request and updating the
respective stat.
- Made merge related stats symmetric across file and memory-based merge,
added:
    - `TotMemMergeIntroductions`
    - `TotMemMergeIntroductionsDone`
    - `TotMemMergeZapIntroductionTime`
    - `MaxMemMergeZapIntroductionTime`
- Removed unused `TotIntroduceRevertBeg` and `TotIntroduceRevertEnd`
stats as we do not support any revert operation anyway.
- Fix a data race when computing the `MaxMemMergeZapTime` stat.

Author: CascadingRadium

index/scorch/introducer.go

@@ -361,29 +361,29 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 
 	var running, docsToPersistCount, memSegments, fileSegments uint64
 	var droppedSegmentFiles []string
-	newSegmentDeleted := make([]*roaring.Bitmap, len(nextMerge.new))
-	for i := range newSegmentDeleted {
+	newSegmentsDeleted := make([]*roaring.Bitmap, len(nextMerge.newSegments))
+	for i := range newSegmentsDeleted {
 		// create a bitmaps to track the obsoletes per newly merged segments
-		newSegmentDeleted[i] = roaring.NewBitmap()
+		newSegmentsDeleted[i] = roaring.NewBitmap()
 	}
 
 	// iterate through current segments
 	for i := range root.segment {
 		segmentID := root.segment[i].id
-		if segSnapAtMerge, ok := nextMerge.mergedSegHistory[segmentID]; ok {
+		if history, ok := nextMerge.mergedSegHistory[segmentID]; ok {
 			// this segment is going away, see if anything else was deleted since we started the merge
-			if segSnapAtMerge != nil && root.segment[i].deleted != nil {
+			if history != nil && root.segment[i].deleted != nil {
 				// assume all these deletes are new
 				deletedSince := root.segment[i].deleted
 				// if we already knew about some of them, remove
-				if segSnapAtMerge.oldSegment.deleted != nil {
-					deletedSince = roaring.AndNot(root.segment[i].deleted, segSnapAtMerge.oldSegment.deleted)
+				if history.oldSegment.deleted != nil {
+					deletedSince = roaring.AndNot(root.segment[i].deleted, history.oldSegment.deleted)
 				}
 				deletedSinceItr := deletedSince.Iterator()
 				for deletedSinceItr.HasNext() {
 					oldDocNum := deletedSinceItr.Next()
-					newDocNum := segSnapAtMerge.oldNewDocIDs[oldDocNum]
-					newSegmentDeleted[segSnapAtMerge.batchID].Add(uint32(newDocNum))
+					newDocNum := history.oldNewDocIDs[oldDocNum]
+					newSegmentsDeleted[history.batchID].Add(uint32(newDocNum))
 				}
 			}
 
@@ -430,28 +430,36 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 			for obsoletedIter.HasNext() {
 				oldDocNum := obsoletedIter.Next()
 				newDocNum := ss.oldNewDocIDs[oldDocNum]
-				newSegmentDeleted[ss.batchID].Add(uint32(newDocNum))
+				newSegmentsDeleted[ss.batchID].Add(uint32(newDocNum))
 			}
 		}
 	}
 
-	skipped := make([]bool, len(nextMerge.new))
+	skipped := make([]bool, len(nextMerge.newSegments))
 	// make the newly merged segments part of the newSnapshot being constructed
-	for i, newMergedSegment := range nextMerge.new {
+	for i, newMergedSegment := range nextMerge.newSegments {
+		if newMergedSegment == nil {
+			if nextMerge.fileMerge {
+				atomic.AddUint64(&s.stats.TotFileMergeIntroductionsSkipped, 1)
+			} else {
+				atomic.AddUint64(&s.stats.TotMemMergeIntroductionsSkipped, 1)
+			}
+			skipped[i] = true
+			continue
+		}
 		// checking if this newly merged segment is worth keeping based on
 		// obsoleted doc count since the merge intro started
-		if newMergedSegment != nil &&
-			newMergedSegment.Count() > newSegmentDeleted[i].GetCardinality() {
+		if newMergedSegment.Count() > newSegmentsDeleted[i].GetCardinality() {
 			stats := newFieldStats()
 			if fsr, ok := newMergedSegment.(segment.FieldStatsReporter); ok {
 				fsr.UpdateFieldStats(stats)
 			}
 
 			// put the merged segment at the end of newSnapshot
 			newSnapshot.segment = append(newSnapshot.segment, &SegmentSnapshot{
-				id:         nextMerge.id[i],
-				segment:    newMergedSegment, // take ownership for nextMerge.new's ref-count
-				deleted:    newSegmentDeleted[i],
+				id:         nextMerge.newSegmentIDs[i],
+				segment:    newMergedSegment, // take ownership for newMergedSegment's ref-count
+				deleted:    newSegmentsDeleted[i],
 				stats:      stats,
 				cachedDocs: &cachedDocs{cache: nil},
 				cachedMeta: newCachedMeta(),
@@ -465,13 +473,17 @@ func (s *Scorch) introduceMerge(nextMerge *segmentMerge) {
 			case segment.PersistedSegment:
 				fileSegments++
 			default:
-				docsToPersistCount += newMergedSegment.Count() - newSegmentDeleted[i].GetCardinality()
+				docsToPersistCount += newMergedSegment.Count() - newSegmentsDeleted[i].GetCardinality()
 				memSegments++
 			}
 			atomic.AddUint64(&s.stats.TotIntroducedSegmentsMerge, 1)
 			skipped[i] = false
 		} else {
-			atomic.AddUint64(&s.stats.TotFileMergeIntroductionsObsoleted, 1)
+			if nextMerge.fileMerge {
+				atomic.AddUint64(&s.stats.TotFileMergeIntroductionsObsoleted, 1)
+			} else {
+				atomic.AddUint64(&s.stats.TotMemMergeIntroductionsObsoleted, 1)
+			}
 			skipped[i] = true
 		}
 	}

index/scorch/merge.go

@@ -185,13 +185,13 @@ func (s *Scorch) ForceMerge(ctx context.Context,
 	mo *mergeplan.MergePlanOptions) error {
 	// check whether force merge is already under processing
 	s.rootLock.Lock()
-	if s.stats.TotFileMergeForceOpsStarted >
-		s.stats.TotFileMergeForceOpsCompleted {
+	if atomic.LoadUint64(&s.stats.TotFileMergeForceOpsStarted) >
+		atomic.LoadUint64(&s.stats.TotFileMergeForceOpsCompleted) {
 		s.rootLock.Unlock()
 		return fmt.Errorf("force merge already in progress")
 	}
 
-	s.stats.TotFileMergeForceOpsStarted++
+	atomic.AddUint64(&s.stats.TotFileMergeForceOpsStarted, 1)
 	s.rootLock.Unlock()
 
 	if mo != nil {
@@ -296,6 +296,7 @@ type mergeBatch struct {
 	newID       uint64
 	newDocNums  [][]uint64
 	newFilename string
+	newTime     uint64
 }
 
 // planMergeAtSnapshot plans and executes the merge operations for a given snapshot
@@ -397,6 +398,7 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 			newID:       newSegmentID,
 			newDocNums:  nil,
 			newFilename: zapFileName(newSegmentID),
+			newTime:     0,
 		}
 		mergeBatches[batchID] = batch
 
@@ -428,17 +430,16 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 		s.markIneligibleForRemoval(batch.newFilename)
 		path := s.path + string(os.PathSeparator) + batch.newFilename
 
+		prevBytesReadTotal := cumulateBytesRead(batch.segments)
 		fileMergeZapStartTime := time.Now()
 		atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
-		prevBytesReadTotal := cumulateBytesRead(batch.segments)
 		batch.newDocNums, _, err = s.segPlugin.MergeUsing(batch.segments, batch.drops, path,
 			cw.cancelCh, s, s.segmentConfig)
 		atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
-
-		fileMergeZapTime := uint64(time.Since(fileMergeZapStartTime))
-		atomic.AddUint64(&s.stats.TotFileMergeZapTime, fileMergeZapTime)
-		if atomic.LoadUint64(&s.stats.MaxFileMergeZapTime) < fileMergeZapTime {
-			atomic.StoreUint64(&s.stats.MaxFileMergeZapTime, fileMergeZapTime)
+		batch.newTime = uint64(time.Since(fileMergeZapStartTime))
+		atomic.AddUint64(&s.stats.TotFileMergeZapTime, batch.newTime)
+		if atomic.LoadUint64(&s.stats.MaxFileMergeZapTime) < batch.newTime {
+			atomic.StoreUint64(&s.stats.MaxFileMergeZapTime, batch.newTime)
 		}
 		if err != nil {
 			atomic.AddUint64(&s.stats.TotFileMergePlanTasksErr, 1)
@@ -473,11 +474,12 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 	}
 
 	sm := &segmentMerge{
-		id:               newSegmentIDs,
-		new:              newSegments,
+		newSegmentIDs:    newSegmentIDs,
+		newSegments:      newSegments,
 		mergedSegHistory: mergedSegHistory,
 		notifyCh:         make(chan *mergeTaskIntroStatus),
 		mmaped:           1,
+		fileMerge:        true,
 	}
 
 	s.fireEvent(EventKindMergeTaskIntroductionStart, 0)
@@ -532,11 +534,12 @@ type mergedSegmentHistory struct {
 }
 
 type segmentMerge struct {
-	id               []uint64
-	new              []segment.Segment
+	newSegmentIDs    []uint64
+	newSegments      []segment.Segment
 	mergedSegHistory map[uint64]*mergedSegmentHistory
 	notifyCh         chan *mergeTaskIntroStatus
 	mmaped           uint32
+	fileMerge        bool
 }
 
 func cumulateBytesRead(sbs []segment.Segment) uint64 {
@@ -577,6 +580,7 @@ func (s *Scorch) mergeAndPersistInMemorySegments(flushes []*flushable, po *persi
 			var err error
 			defer func() {
 				if err != nil {
+					atomic.AddUint64(&s.stats.TotMemMergeErr, 1)
 					errM.Lock()
 					errs = append(errs, err)
 					errM.Unlock()
@@ -597,10 +601,10 @@ func (s *Scorch) mergeAndPersistInMemorySegments(flushes []*flushable, po *persi
 				newID:       newSegmentID,
 				newDocNums:  nil,
 				newFilename: zapFileName(newSegmentID),
+				newTime:     0,
 			}
 			mergeBatches[batchID] = batch
 
-			atomic.AddUint64(&numMergedSegments, uint64(len(flush.sbsBatch)))
 			s.markIneligibleForRemoval(batch.newFilename)
 			path := s.path + string(os.PathSeparator) + batch.newFilename
 
@@ -609,26 +613,32 @@ func (s *Scorch) mergeAndPersistInMemorySegments(flushes []*flushable, po *persi
 			batch.newDocNums, _, err = s.segPlugin.MergeUsing(batch.segments, batch.drops, path,
 				s.closeCh, s, s.segmentConfig)
 			atomic.AddUint64(&s.stats.TotMemMergeZapEnd, 1)
-			memMergeZapTime := uint64(time.Since(memMergeZapStartTime))
-			atomic.AddUint64(&s.stats.TotMemMergeZapTime, memMergeZapTime)
-			if atomic.LoadUint64(&s.stats.MaxMemMergeZapTime) < memMergeZapTime {
-				atomic.StoreUint64(&s.stats.MaxMemMergeZapTime, memMergeZapTime)
-			}
+			batch.newTime = uint64(time.Since(memMergeZapStartTime))
+			atomic.AddUint64(&s.stats.TotMemMergeZapTime, batch.newTime)
 			if err != nil {
-				atomic.AddUint64(&s.stats.TotMemMergeErr, 1)
 				return
 			}
-
 			batch.new, err = s.segPlugin.OpenUsing(path, s.segmentConfig)
 			if err != nil {
-				atomic.AddUint64(&s.stats.TotMemMergeErr, 1)
 				return
 			}
+			atomic.AddUint64(&numMergedSegments, uint64(len(batch.segments)))
 		}(batchID)
 	}
 	wg.Wait()
 	close(sem)
 
+	var maxZapTime uint64
+	for _, batch := range mergeBatches {
+		if batch.newTime > maxZapTime {
+			maxZapTime = batch.newTime
+		}
+	}
+	if maxZapTime > atomic.LoadUint64(&s.stats.MaxMemMergeZapTime) {
+		atomic.StoreUint64(&s.stats.MaxMemMergeZapTime, maxZapTime)
+	}
+	atomic.AddUint64(&s.stats.TotMemMergeSegments, numMergedSegments)
+
 	if len(errs) > 0 {
 		var errf error
 		numClosed := 0
@@ -666,22 +676,31 @@ func (s *Scorch) mergeAndPersistInMemorySegments(flushes []*flushable, po *persi
 	}
 
 	sm := &segmentMerge{
-		id:               newSegmentIDs,
-		new:              newSegments,
+		newSegmentIDs:    newSegmentIDs,
+		newSegments:      newSegments,
 		mergedSegHistory: mergedSegHistory,
 		notifyCh:         make(chan *mergeTaskIntroStatus),
 		mmaped:           1,
+		fileMerge:        false,
 	}
 
 	select {
 	case <-s.closeCh:
 		err = segment.ErrClosed
 		return nil, nil, err
 	case s.merges <- sm:
+		atomic.AddUint64(&s.stats.TotMemMergeIntroductions, uint64(numBatches))
 	}
 
+	introStartTime := time.Now()
 	// blockingly wait for the introduction to complete
 	introStatus := <-sm.notifyCh
+	introTime := uint64(time.Since(introStartTime))
+	atomic.AddUint64(&s.stats.TotMemMergeZapIntroductionTime, introTime)
+	if atomic.LoadUint64(&s.stats.MaxMemMergeZapIntroductionTime) < introTime {
+		atomic.StoreUint64(&s.stats.MaxMemMergeZapIntroductionTime, introTime)
+	}
+	atomic.AddUint64(&s.stats.TotMemMergeIntroductionsDone, uint64(numBatches))
 	introducedSnapshot := introStatus.indexSnapshot
 	introducedSegmentIDs := make(map[uint64]struct{}, numBatches)
 	for batchID, skipped := range introStatus.skipped {
@@ -703,8 +722,6 @@ func (s *Scorch) mergeAndPersistInMemorySegments(flushes []*flushable, po *persi
 		return nil, nil, nil
 	}
 
-	atomic.AddUint64(&s.stats.TotMemMergeSegments, numMergedSegments)
-
 	return introducedSnapshot, introducedSegmentIDs, nil
 }
 

index/scorch/stats.go

@@ -64,8 +64,6 @@ type Stats struct {
 	TotIntroducePersistEnd uint64
 	TotIntroduceMergeBeg   uint64
 	TotIntroduceMergeEnd   uint64
-	TotIntroduceRevertBeg  uint64
-	TotIntroduceRevertEnd  uint64
 
 	TotIntroducedItems         uint64
 	TotIntroducedSegmentsBatch uint64
@@ -127,14 +125,23 @@ type Stats struct {
 	CurFilesIneligibleForRemoval     uint64
 	TotSnapshotsRemovedFromMetaStore uint64
 
-	TotMemMergeBeg          uint64
-	TotMemMergeErr          uint64
-	TotMemMergeDone         uint64
-	TotMemMergeZapBeg       uint64
-	TotMemMergeZapEnd       uint64
-	TotMemMergeZapTime      uint64
-	MaxMemMergeZapTime      uint64
-	TotMemMergeSegments     uint64
+	TotMemMergeBeg      uint64
+	TotMemMergeErr      uint64
+	TotMemMergeDone     uint64
+	TotMemMergeSegments uint64
+
+	TotMemMergeZapBeg              uint64
+	TotMemMergeZapEnd              uint64
+	TotMemMergeZapTime             uint64
+	MaxMemMergeZapTime             uint64
+	TotMemMergeZapIntroductionTime uint64
+	MaxMemMergeZapIntroductionTime uint64
+
+	TotMemMergeIntroductions          uint64
+	TotMemMergeIntroductionsDone      uint64
+	TotMemMergeIntroductionsSkipped   uint64
+	TotMemMergeIntroductionsObsoleted uint64
+
 	TotMemorySegmentsAtRoot uint64
 
 	TotTrainedSamples uint64