Improve commit log reader speed

glide.yaml

@@ -157,6 +157,9 @@ import:
   - package: github.com/hydrogen18/stalecucumber
     version: 9b38526d4bdf8e197c31344777fc28f7f48d250d
 
+  - package: github.com/c2h5oh/datasize
+    version: 4eba002a5eaea69cf8d235a388fc6b65ae68d2dd
+
   # START_PROMETHEUS_DEPS
   - package: github.com/prometheus/prometheus
     version: 998dfcbac689ae832ea64ca134fcb096f61a7f62

src/dbnode/integration/commitlog_bootstrap_index_perf_speed_test.go

@@ -0,0 +1,221 @@
+// +build integration
+
+// Copyright (c) 2018 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package integration
+
+import (
+	"encoding/binary"
+	"encoding/hex"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"os"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/m3db/m3/src/dbnode/clock"
+	"github.com/m3db/m3/src/dbnode/namespace"
+	"github.com/m3db/m3/src/dbnode/persist/fs"
+	"github.com/m3db/m3/src/dbnode/persist/fs/commitlog"
+	"github.com/m3db/m3/src/dbnode/retention"
+	"github.com/m3db/m3/src/dbnode/ts"
+	"github.com/m3db/m3/src/x/checked"
+	"github.com/m3db/m3/src/x/context"
+	"github.com/m3db/m3/src/x/ident"
+	xtime "github.com/m3db/m3/src/x/time"
+
+	"github.com/c2h5oh/datasize"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+// TestCommitLogIndexPerfSpeedBootstrap tests the performance of the commit log
+// bootstrapper when a large amount of data is present that needs to be read.
+// Note: Also useful for doing adhoc testing, using the environment variables
+// to turn up and down the number of IDs and points.
+func TestCommitLogIndexPerfSpeedBootstrap(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow() // Just skip if we're doing a short run
+	}
+
+	// Test setup
+	var (
+		rOpts     = retention.NewOptions().SetRetentionPeriod(12 * time.Hour)
+		blockSize = rOpts.BlockSize()
+	)
+
+	nsOpts := namespace.NewOptions().
+		SetRetentionOptions(rOpts).
+		SetIndexOptions(namespace.NewIndexOptions().
+			SetEnabled(true).
+			SetBlockSize(2 * blockSize))
+	ns, err := namespace.NewMetadata(testNamespaces[0], nsOpts)
+	require.NoError(t, err)
+	opts := newTestOptions(t).
+		SetNamespaces([]namespace.Metadata{ns}).
+		// Allow for wall clock timing
+		SetNowFn(time.Now)
+
+	setup, err := newTestSetup(t, opts, nil)
+	require.NoError(t, err)
+	defer setup.close()
+
+	commitLogOpts := setup.storageOpts.CommitLogOptions().
+		SetFlushInterval(defaultIntegrationTestFlushInterval)
+	setup.storageOpts = setup.storageOpts.SetCommitLogOptions(commitLogOpts)
+
+	log := setup.storageOpts.InstrumentOptions().Logger()
+	log.Info("commit log bootstrap test")
+
+	// Write test data
+	log.Info("generating data")
+
+	// NB(r): Use TEST_NUM_SERIES=50000 for a representative large data set to
+	// test loading locally
+	numSeries := 1024
+	if str := os.Getenv("TEST_NUM_SERIES"); str != "" {
+		numSeries, err = strconv.Atoi(str)
+		require.NoError(t, err)
+	}
+
+	step := time.Second
+	numPoints := 128
+	if str := os.Getenv("TEST_NUM_POINTS"); str != "" {
+		numPoints, err = strconv.Atoi(str)
+		require.NoError(t, err)
+	}
+
+	require.True(t, (time.Duration(numPoints)*step) < blockSize,
+		fmt.Sprintf("num points %d multiplied by step %s is greater than block size %s",
+			numPoints, step.String(), blockSize.String()))
+
+	numTags := 8
+	if str := os.Getenv("TEST_NUM_TAGS"); str != "" {
+		numTags, err = strconv.Atoi(str)
+		require.NoError(t, err)
+	}
+
+	numTagSets := 128
+	if str := os.Getenv("TEST_NUM_TAG_SETS"); str != "" {
+		numTagSets, err = strconv.Atoi(str)
+		require.NoError(t, err)
+	}
+
+	// Pre-generate tag sets, but not too many to reduce heap size.
+	tagSets := make([]ident.Tags, 0, numTagSets)
+	for i := 0; i < numTagSets; i++ {
+		tags := ident.NewTags()
+		for j := 0; j < numTags; j++ {
+			tag := ident.Tag{
+				Name:  ident.StringID(fmt.Sprintf("series.%d.tag.%d", i, j)),
+				Value: ident.StringID(fmt.Sprintf("series.%d.tag-value.%d", i, j)),
+			}
+			tags.Append(tag)
+		}
+		tagSets = append(tagSets, tags)
+	}
+
+	log.Info("writing data")
+
+	now := setup.getNowFn()
+	blockStart := now.Add(-3 * blockSize)
+
+	// create new commit log
+	commitLog, err := commitlog.NewCommitLog(commitLogOpts)
+	require.NoError(t, err)
+	require.NoError(t, commitLog.Open())
+
+	// NB(r): Write points using no up front series metadata or point
+	// generation so that the memory usage is constant during the write phase
+	ctx := context.NewContext()
+	defer ctx.Close()
+	shardSet := setup.shardSet
+	idPrefix := "test.id.test.id.test.id.test.id.test.id.test.id.test.id.test.id"
+	idPrefixBytes := []byte(idPrefix)
+	checkedBytes := checked.NewBytes(nil, nil)
+	seriesID := ident.BinaryID(checkedBytes)
+	numBytes := make([]byte, 8)
+	numHexBytes := make([]byte, hex.EncodedLen(len(numBytes)))
+	for i := 0; i < numPoints; i++ {
+		for j := 0; j < numSeries; j++ {
+			// Write the ID prefix
+			checkedBytes.Resize(0)
+			checkedBytes.AppendAll(idPrefixBytes)
+
+			// Write out the binary representation then hex encode the
+			// that into the ID to give it a unique ID for this series number
+			binary.LittleEndian.PutUint64(numBytes, uint64(j))
+			hex.Encode(numHexBytes, numBytes)
+			checkedBytes.AppendAll(numHexBytes)
+
+			// Use the tag sets appropriate for this series number
+			seriesTags := tagSets[j%len(tagSets)]
+
+			series := ts.Series{
+				Namespace:   ns.ID(),
+				Shard:       shardSet.Lookup(seriesID),
+				ID:          seriesID,
+				Tags:        seriesTags,
+				UniqueIndex: uint64(j),
+			}
+			dp := ts.Datapoint{
+				Timestamp: blockStart.Add(time.Duration(i) * step),
+				Value:     rand.Float64(),
+			}
+			require.NoError(t, commitLog.Write(ctx, series, dp, xtime.Second, nil))
+		}
+	}
+
+	// ensure writes finished
+	require.NoError(t, commitLog.Close())
+
+	log.Info("finished writing data")
+
+	// emit how big commit logs are
+	commitLogsDirPath := fs.CommitLogsDirPath(commitLogOpts.FilesystemOptions().FilePathPrefix())
+	files, err := ioutil.ReadDir(commitLogsDirPath)
+	require.NoError(t, err)
+
+	log.Info("test wrote commit logs", zap.Int("numFiles", len(files)))
+	for _, file := range files {
+		log.Info("test wrote commit logs",
+			zap.String("file", file.Name()),
+			zap.String("size", datasize.ByteSize(file.Size()).HR()))
+	}
+
+	// Setup bootstrapper after writing data so filesystem inspection can find it.
+	setupCommitLogBootstrapperWithFSInspection(t, setup, commitLogOpts)
+
+	// restore now time so measurements take effect
+	setup.storageOpts = setup.storageOpts.SetClockOptions(clock.NewOptions())
+
+	// Start the server with filesystem bootstrapper
+	require.NoError(t, setup.startServer())
+	log.Debug("server is now up")
+
+	// Stop the server
+	defer func() {
+		require.NoError(t, setup.stopServer())
+		log.Debug("server is now down")
+	}()
+}

