[dbnode] Use already encoded tags when writing time series to commit log

glide.yaml

@@ -28,7 +28,7 @@ import:
     version: ^0.8
 
   - package: github.com/apache/thrift
-    version: 0.9.3-pool-read-binary-2
+    version: 0.9.3-pool-read-binary-3
     subpackages:
       - lib/go/thrift
     repo: https://github.com/m3db/thrift

src/cmd/services/m3dbnode/config/config_test.go

@@ -133,6 +133,7 @@ db:
 
   pooling:
       blockAllocSize: 16
+      thriftBytesPoolMaxAllocSize: 2048
       type: simple
       seriesPool:
           size: 5242880
@@ -445,6 +446,7 @@ func TestConfiguration(t *testing.T) {
   replication: null
   pooling:
     blockAllocSize: 16
+    thriftBytesPoolMaxAllocSize: 2048
     type: simple
     bytesPool:
       buckets:

src/cmd/services/m3dbnode/config/pooling.go

@@ -33,8 +33,9 @@ const (
 )
 
 const (
-	defaultMaxFinalizerCapacity = 4
-	defaultBlockAllocSize       = 16
+	defaultMaxFinalizerCapacity        = 4
+	defaultBlockAllocSize              = 16
+	defaultThriftBytesPoolMaxAllocSize = 1024
 )
 
 type poolPolicyDefault struct {
@@ -252,6 +253,9 @@ type PoolingPolicy struct {
 	// The initial alloc size for a block.
 	BlockAllocSize *int `yaml:"blockAllocSize"`
 
+	// The thrift bytes pool max bytes slice allocation for a single binary field.
+	ThriftBytesPoolMaxAllocSize *int `yaml:"thriftBytesPoolMaxAllocSize"`
+
 	// The general pool type (currently only supported: simple).
 	Type *PoolingType `yaml:"type"`
 
@@ -418,6 +422,16 @@ func (p *PoolingPolicy) BlockAllocSizeOrDefault() int {
 	return defaultBlockAllocSize
 }
 
+// ThriftBytesPoolMaxAllocSizeOrDefault returns the configured thrift bytes pool
+// max alloc size if provided, or a default value otherwise.
+func (p *PoolingPolicy) ThriftBytesPoolMaxAllocSizeOrDefault() int {
+	if p.ThriftBytesPoolMaxAllocSize != nil {
+		return *p.ThriftBytesPoolMaxAllocSize
+	}
+
+	return defaultThriftBytesPoolMaxAllocSize
+}
+
 // TypeOrDefault returns the configured pooling type if provided, or a default
 // value otherwise.
 func (p *PoolingPolicy) TypeOrDefault() PoolingType {

src/cmd/services/m3dbnode/config/pooling_test.go

@@ -51,3 +51,13 @@ func TestContextPoolMaxFinalizerCapacityOrDefault(t *testing.T) {
 	cpp.MaxFinalizerCapacity = 10
 	require.Equal(t, 10, cpp.MaxFinalizerCapacityOrDefault())
 }
+
+func TestPoolingPolicyThriftBytesPoolMaxAllocSizeOrDefault(t *testing.T) {
+	policy := PoolingPolicy{}
+	require.Equal(t, defaultThriftBytesPoolMaxAllocSize,
+		policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+
+	value := 42
+	policy.ThriftBytesPoolMaxAllocSize = &value
+	require.Equal(t, 42, policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+}

src/dbnode/network/server/tchannelthrift/node/service.go

@@ -1163,6 +1163,7 @@ func (s *service) WriteBatchRaw(tctx thrift.Context, req *rpc.WriteBatchRawReque
 	if err != nil {
 		return convert.ToRPCError(err)
 	}
+
 	// The lifecycle of the annotations is more involved than the rest of the data
 	// so we set the annotation pool put method as the finalization function and
 	// let the database take care of returning them to the pool.
@@ -1247,9 +1248,12 @@ func (s *service) WriteTaggedBatchRaw(tctx thrift.Context, req *rpc.WriteTaggedB
 	if err != nil {
 		return convert.ToRPCError(err)
 	}
-	// The lifecycle of the annotations is more involved than the rest of the data
-	// so we set the annotation pool put method as the finalization function and
-	// let the database take care of returning them to the pool.
+
+	// The lifecycle of the encoded tags and annotations is more involved than
+	// the rest of the data so we set the encoded tags and annotation pool put
+	// calls as finalization functions and let the database take care of
+	// returning them to the pool.
+	batchWriter.SetFinalizeEncodedTagsFn(finalizeEncodedTagsFn)
 	batchWriter.SetFinalizeAnnotationFn(finalizeAnnotationFn)
 
 	for i, elem := range req.Elements {
@@ -1279,6 +1283,7 @@ func (s *service) WriteTaggedBatchRaw(tctx thrift.Context, req *rpc.WriteTaggedB
 			i,
 			seriesID,
 			dec,
+			elem.EncodedTags,
 			xtime.FromNormalizedTime(elem.Datapoint.Timestamp, d),
 			elem.Datapoint.Value,
 			unit,
@@ -1764,7 +1769,11 @@ func (r *writeBatchPooledReq) Finalize() {
 	if r.writeTaggedReq != nil {
 		for _, elem := range r.writeTaggedReq.Elements {
 			apachethrift.BytesPoolPut(elem.ID)
-			apachethrift.BytesPoolPut(elem.EncodedTags)
+			// Ownership of the encoded tagts has been transferred to the BatchWriter
+			// so they will get returned the pool automatically by the commitlog once
+			// it finishes writing them to disk via the finalization function that
+			// gets set on the WriteBatch.
+
 			// See comment above about not finalizing annotations here.
 		}
 		r.writeTaggedReq = nil
@@ -1874,6 +1883,13 @@ func (p *writeBatchPooledReqPool) Put(v *writeBatchPooledReq) {
 	p.pool.Put(v)
 }
 
+// finalizeEncodedTagsFn implements ts.FinalizeEncodedTagsFn because
+// apachethrift.BytesPoolPut(b) returns a bool but ts.FinalizeEncodedTagsFn
+// does not.
+func finalizeEncodedTagsFn(b []byte) {
+	apachethrift.BytesPoolPut(b)
+}
+
 // finalizeAnnotationFn implements ts.FinalizeAnnotationFn because
 // apachethrift.BytesPoolPut(b) returns a bool but ts.FinalizeAnnotationFn
 // does not.

src/dbnode/persist/fs/commitlog/read_write_prop_test.go

@@ -33,6 +33,9 @@ import (
 	"testing"
 	"time"
 
+	"github.com/m3db/m3/src/x/pool"
+	"github.com/m3db/m3/src/x/serialize"
+
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
@@ -114,6 +117,7 @@ func TestCommitLogReadWrite(t *testing.T) {
 		write := seriesWrites.writes[seriesWrites.readPosition]
 
 		require.Equal(t, write.series.ID.String(), series.ID.String())
+		require.True(t, write.series.Tags.Equal(series.Tags))
 		require.Equal(t, write.series.Namespace.String(), series.Namespace.String())
 		require.Equal(t, write.series.Shard, series.Shard)
 		require.Equal(t, write.datapoint.Value, datapoint.Value)
@@ -555,22 +559,63 @@ func (w generatedWrite) String() string {
 
 // generator for commit log write
 func genWrite() gopter.Gen {
+	testTagEncodingPool := serialize.NewTagEncoderPool(serialize.NewTagEncoderOptions(),
+		pool.NewObjectPoolOptions().SetSize(1))
+	testTagEncodingPool.Init()
+
 	return gopter.CombineGens(
 		gen.Identifier(),
 		gen.TimeRange(time.Now(), 15*time.Minute),
 		gen.Float64(),
 		gen.Identifier(),
 		gen.UInt32(),
+		gen.Identifier(),
+		gen.Identifier(),
+		gen.Identifier(),
+		gen.Identifier(),
+		gen.Bool(),
 	).Map(func(val []interface{}) generatedWrite {
 		id := val[0].(string)
 		t := val[1].(time.Time)
 		v := val[2].(float64)
 		ns := val[3].(string)
 		shard := val[4].(uint32)
+		tags := map[string]string{
+			val[5].(string): val[6].(string),
+			val[7].(string): val[8].(string),
+		}
+		encodeTags := val[9].(bool)
+
+		var (
+			seriesTags        ident.Tags
+			seriesEncodedTags []byte
+		)
+		for k, v := range tags {
+			seriesTags.Append(ident.Tag{
+				Name:  ident.StringID(k),
+				Value: ident.StringID(v),
+			})
+		}
+
+		if encodeTags {
+			encoder := testTagEncodingPool.Get()
+			if err := encoder.Encode(ident.NewTagsIterator(seriesTags)); err != nil {
+				panic(err)
+			}
+			data, ok := encoder.Data()
+			if !ok {
+				panic("could not encode tags")
+			}
+
+			// Set encoded tags so the "fast" path is activated.
+			seriesEncodedTags = data.Bytes()
+		}
 
 		return generatedWrite{
 			series: ts.Series{
 				ID:          ident.StringID(id),
+				Tags:        seriesTags,
+				EncodedTags: seriesEncodedTags,
 				Namespace:   ident.StringID(ns),
 				Shard:       shard,
 				UniqueIndex: uniqueID(ns, id),

src/dbnode/persist/fs/commitlog/writer.go

@@ -35,9 +35,9 @@ import (
 	"github.com/m3db/m3/src/dbnode/persist/fs/msgpack"
 	"github.com/m3db/m3/src/dbnode/persist/schema"
 	"github.com/m3db/m3/src/dbnode/ts"
-	"github.com/m3db/m3/src/x/os"
-	"github.com/m3db/m3/src/x/serialize"
 	"github.com/m3db/m3/src/x/ident"
+	xos "github.com/m3db/m3/src/x/os"
+	"github.com/m3db/m3/src/x/serialize"
 	xtime "github.com/m3db/m3/src/x/time"
 )
 
@@ -203,13 +203,13 @@ func (w *writer) Write(
 
 	seen := w.seen.Test(uint(series.UniqueIndex))
 	if !seen {
-		var (
-			tags        = series.Tags
-			encodedTags []byte
-		)
-
-		if tags.Values() != nil {
-			w.tagSliceIter.Reset(tags)
+		var encodedTags []byte
+		if series.EncodedTags != nil {
+			// If already serialized use the serialized tags.
+			encodedTags = series.EncodedTags
+		} else if series.Tags.Values() != nil {
+			// Otherwise serialize the tags.
+			w.tagSliceIter.Reset(series.Tags)
 			w.tagEncoder.Reset()
 			err := w.tagEncoder.Encode(w.tagSliceIter)
 			if err != nil {

src/dbnode/server/server.go

@@ -81,6 +81,7 @@ import (
 	"github.com/m3db/m3/src/x/serialize"
 	xsync "github.com/m3db/m3/src/x/sync"
 
+	apachethrift "github.com/apache/thrift/lib/go/thrift"
 	"github.com/coreos/etcd/embed"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/uber-go/tally"
@@ -1147,6 +1148,9 @@ func withEncodingAndPoolingOptions(
 	iopts := opts.InstrumentOptions()
 	scope := opts.InstrumentOptions().MetricsScope()
 
+	// Set the max bytes pool byte slice alloc size for the thrift pooling.
+	apachethrift.SetMaxBytesPoolAlloc(policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+
 	bytesPoolOpts := pool.NewObjectPoolOptions().
 		SetInstrumentOptions(iopts.SetMetricsScope(scope.SubScope("bytes-pool")))
 	checkedBytesPoolOpts := bytesPoolOpts.

src/dbnode/storage/types.go

@@ -117,11 +117,12 @@ type Database interface {
 	// or WriteTaggedBatch.
 	//
 	// Note that when using the BatchWriter the caller owns the lifecycle of the series
-	// IDs and tag iterators (I.E) if they're being pooled its the callers responsibility
-	// to return them to the appropriate pool, but the annotations are owned by the
+	// IDs if they're being pooled its the callers responsibility to return them to the
+	// appropriate pool, but the encoded tags and annotations are owned by the
 	// ts.WriteBatch itself and will be finalized when the entire ts.WriteBatch is finalized
-	// due to their lifecycle being more complicated. Callers can still control the pooling
-	// of the annotations by using the SetFinalizeAnnotationFn on the WriteBatch itself.
+	// due to their lifecycle being more complicated.
+	// Callers can still control the pooling of the encoded tags and annotations by using
+	// the SetFinalizeEncodedTagsFn and SetFinalizeAnnotationFn on the WriteBatch itself.
 	BatchWriter(namespace ident.ID, batchSize int) (ts.BatchWriter, error)
 
 	// WriteBatch is the same as Write, but in batch.

src/dbnode/ts/types.go

@@ -27,6 +27,10 @@ import (
 	xtime "github.com/m3db/m3/src/x/time"
 )
 
+// FinalizeEncodedTagsFn is a function that will be called for each encoded tags once
+// the WriteBatch itself is finalized.
+type FinalizeEncodedTagsFn func(b []byte)
+
 // FinalizeAnnotationFn is a function that will be called for each annotation once
 // the WriteBatch itself is finalized.
 type FinalizeAnnotationFn func(b []byte)
@@ -72,9 +76,13 @@ type Series struct {
 	// ID is the series identifier.
 	ID ident.ID
 
-	// Tags are the series tags.
+	// Tags is the series tags.
 	Tags ident.Tags
 
+	// EncodedTags are the series encoded tags, if set then call sites can
+	// avoid needing to encoded the tags from the series tags provided.
+	EncodedTags EncodedTags
+
 	// Shard is the shard the series belongs to.
 	Shard uint32
 }
@@ -90,6 +98,9 @@ func (d Datapoint) Equal(x Datapoint) bool {
 	return d.Timestamp.Equal(x.Timestamp) && d.Value == x.Value
 }
 
+// EncodedTags represents the encoded tags for the series.
+type EncodedTags []byte
+
 // Annotation represents information used to annotate datapoints.
 type Annotation []byte
 
@@ -120,17 +131,20 @@ type BatchWriter interface {
 		value float64,
 		unit xtime.Unit,
 		annotation []byte,
-	)
+	) error
 
 	AddTagged(
 		originalIndex int,
 		id ident.ID,
 		tags ident.TagIterator,
+		encodedTags EncodedTags,
 		timestamp time.Time,
 		value float64,
 		unit xtime.Unit,
 		annotation []byte,
-	)
+	) error
+
+	SetFinalizeEncodedTagsFn(f FinalizeEncodedTagsFn)
 
 	SetFinalizeAnnotationFn(f FinalizeAnnotationFn)
 }