adding compaction_written_bytes/sec stats to scorch

index/scorch/merge.go

@@ -180,7 +180,7 @@ func (s *Scorch) planMergeAtSnapshot(ourSnapshot *IndexSnapshot,
 			s.markIneligibleForRemoval(filename)
 			path := s.path + string(os.PathSeparator) + filename
 			atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1)
-			newDocNums, err := zap.Merge(segmentsToMerge, docsToDrop, path, 1024)
+			newDocNums, err := zap.Merge(segmentsToMerge, docsToDrop, path, 1024, &s.stats)
 			atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1)
 			if err != nil {
 				s.unmarkIneligibleForRemoval(filename)

index/scorch/scorch.go

@@ -430,6 +430,7 @@ func (s *Scorch) StatsMap() map[string]interface{} {
 	m["num_items_persisted"] = m["TotPersistedItems"]
 	m["num_bytes_used_disk"] = m["CurOnDiskBytes"]
 	m["num_files_on_disk"] = m["CurOnDiskFiles"]
+	m["total_compaction_written_bytes"] = m["TotCompactionWrittenBytes"]
 
 	return m
 }

index/scorch/segment/zap/merge.go

@@ -31,12 +31,17 @@ import (
 
 const docDropped = math.MaxUint64 // sentinel docNum to represent a deleted doc
 
+// StatsReporter interface represents stats reporting methods.
+type StatsReporter interface {
+	ReportBytesWritten(numBytesWritten uint64)
+}
+
 // Merge takes a slice of zap segments and bit masks describing which
 // documents may be dropped, and creates a new segment containing the
 // remaining data.  This new segment is built at the specified path,
 // with the provided chunkFactor.
 func Merge(segments []*Segment, drops []*roaring.Bitmap, path string,
-	chunkFactor uint32) ([][]uint64, error) {
+	chunkFactor uint32, stats StatsReporter) ([][]uint64, error) {
 	flag := os.O_RDWR | os.O_CREATE
 
 	f, err := os.OpenFile(path, flag, 0600)
@@ -92,6 +97,8 @@ func Merge(segments []*Segment, drops []*roaring.Bitmap, path string,
 		return nil, err
 	}
 
+	stats.ReportBytesWritten(uint64(cr.Count()))
+
 	return newDocNums, nil
 }
 

index/scorch/stats.go

@@ -100,6 +100,8 @@ type Stats struct {
 	TotMemMergeZapBeg   uint64
 	TotMemMergeZapEnd   uint64
 	TotMemMergeSegments uint64
+
+	TotCompactionWrittenBytes uint64
 }
 
 // atomically populates the returned map
@@ -122,3 +124,7 @@ func (s *Stats) ToMap() map[string]interface{} {
 func (s *Stats) MarshalJSON() ([]byte, error) {
 	return json.Marshal(s.ToMap())
 }
+
+func (s *Stats) ReportBytesWritten(numBytesWritten uint64) {
+	atomic.AddUint64(&s.TotCompactionWrittenBytes, numBytesWritten)
+}