src/dbnode/integration/fs_commitlog_mixed_mode_read_write_test.go

@@ -28,21 +28,21 @@ import (
 	"time"
 
 	"github.com/m3db/m3/src/dbnode/integration/generate"
+	"github.com/m3db/m3/src/dbnode/namespace"
 	persistfs "github.com/m3db/m3/src/dbnode/persist/fs"
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/dbnode/runtime"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper"
 	bcl "github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/commitlog"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper/fs"
-	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
 	xtime "github.com/m3db/m3/src/x/time"
 
-	"github.com/stretchr/testify/require"
 	"github.com/m3db/m3/src/dbnode/testdata/prototest"
+	"github.com/stretchr/testify/require"
 )
 
 type annotationGenerator interface {
@@ -87,6 +87,8 @@ func testFsCommitLogMixedModeReadWrite(t *testing.T, setTestOpts setTestOptions,
 	log := setup.storageOpts.InstrumentOptions().Logger()
 	log.Info("commit log & fileset files, write, read, and merge bootstrap test")
 
+	filePathPrefix := setup.storageOpts.CommitLogOptions().FilesystemOptions().FilePathPrefix()
+
 	// setting time to 2017/02/13 15:30:10
 	fakeStart := time.Date(2017, time.February, 13, 15, 30, 10, 0, time.Local)
 	blkStart15 := fakeStart.Truncate(ns1BlockSize)
@@ -137,7 +139,7 @@ func testFsCommitLogMixedModeReadWrite(t *testing.T, setTestOpts setTestOptions,
 	expectedFlushedData := datapoints.toSeriesMap(ns1BlockSize)
 	delete(expectedFlushedData, xtime.ToUnixNano(blkStart18))
 	waitTimeout := 5 * time.Minute
-	filePathPrefix := setup.storageOpts.CommitLogOptions().FilesystemOptions().FilePathPrefix()
+
 	log.Info("waiting till expected fileset files have been written")
 	require.NoError(t, waitUntilDataFilesFlushed(filePathPrefix, setup.shardSet, nsID, expectedFlushedData, waitTimeout))
 	log.Info("expected fileset files have been written")

src/dbnode/integration/options.go

@@ -24,9 +24,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/dbnode/storage/block"
-	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/topology"
 
 	"github.com/stretchr/testify/require"
@@ -256,7 +256,7 @@ type testOptions interface {
 	FilePathPrefix() string
 
 	// SetProtoEncoding turns on proto encoder.
-	SetProtoEncoding (value bool) testOptions
+	SetProtoEncoding(value bool) testOptions
 
 	// ProtoEncoding returns whether proto encoder is turned on.
 	ProtoEncoding() bool
@@ -267,6 +267,12 @@ type testOptions interface {
 
 	// AssertTestDataEqual returns a comparator to compare two byte arrays.
 	AssertTestDataEqual() assertTestDataEqual
+
+	// SetNowFn will set the now fn.
+	SetNowFn(value func() time.Time) testOptions
+
+	// NowFn returns the now fn.
+	NowFn() func() time.Time
 }
 
 type options struct {
@@ -298,6 +304,7 @@ type options struct {
 	writeNewSeriesAsync                bool
 	protoEncoding                      bool
 	assertEqual                        assertTestDataEqual
+	nowFn                              func() time.Time
 }
 
 func newTestOptions(t *testing.T) testOptions {
@@ -613,3 +620,13 @@ func (o *options) SetAssertTestDataEqual(value assertTestDataEqual) testOptions
 func (o *options) AssertTestDataEqual() assertTestDataEqual {
 	return o.assertEqual
 }
+
+func (o *options) SetNowFn(value func() time.Time) testOptions {
+	opts := *o
+	opts.nowFn = value
+	return &opts
+}
+
+func (o *options) NowFn() func() time.Time {
+	return o.nowFn
+}

src/dbnode/integration/setup.go

@@ -39,6 +39,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/generated/thrift/rpc"
 	"github.com/m3db/m3/src/dbnode/integration/fake"
 	"github.com/m3db/m3/src/dbnode/integration/generate"
+	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/persist/fs"
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/dbnode/sharding"
@@ -47,7 +48,6 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/bootstrapper"
 	"github.com/m3db/m3/src/dbnode/storage/cluster"
 	"github.com/m3db/m3/src/dbnode/storage/index"
-	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/storage/series"
 	"github.com/m3db/m3/src/dbnode/testdata/prototest"
 	"github.com/m3db/m3/src/dbnode/topology"
@@ -82,7 +82,7 @@ var (
 	testProtoEqual    = func(expect, actual []byte) bool {
 		return prototest.ProtoEqual(testSchema, expect, actual)
 	}
-	testProtoIter    = prototest.NewProtoMessageIterator(testProtoMessages)
+	testProtoIter = prototest.NewProtoMessageIterator(testProtoMessages)
 )
 
 // nowSetterFn is the function that sets the current time
@@ -293,8 +293,14 @@ func newTestSetup(t *testing.T, opts testOptions, fsOpts fs.Options) (*testSetup
 		now = t
 		lock.Unlock()
 	}
-	storageOpts = storageOpts.SetClockOptions(
-		storageOpts.ClockOptions().SetNowFn(getNowFn))
+	if overrideTimeNow := opts.NowFn(); overrideTimeNow != nil {
+		// Allow overriding the frozen time
+		storageOpts = storageOpts.SetClockOptions(
+			storageOpts.ClockOptions().SetNowFn(overrideTimeNow))
+	} else {
+		storageOpts = storageOpts.SetClockOptions(
+			storageOpts.ClockOptions().SetNowFn(getNowFn))
+	}
 
 	// Set up file path prefix
 	idx := atomic.AddUint64(&created, 1) - 1