Only read the commit log once during bootstrapping

src/dbnode/integration/commitlog_bootstrap_helpers.go

@@ -106,8 +106,8 @@ func writeCommitLogData(
 	data generate.SeriesBlocksByStart,
 	namespace namespace.Metadata,
 	genSnapshots bool,
-) {
-	writeCommitLogDataBase(t, s, opts, data, namespace, nil, nil)
+) int {
+	return writeCommitLogDataBase(t, s, opts, data, namespace, nil, nil)
 }
 
 func writeCommitLogDataSpecifiedTS(
@@ -118,8 +118,8 @@ func writeCommitLogDataSpecifiedTS(
 	namespace namespace.Metadata,
 	ts time.Time,
 	genSnapshots bool,
-) {
-	writeCommitLogDataBase(t, s, opts, data, namespace, &ts, nil)
+) int {
+	return writeCommitLogDataBase(t, s, opts, data, namespace, &ts, nil)
 }
 
 func writeCommitLogDataWithPredicate(
@@ -129,10 +129,11 @@ func writeCommitLogDataWithPredicate(
 	data generate.SeriesBlocksByStart,
 	namespace namespace.Metadata,
 	pred generate.WriteDatapointPredicate,
-) {
-	writeCommitLogDataBase(t, s, opts, data, namespace, nil, pred)
+) int {
+	return writeCommitLogDataBase(t, s, opts, data, namespace, nil, pred)
 }
 
+// returns the number of data points written to the commit log
 func writeCommitLogDataBase(
 	t *testing.T,
 	s TestSetup,
@@ -141,7 +142,7 @@ func writeCommitLogDataBase(
 	namespace namespace.Metadata,
 	specifiedTS *time.Time,
 	pred generate.WriteDatapointPredicate,
-) {
+) int {
 	if pred == nil {
 		pred = generate.WriteAllPredicate
 	}
@@ -155,6 +156,7 @@ func writeCommitLogDataBase(
 		shardSet       = s.ShardSet()
 		tagEncoderPool = opts.FilesystemOptions().TagEncoderPool()
 		tagSliceIter   = ident.NewTagsIterator(ident.Tags{})
+		writes int
 	)
 
 	// Write out commit log data.
@@ -203,12 +205,14 @@ func writeCommitLogDataBase(
 			}
 			if pred(point.Value) {
 				require.NoError(t, commitLog.Write(ctx, cID, point.Value.Datapoint, xtime.Second, point.Value.Annotation))
+				writes++
 			}
 		}
 
 		// ensure writes finished.
 		require.NoError(t, commitLog.Close())
 	}
+	return writes
 }
 
 func writeSnapshotsWithPredicate(

src/dbnode/integration/commitlog_bootstrap_test.go

@@ -23,6 +23,7 @@
 package integration
 
 import (
+	"github.com/uber-go/tally"
 	"testing"
 	"time"
 
@@ -79,7 +80,7 @@ func testCommitLogBootstrap(t *testing.T, setTestOpts setTestOptions, updateInpu
 	now := setup.NowFn()()
 	seriesMaps := generateSeriesMaps(30, updateInputConfig, now.Add(-2*blockSize), now.Add(-blockSize))
 	log.Info("writing data")
-	writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
+	writes := writeCommitLogData(t, setup, commitLogOpts, seriesMaps, ns1, false)
 	log.Info("finished writing data")
 
 	// Setup bootstrapper after writing data so filesystem inspection can find it.
@@ -109,4 +110,6 @@ func testCommitLogBootstrap(t *testing.T, setTestOpts setTestOptions, updateInpu
 	observedSeriesMaps2 := testSetupToSeriesMaps(t, setup, ns2, metadatasByShard2)
 	verifySeriesMapsEqual(t, emptySeriesMaps, observedSeriesMaps2)
 
+	counters := commitLogOpts.InstrumentOptions().MetricsScope().(tally.TestScope).Snapshot().Counters()
+	require.Equal(t, writes, int(counters["bootstrapper-commitlog.commitlog.entries-read+"].Value()))
 }

src/dbnode/storage/bootstrap/bootstrapper/commitlog/source.go

@@ -419,6 +419,7 @@ func (s *commitLogSource) readCommitLog(namespaces bootstrap.Namespaces, span op
 	// to read.
 	var lastFileReadID uint64
 	for iter.Next() {
+		s.metrics.commitLogEntriesRead.Inc(1)
 		entry := iter.Current()
 
 		currFileReadID := entry.Metadata.FileReadID
@@ -1112,12 +1113,14 @@ func (s *commitLogSource) shardsReplicated(
 type commitLogSourceMetrics struct {
 	corruptCommitlogFile tally.Counter
 	bootstrapping        tally.Gauge
+	commitLogEntriesRead tally.Counter
 }
 
 func newCommitLogSourceMetrics(scope tally.Scope) commitLogSourceMetrics {
 	return commitLogSourceMetrics{
 		corruptCommitlogFile: scope.SubScope("commitlog").Counter("corrupt"),
 		bootstrapping:        scope.SubScope("status").Gauge("bootstrapping"),
+		commitLogEntriesRead: scope.SubScope("commitlog").Counter("entries-read"),
 	}
 }