From dbd2a9220ad11f9560daf5d7eaec05d7a55efc3f Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Tue, 20 Aug 2019 15:56:39 -0400
Subject: [PATCH 1/9] [dbnode] Use already encoded tags when writing time
 series to commit log

---
 glide.lock                                    |  2 +-
 glide.yaml                                    |  2 +-
 scripts/vagrant/Vagrantfile                   |  4 +-
 src/cmd/services/m3dbnode/config/pooling.go   | 18 ++++-
 .../services/m3dbnode/config/pooling_test.go  | 10 +++
 .../server/tchannelthrift/node/service.go     | 24 +++++--
 .../fs/commitlog/read_write_prop_test.go      | 45 ++++++++++++
 src/dbnode/persist/fs/commitlog/writer.go     | 18 ++---
 src/dbnode/server/server.go                   |  4 ++
 src/dbnode/storage/types.go                   |  9 +--
 src/dbnode/ts/types.go                        | 22 ++++--
 src/dbnode/ts/write_batch.go                  | 70 +++++++++++++++----
 src/dbnode/ts/write_batch_test.go             |  6 ++
 13 files changed, 194 insertions(+), 40 deletions(-)

diff --git a/glide.lock b/glide.lock
index e45bab8422..93ddbd2a0b 100644
--- a/glide.lock
+++ b/glide.lock
@@ -2,7 +2,7 @@ hash: b87408f5ccab483e54a2627f39fbc395c4f5c7cac9f3014aabd110af2de4c20b
 updated: 2019-08-16T22:03:27.199359-04:00
 imports:
 - name: github.com/apache/thrift
-  version: c2fb1c4e8c931d22617bebb0bf388cb4d5e6fcff
+  version: 05b5a2227fe44056ce829fe59583126fd6478a58
   repo: https://github.com/m3db/thrift
   vcs: git
   subpackages:
diff --git a/glide.yaml b/glide.yaml
index 0d43084f8e..8de2ab8d15 100644
--- a/glide.yaml
+++ b/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
diff --git a/scripts/vagrant/Vagrantfile b/scripts/vagrant/Vagrantfile
index 981e7db4a3..c9ace738d3 100644
--- a/scripts/vagrant/Vagrantfile
+++ b/scripts/vagrant/Vagrantfile
@@ -36,11 +36,11 @@ Vagrant.configure("2") do |config|
 
     google.name = "m3-dev-" + $USER
     google.image_family = "ubuntu-1604-lts"
-    google.machine_type = "n1-standard-16"
+    google.machine_type = "n1-highcpu-64"
     google.zone = "us-central1-f"
     google.metadata = {}
     google.tags = ['vagrantbox', 'dev']
-    google.disk_size = '50' # 50gb
+    google.disk_size = '100' # 100gb
     google.autodelete_disk = true
     google.preemptible = false
 
diff --git a/src/cmd/services/m3dbnode/config/pooling.go b/src/cmd/services/m3dbnode/config/pooling.go
index fabcafe34a..65cc53d18d 100644
--- a/src/cmd/services/m3dbnode/config/pooling.go
+++ b/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 {
diff --git a/src/cmd/services/m3dbnode/config/pooling_test.go b/src/cmd/services/m3dbnode/config/pooling_test.go
index c33c305d86..a363508b9f 100644
--- a/src/cmd/services/m3dbnode/config/pooling_test.go
+++ b/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())
+}
diff --git a/src/dbnode/network/server/tchannelthrift/node/service.go b/src/dbnode/network/server/tchannelthrift/node/service.go
index 964dba0bc8..75ad1af024 100644
--- a/src/dbnode/network/server/tchannelthrift/node/service.go
+++ b/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.
diff --git a/src/dbnode/persist/fs/commitlog/read_write_prop_test.go b/src/dbnode/persist/fs/commitlog/read_write_prop_test.go
index 909e9ab28d..91ba2d8eb6 100644
--- a/src/dbnode/persist/fs/commitlog/read_write_prop_test.go
+++ b/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),
diff --git a/src/dbnode/persist/fs/commitlog/writer.go b/src/dbnode/persist/fs/commitlog/writer.go
index 0b9bcf1721..516b8c9136 100644
--- a/src/dbnode/persist/fs/commitlog/writer.go
+++ b/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 {
diff --git a/src/dbnode/server/server.go b/src/dbnode/server/server.go
index e2930b5628..a8c1cec549 100644
--- a/src/dbnode/server/server.go
+++ b/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.
diff --git a/src/dbnode/storage/types.go b/src/dbnode/storage/types.go
index 30acebb96c..f1cb0f1e12 100644
--- a/src/dbnode/storage/types.go
+++ b/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.
diff --git a/src/dbnode/ts/types.go b/src/dbnode/ts/types.go
index 29dde3f1ba..387332c17d 100644
--- a/src/dbnode/ts/types.go
+++ b/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,
+		tags ident.TagsIterator,
+		encodedTags EncodedTags,
 		timestamp time.Time,
 		value float64,
 		unit xtime.Unit,
 		annotation []byte,
-	)
+	) error
+
+	SetFinalizeEncodedTagsFn(f FinalizeEncodedTagsFn)
 
 	SetFinalizeAnnotationFn(f FinalizeAnnotationFn)
 }
diff --git a/src/dbnode/ts/write_batch.go b/src/dbnode/ts/write_batch.go
index 0f5ee95484..da3adf6070 100644
--- a/src/dbnode/ts/write_batch.go
+++ b/src/dbnode/ts/write_batch.go
@@ -21,15 +21,24 @@
 package ts
 
 import (
+	"errors"
 	"time"
 
 	"github.com/m3db/m3/src/x/ident"
 	xtime "github.com/m3db/m3/src/x/time"
 )
 
+var (
+	errTagsAndEncodedTagsRequired = errors.New("tags iterator and encoded tags required to be provided")
+)
+
 type writeBatch struct {
 	writes []BatchWrite
 	ns     ident.ID
+	// Enables callers to pool encoded tags by allowing them to
+	// provide a function to finalize all encoded tags once the
+	// writeBatch itself gets finalized.
+	finalizeEncodedTagsFn FinalizeEncodedTagsFn
 	// Enables callers to pool annotations by allowing them to
 	// provide a function to finalize all annotations once the
 	// writeBatch itself gets finalized.
@@ -57,24 +66,33 @@ func (b *writeBatch) Add(
 	value float64,
 	unit xtime.Unit,
 	annotation []byte,
-) {
-	write := newBatchWriterWrite(
-		originalIndex, b.ns, id, nil, timestamp, value, unit, annotation)
+) error {
+	write, err := newBatchWriterWrite(
+		originalIndex, b.ns, id, nil, nil, timestamp, value, unit, annotation)
+	if err != nil {
+		return err
+	}
 	b.writes = append(b.writes, write)
+	return nil
 }
 
 func (b *writeBatch) AddTagged(
 	originalIndex int,
 	id ident.ID,
-	tagIter ident.TagIterator,
+	tagIter ident.TagsIterator,
+	encodedTags EncodedTags,
 	timestamp time.Time,
 	value float64,
 	unit xtime.Unit,
 	annotation []byte,
-) {
-	write := newBatchWriterWrite(
-		originalIndex, b.ns, id, tagIter, timestamp, value, unit, annotation)
+) error {
+	write, err := newBatchWriterWrite(
+		originalIndex, b.ns, id, tagIter, encodedTags, timestamp, value, unit, annotation)
+	if err != nil {
+		return err
+	}
 	b.writes = append(b.writes, write)
+	return nil
 }
 
 func (b *writeBatch) Reset(
@@ -90,6 +108,7 @@ func (b *writeBatch) Reset(
 
 	b.writes = writes
 	b.ns = ns
+	b.finalizeEncodedTagsFn = nil
 	b.finalizeAnnotationFn = nil
 }
 
@@ -107,6 +126,12 @@ func (b *writeBatch) SetSkipWrite(idx int) {
 	b.writes[idx].SkipWrite = true
 }
 
+// Set the function that will be called to finalize annotations when a WriteBatch
+// is finalized, allowing the caller to pool them.
+func (b *writeBatch) SetFinalizeEncodedTagsFn(f FinalizeEncodedTagsFn) {
+	b.finalizeEncodedTagsFn = f
+}
+
 // Set the function that will be called to finalize annotations when a WriteBatch
 // is finalized, allowing the caller to pool them.
 func (b *writeBatch) SetFinalizeAnnotationFn(f FinalizeAnnotationFn) {
@@ -114,6 +139,18 @@ func (b *writeBatch) SetFinalizeAnnotationFn(f FinalizeAnnotationFn) {
 }
 
 func (b *writeBatch) Finalize() {
+	if b.finalizeEncodedTagsFn != nil {
+		for _, write := range b.writes {
+			encodedTags := write.Write.Series.EncodedTags
+			if encodedTags == nil {
+				continue
+			}
+
+			b.finalizeEncodedTagsFn(encodedTags)
+		}
+	}
+	b.finalizeEncodedTagsFn = nil
+
 	if b.finalizeAnnotationFn != nil {
 		for _, write := range b.writes {
 			annotation := write.Write.Annotation
@@ -146,17 +183,24 @@ func newBatchWriterWrite(
 	originalIndex int,
 	namespace ident.ID,
 	id ident.ID,
-	tagsIter ident.TagIterator,
+	tagIter ident.TagsIterator,
+	encodedTags EncodedTags,
 	timestamp time.Time,
 	value float64,
 	unit xtime.Unit,
 	annotation []byte,
-) BatchWrite {
+) (BatchWrite, error) {
+	write := tagIter == nil && encodedTags == nil
+	writeTagged := tagIter != nil && encodedTags != nil
+	if !write && !writeTagged {
+		return BatchWrite{}, errTagsAndEncodedTagsRequired
+	}
 	return BatchWrite{
 		Write: Write{
 			Series: Series{
-				ID:        id,
-				Namespace: namespace,
+				ID:          id,
+				EncodedTags: encodedTags,
+				Namespace:   namespace,
 			},
 			Datapoint: Datapoint{
 				Timestamp: timestamp,
@@ -165,7 +209,7 @@ func newBatchWriterWrite(
 			Unit:       unit,
 			Annotation: annotation,
 		},
-		TagIter:       tagsIter,
+		TagIter:       tagIter,
 		OriginalIndex: originalIndex,
-	}
+	}, nil
 }
diff --git a/src/dbnode/ts/write_batch_test.go b/src/dbnode/ts/write_batch_test.go
index dec8c07600..8b59ebdd60 100644
--- a/src/dbnode/ts/write_batch_test.go
+++ b/src/dbnode/ts/write_batch_test.go
@@ -238,9 +238,13 @@ func assertDataPresent(t *testing.T, writes []testWrite, batchWriter WriteBatch)
 
 func TestBatchWriterFinalizer(t *testing.T) {
 	var (
+		numEncodedTagsFinalized = 0
 		numAnnotationsFinalized = 0
 		numFinalized            = 0
 
+		finalizeEncodedTagsFn = func(b []byte) {
+			numEncodedTagsFinalized++
+		}
 		finalizeAnnotationFn = func(b []byte) {
 			numAnnotationsFinalized++
 		}
@@ -250,6 +254,7 @@ func TestBatchWriterFinalizer(t *testing.T) {
 	)
 
 	writeBatch := NewWriteBatch(batchSize, namespace, finalizeFn)
+	writeBatch.SetFinalizeEncodedTagsFn(finalizeAnnotationFn)
 	writeBatch.SetFinalizeAnnotationFn(finalizeAnnotationFn)
 
 	for i, write := range writes {
@@ -267,5 +272,6 @@ func TestBatchWriterFinalizer(t *testing.T) {
 	writeBatch.Finalize()
 	require.Equal(t, 0, len(writeBatch.Iter()))
 	require.Equal(t, 1, numFinalized)
+	require.Equal(t, 3, numEncodedTagsFinalized)
 	require.Equal(t, 3, numAnnotationsFinalized)
 }

From 58e51a72de807acad7831f40bde42525a1522281 Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Tue, 20 Aug 2019 16:00:01 -0400
Subject: [PATCH 2/9] Fix config test

---
 src/cmd/services/m3dbnode/config/config_test.go | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/cmd/services/m3dbnode/config/config_test.go b/src/cmd/services/m3dbnode/config/config_test.go
index cf4e92364a..2990290829 100644
--- a/src/cmd/services/m3dbnode/config/config_test.go
+++ b/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:

From f519dc1fa9015aae98f47e22707874f2f236bcbd Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Tue, 20 Aug 2019 16:02:48 -0400
Subject: [PATCH 3/9] Revert vagrantfile changes

---
 scripts/vagrant/Vagrantfile | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/scripts/vagrant/Vagrantfile b/scripts/vagrant/Vagrantfile
index c9ace738d3..981e7db4a3 100644
--- a/scripts/vagrant/Vagrantfile
+++ b/scripts/vagrant/Vagrantfile
@@ -36,11 +36,11 @@ Vagrant.configure("2") do |config|
 
     google.name = "m3-dev-" + $USER
     google.image_family = "ubuntu-1604-lts"
-    google.machine_type = "n1-highcpu-64"
+    google.machine_type = "n1-standard-16"
     google.zone = "us-central1-f"
     google.metadata = {}
     google.tags = ['vagrantbox', 'dev']
-    google.disk_size = '100' # 100gb
+    google.disk_size = '50' # 50gb
     google.autodelete_disk = true
     google.preemptible = false
 

From 2a08f53cf00332b8cedd06b9ce4d387dfd960a4d Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Tue, 20 Aug 2019 20:14:34 -0400
Subject: [PATCH 4/9] Fix build

---
 src/dbnode/ts/types.go       | 2 +-
 src/dbnode/ts/write_batch.go | 4 ++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/dbnode/ts/types.go b/src/dbnode/ts/types.go
index 387332c17d..5dc9c8a261 100644
--- a/src/dbnode/ts/types.go
+++ b/src/dbnode/ts/types.go
@@ -136,7 +136,7 @@ type BatchWriter interface {
 	AddTagged(
 		originalIndex int,
 		id ident.ID,
-		tags ident.TagsIterator,
+		tags ident.TagIterator,
 		encodedTags EncodedTags,
 		timestamp time.Time,
 		value float64,
diff --git a/src/dbnode/ts/write_batch.go b/src/dbnode/ts/write_batch.go
index da3adf6070..1c6e21d0f9 100644
--- a/src/dbnode/ts/write_batch.go
+++ b/src/dbnode/ts/write_batch.go
@@ -79,7 +79,7 @@ func (b *writeBatch) Add(
 func (b *writeBatch) AddTagged(
 	originalIndex int,
 	id ident.ID,
-	tagIter ident.TagsIterator,
+	tagIter ident.TagIterator,
 	encodedTags EncodedTags,
 	timestamp time.Time,
 	value float64,
@@ -183,7 +183,7 @@ func newBatchWriterWrite(
 	originalIndex int,
 	namespace ident.ID,
 	id ident.ID,
-	tagIter ident.TagsIterator,
+	tagIter ident.TagIterator,
 	encodedTags EncodedTags,
 	timestamp time.Time,
 	value float64,

From de0804087c4cd9fa5a0a4bb501cd458f5ec91f12 Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Wed, 21 Aug 2019 01:21:08 +0000
Subject: [PATCH 5/9] Generate mocks

---
 src/dbnode/ts/write_batch_mock.go | 56 ++++++++++++++++++++++++-------
 1 file changed, 44 insertions(+), 12 deletions(-)

diff --git a/src/dbnode/ts/write_batch_mock.go b/src/dbnode/ts/write_batch_mock.go
index 27189ae288..1e1e3b9d60 100644
--- a/src/dbnode/ts/write_batch_mock.go
+++ b/src/dbnode/ts/write_batch_mock.go
@@ -58,9 +58,11 @@ func (m *MockWriteBatch) EXPECT() *MockWriteBatchMockRecorder {
 }
 
 // Add mocks base method
-func (m *MockWriteBatch) Add(originalIndex int, id ident.ID, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) {
+func (m *MockWriteBatch) Add(originalIndex int, id ident.ID, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Add", originalIndex, id, timestamp, value, unit, annotation)
+	ret := m.ctrl.Call(m, "Add", originalIndex, id, timestamp, value, unit, annotation)
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // Add indicates an expected call of Add
@@ -70,15 +72,29 @@ func (mr *MockWriteBatchMockRecorder) Add(originalIndex, id, timestamp, value, u
 }
 
 // AddTagged mocks base method
-func (m *MockWriteBatch) AddTagged(originalIndex int, id ident.ID, tags ident.TagIterator, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) {
+func (m *MockWriteBatch) AddTagged(originalIndex int, id ident.ID, tags ident.TagsIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, timestamp, value, unit, annotation)
+	ret := m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // AddTagged indicates an expected call of AddTagged
-func (mr *MockWriteBatchMockRecorder) AddTagged(originalIndex, id, tags, timestamp, value, unit, annotation interface{}) *gomock.Call {
+func (mr *MockWriteBatchMockRecorder) AddTagged(originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTagged", reflect.TypeOf((*MockWriteBatch)(nil).AddTagged), originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
+}
+
+// SetFinalizeEncodedTagsFn mocks base method
+func (m *MockWriteBatch) SetFinalizeEncodedTagsFn(f FinalizeEncodedTagsFn) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetFinalizeEncodedTagsFn", f)
+}
+
+// SetFinalizeEncodedTagsFn indicates an expected call of SetFinalizeEncodedTagsFn
+func (mr *MockWriteBatchMockRecorder) SetFinalizeEncodedTagsFn(f interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTagged", reflect.TypeOf((*MockWriteBatch)(nil).AddTagged), originalIndex, id, tags, timestamp, value, unit, annotation)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetFinalizeEncodedTagsFn", reflect.TypeOf((*MockWriteBatch)(nil).SetFinalizeEncodedTagsFn), f)
 }
 
 // SetFinalizeAnnotationFn mocks base method
@@ -193,9 +209,11 @@ func (m *MockBatchWriter) EXPECT() *MockBatchWriterMockRecorder {
 }
 
 // Add mocks base method
-func (m *MockBatchWriter) Add(originalIndex int, id ident.ID, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) {
+func (m *MockBatchWriter) Add(originalIndex int, id ident.ID, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Add", originalIndex, id, timestamp, value, unit, annotation)
+	ret := m.ctrl.Call(m, "Add", originalIndex, id, timestamp, value, unit, annotation)
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // Add indicates an expected call of Add
@@ -205,15 +223,29 @@ func (mr *MockBatchWriterMockRecorder) Add(originalIndex, id, timestamp, value,
 }
 
 // AddTagged mocks base method
-func (m *MockBatchWriter) AddTagged(originalIndex int, id ident.ID, tags ident.TagIterator, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) {
+func (m *MockBatchWriter) AddTagged(originalIndex int, id ident.ID, tags ident.TagsIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, timestamp, value, unit, annotation)
+	ret := m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
+	ret0, _ := ret[0].(error)
+	return ret0
 }
 
 // AddTagged indicates an expected call of AddTagged
-func (mr *MockBatchWriterMockRecorder) AddTagged(originalIndex, id, tags, timestamp, value, unit, annotation interface{}) *gomock.Call {
+func (mr *MockBatchWriterMockRecorder) AddTagged(originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTagged", reflect.TypeOf((*MockBatchWriter)(nil).AddTagged), originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
+}
+
+// SetFinalizeEncodedTagsFn mocks base method
+func (m *MockBatchWriter) SetFinalizeEncodedTagsFn(f FinalizeEncodedTagsFn) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "SetFinalizeEncodedTagsFn", f)
+}
+
+// SetFinalizeEncodedTagsFn indicates an expected call of SetFinalizeEncodedTagsFn
+func (mr *MockBatchWriterMockRecorder) SetFinalizeEncodedTagsFn(f interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTagged", reflect.TypeOf((*MockBatchWriter)(nil).AddTagged), originalIndex, id, tags, timestamp, value, unit, annotation)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetFinalizeEncodedTagsFn", reflect.TypeOf((*MockBatchWriter)(nil).SetFinalizeEncodedTagsFn), f)
 }
 
 // SetFinalizeAnnotationFn mocks base method

From dedfc8949956361f3608cfea7ec412b05ce17752 Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Wed, 21 Aug 2019 11:50:00 -0400
Subject: [PATCH 6/9] Feedback and fix tests

---
 scripts/vagrant/Vagrantfile                   | 13 ++++++---
 .../services/m3dbnode/config/config_test.go   |  4 +--
 src/cmd/services/m3dbnode/config/pooling.go   | 18 ++++++-------
 .../services/m3dbnode/config/pooling_test.go  | 10 +++----
 .../server/tchannelthrift/node/service.go     |  2 +-
 src/dbnode/server/server.go                   |  2 +-
 src/dbnode/storage/database_test.go           | 27 ++++++++++++++++---
 src/dbnode/ts/types.go                        |  2 +-
 src/dbnode/ts/write_batch.go                  |  2 +-
 src/dbnode/ts/write_batch_mock.go             |  4 +--
 10 files changed, 56 insertions(+), 28 deletions(-)

diff --git a/scripts/vagrant/Vagrantfile b/scripts/vagrant/Vagrantfile
index 981e7db4a3..d714b51b6b 100644
--- a/scripts/vagrant/Vagrantfile
+++ b/scripts/vagrant/Vagrantfile
@@ -30,19 +30,26 @@ Vagrant.configure("2") do |config|
   end
 
   config.vm.provider :google do |google, override|
-    
+    # Normal benchmarks:
+    google.machine_type = "n2-standard-16"
+    google.preemptible = false
+
+    # For heavy benchmarks, use pre-emptible n2-standard-64:
+    # google.machine_type = "n2-standard-64"
+    # google.preemptible = true
+    # google.auto_restart = false
+    # google.on_host_maintenance = "TERMINATE"
+
     google.google_project_id = $GOOGLE_PROJECT_ID
     google.google_json_key_location = $GOOGLE_JSON_KEY_LOCATION
 
     google.name = "m3-dev-" + $USER
     google.image_family = "ubuntu-1604-lts"
-    google.machine_type = "n1-standard-16"
     google.zone = "us-central1-f"
     google.metadata = {}
     google.tags = ['vagrantbox', 'dev']
     google.disk_size = '50' # 50gb
     google.autodelete_disk = true
-    google.preemptible = false
 
     override.ssh.username = $USER
     override.ssh.private_key_path = $SSH_KEY
diff --git a/src/cmd/services/m3dbnode/config/config_test.go b/src/cmd/services/m3dbnode/config/config_test.go
index 2990290829..750c0ed08d 100644
--- a/src/cmd/services/m3dbnode/config/config_test.go
+++ b/src/cmd/services/m3dbnode/config/config_test.go
@@ -133,7 +133,7 @@ db:
 
   pooling:
       blockAllocSize: 16
-      thriftBytesPoolMaxAllocSize: 2048
+      thriftBytesPoolAllocSize: 2048
       type: simple
       seriesPool:
           size: 5242880
@@ -446,7 +446,7 @@ func TestConfiguration(t *testing.T) {
   replication: null
   pooling:
     blockAllocSize: 16
-    thriftBytesPoolMaxAllocSize: 2048
+    thriftBytesPoolAllocSize: 2048
     type: simple
     bytesPool:
       buckets:
diff --git a/src/cmd/services/m3dbnode/config/pooling.go b/src/cmd/services/m3dbnode/config/pooling.go
index 65cc53d18d..53f10bf8a5 100644
--- a/src/cmd/services/m3dbnode/config/pooling.go
+++ b/src/cmd/services/m3dbnode/config/pooling.go
@@ -33,9 +33,9 @@ const (
 )
 
 const (
-	defaultMaxFinalizerCapacity        = 4
-	defaultBlockAllocSize              = 16
-	defaultThriftBytesPoolMaxAllocSize = 1024
+	defaultMaxFinalizerCapacity     = 4
+	defaultBlockAllocSize           = 16
+	defaultThriftBytesPoolAllocSize = 1024
 )
 
 type poolPolicyDefault struct {
@@ -254,7 +254,7 @@ type PoolingPolicy struct {
 	BlockAllocSize *int `yaml:"blockAllocSize"`
 
 	// The thrift bytes pool max bytes slice allocation for a single binary field.
-	ThriftBytesPoolMaxAllocSize *int `yaml:"thriftBytesPoolMaxAllocSize"`
+	ThriftBytesPoolAllocSize *int `yaml:"thriftBytesPoolAllocSize"`
 
 	// The general pool type (currently only supported: simple).
 	Type *PoolingType `yaml:"type"`
@@ -422,14 +422,14 @@ func (p *PoolingPolicy) BlockAllocSizeOrDefault() int {
 	return defaultBlockAllocSize
 }
 
-// ThriftBytesPoolMaxAllocSizeOrDefault returns the configured thrift bytes pool
+// ThriftBytesPoolAllocSizeOrDefault 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
+func (p *PoolingPolicy) ThriftBytesPoolAllocSizeOrDefault() int {
+	if p.ThriftBytesPoolAllocSize != nil {
+		return *p.ThriftBytesPoolAllocSize
 	}
 
-	return defaultThriftBytesPoolMaxAllocSize
+	return defaultThriftBytesPoolAllocSize
 }
 
 // TypeOrDefault returns the configured pooling type if provided, or a default
diff --git a/src/cmd/services/m3dbnode/config/pooling_test.go b/src/cmd/services/m3dbnode/config/pooling_test.go
index a363508b9f..83d5163d35 100644
--- a/src/cmd/services/m3dbnode/config/pooling_test.go
+++ b/src/cmd/services/m3dbnode/config/pooling_test.go
@@ -52,12 +52,12 @@ func TestContextPoolMaxFinalizerCapacityOrDefault(t *testing.T) {
 	require.Equal(t, 10, cpp.MaxFinalizerCapacityOrDefault())
 }
 
-func TestPoolingPolicyThriftBytesPoolMaxAllocSizeOrDefault(t *testing.T) {
+func TestPoolingPolicyThriftBytesPoolAllocSizeOrDefault(t *testing.T) {
 	policy := PoolingPolicy{}
-	require.Equal(t, defaultThriftBytesPoolMaxAllocSize,
-		policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+	require.Equal(t, defaultThriftBytesPoolAllocSize,
+		policy.ThriftBytesPoolAllocSizeOrDefault())
 
 	value := 42
-	policy.ThriftBytesPoolMaxAllocSize = &value
-	require.Equal(t, 42, policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+	policy.ThriftBytesPoolAllocSize = &value
+	require.Equal(t, 42, policy.ThriftBytesPoolAllocSizeOrDefault())
 }
diff --git a/src/dbnode/network/server/tchannelthrift/node/service.go b/src/dbnode/network/server/tchannelthrift/node/service.go
index 75ad1af024..3594a90c92 100644
--- a/src/dbnode/network/server/tchannelthrift/node/service.go
+++ b/src/dbnode/network/server/tchannelthrift/node/service.go
@@ -1769,7 +1769,7 @@ func (r *writeBatchPooledReq) Finalize() {
 	if r.writeTaggedReq != nil {
 		for _, elem := range r.writeTaggedReq.Elements {
 			apachethrift.BytesPoolPut(elem.ID)
-			// Ownership of the encoded tagts has been transferred to the BatchWriter
+			// Ownership of the encoded tags 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.
diff --git a/src/dbnode/server/server.go b/src/dbnode/server/server.go
index a8c1cec549..155849a718 100644
--- a/src/dbnode/server/server.go
+++ b/src/dbnode/server/server.go
@@ -1149,7 +1149,7 @@ func withEncodingAndPoolingOptions(
 	scope := opts.InstrumentOptions().MetricsScope()
 
 	// Set the max bytes pool byte slice alloc size for the thrift pooling.
-	apachethrift.SetMaxBytesPoolAlloc(policy.ThriftBytesPoolMaxAllocSizeOrDefault())
+	apachethrift.SetMaxBytesPoolAlloc(policy.ThriftBytesPoolAllocSizeOrDefault())
 
 	bytesPoolOpts := pool.NewObjectPoolOptions().
 		SetInstrumentOptions(iopts.SetMetricsScope(scope.SubScope("bytes-pool")))
diff --git a/src/dbnode/storage/database_test.go b/src/dbnode/storage/database_test.go
index 58ea1a0918..112f360c6d 100644
--- a/src/dbnode/storage/database_test.go
+++ b/src/dbnode/storage/database_test.go
@@ -48,6 +48,8 @@ import (
 	"github.com/m3db/m3/src/x/context"
 	xerrors "github.com/m3db/m3/src/x/errors"
 	"github.com/m3db/m3/src/x/ident"
+	"github.com/m3db/m3/src/x/pool"
+	"github.com/m3db/m3/src/x/serialize"
 	xtime "github.com/m3db/m3/src/x/time"
 	xwatch "github.com/m3db/m3/src/x/watch"
 
@@ -964,9 +966,26 @@ func testDatabaseWriteBatch(t *testing.T,
 	var (
 		namespace = ident.StringID("testns")
 		ctx       = context.NewContext()
-		tagsIter  = ident.EmptyTagIterator
+		tags      = ident.NewTags(ident.Tag{
+			Name:  ident.StringID("foo"),
+			Value: ident.StringID("bar"),
+		}, ident.Tag{
+			Name:  ident.StringID("baz"),
+			Value: ident.StringID("qux"),
+		})
+		tagsIter = ident.NewTagsIterator(tags)
 	)
 
+	testTagEncodingPool := serialize.NewTagEncoderPool(serialize.NewTagEncoderOptions(),
+		pool.NewObjectPoolOptions().SetSize(1))
+	testTagEncodingPool.Init()
+	encoder := testTagEncodingPool.Get()
+	err := encoder.Encode(tagsIter)
+	require.NoError(t, err)
+
+	encodedTags, ok := encoder.Data()
+	require.True(t, ok)
+
 	writes := []struct {
 		series string
 		t      time.Time
@@ -1025,7 +1044,8 @@ func testDatabaseWriteBatch(t *testing.T,
 		// ErrorHandler is called with the provided index, not the actual position
 		// in the WriteBatch slice.
 		if tagged {
-			batchWriter.AddTagged(i*2, ident.StringID(write.series), tagsIter, write.t, write.v, xtime.Second, nil)
+			batchWriter.AddTagged(i*2, ident.StringID(write.series),
+				tagsIter.Duplicate(), encodedTags.Bytes(), write.t, write.v, xtime.Second, nil)
 			wasWritten := write.err == nil
 			ns.EXPECT().WriteTagged(ctx, ident.NewIDMatcher(write.series), gomock.Any(),
 				write.t, write.v, xtime.Second, nil).Return(
@@ -1035,7 +1055,8 @@ func testDatabaseWriteBatch(t *testing.T,
 					Tags:      ident.Tags{},
 				}, wasWritten, write.err)
 		} else {
-			batchWriter.Add(i*2, ident.StringID(write.series), write.t, write.v, xtime.Second, nil)
+			batchWriter.Add(i*2, ident.StringID(write.series),
+				write.t, write.v, xtime.Second, nil)
 			wasWritten := write.err == nil
 			ns.EXPECT().Write(ctx, ident.NewIDMatcher(write.series),
 				write.t, write.v, xtime.Second, nil).Return(
diff --git a/src/dbnode/ts/types.go b/src/dbnode/ts/types.go
index 5dc9c8a261..d97b5f1157 100644
--- a/src/dbnode/ts/types.go
+++ b/src/dbnode/ts/types.go
@@ -76,7 +76,7 @@ type Series struct {
 	// ID is the series identifier.
 	ID ident.ID
 
-	// Tags is the series tags.
+	// Tags are the series tags.
 	Tags ident.Tags
 
 	// EncodedTags are the series encoded tags, if set then call sites can
diff --git a/src/dbnode/ts/write_batch.go b/src/dbnode/ts/write_batch.go
index 1c6e21d0f9..64d439ead5 100644
--- a/src/dbnode/ts/write_batch.go
+++ b/src/dbnode/ts/write_batch.go
@@ -29,7 +29,7 @@ import (
 )
 
 var (
-	errTagsAndEncodedTagsRequired = errors.New("tags iterator and encoded tags required to be provided")
+	errTagsAndEncodedTagsRequired = errors.New("tags iterator and encoded tags must be provided")
 )
 
 type writeBatch struct {
diff --git a/src/dbnode/ts/write_batch_mock.go b/src/dbnode/ts/write_batch_mock.go
index 1e1e3b9d60..f8a5ff682f 100644
--- a/src/dbnode/ts/write_batch_mock.go
+++ b/src/dbnode/ts/write_batch_mock.go
@@ -72,7 +72,7 @@ func (mr *MockWriteBatchMockRecorder) Add(originalIndex, id, timestamp, value, u
 }
 
 // AddTagged mocks base method
-func (m *MockWriteBatch) AddTagged(originalIndex int, id ident.ID, tags ident.TagsIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
+func (m *MockWriteBatch) AddTagged(originalIndex int, id ident.ID, tags ident.TagIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
 	ret0, _ := ret[0].(error)
@@ -223,7 +223,7 @@ func (mr *MockBatchWriterMockRecorder) Add(originalIndex, id, timestamp, value,
 }
 
 // AddTagged mocks base method
-func (m *MockBatchWriter) AddTagged(originalIndex int, id ident.ID, tags ident.TagsIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
+func (m *MockBatchWriter) AddTagged(originalIndex int, id ident.ID, tags ident.TagIterator, encodedTags EncodedTags, timestamp time.Time, value float64, unit time0.Unit, annotation []byte) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "AddTagged", originalIndex, id, tags, encodedTags, timestamp, value, unit, annotation)
 	ret0, _ := ret[0].(error)

From d0dfa3c912438d6e3f8981ce6bfe92e971b50f7d Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Wed, 21 Aug 2019 17:25:51 -0400
Subject: [PATCH 7/9] Fix tests

---
 src/dbnode/ts/write_batch_test.go | 34 ++++++++++++++++++++++++++++++-
 1 file changed, 33 insertions(+), 1 deletion(-)

diff --git a/src/dbnode/ts/write_batch_test.go b/src/dbnode/ts/write_batch_test.go
index 8b59ebdd60..12b76a2c2f 100644
--- a/src/dbnode/ts/write_batch_test.go
+++ b/src/dbnode/ts/write_batch_test.go
@@ -24,10 +24,14 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/ident"
+	"github.com/m3db/m3/src/x/pool"
+	"github.com/m3db/m3/src/x/serialize"
 	xtime "github.com/m3db/m3/src/x/time"
 
 	"github.com/stretchr/testify/require"
@@ -79,6 +83,23 @@ var (
 	}
 )
 
+var (
+	testTagEncodingPool = struct {
+		once sync.Once
+		pool serialize.TagEncoderPool
+	}{
+		pool: serialize.NewTagEncoderPool(serialize.NewTagEncoderOptions(),
+			pool.NewObjectPoolOptions().SetSize(1)),
+	}
+)
+
+func getTagEncoder() serialize.TagEncoder {
+	testTagEncodingPool.once.Do(func() {
+		testTagEncodingPool.pool.Init()
+	})
+	return testTagEncodingPool.pool.Get()
+}
+
 type testWrite struct {
 	id         ident.ID
 	tagIter    ident.TagIterator
@@ -88,6 +109,14 @@ type testWrite struct {
 	annotation []byte
 }
 
+func (w testWrite) encodedTags(t *testing.T) checked.Bytes {
+	encoder := getTagEncoder()
+	require.NoError(t, encoder.Encode(w.tagIter.Duplicate()))
+	data, ok := encoder.Data()
+	require.True(t, ok)
+	return data
+}
+
 func TestBatchWriterAddAndIter(t *testing.T) {
 	writeBatch := NewWriteBatch(batchSize, namespace, nil)
 
@@ -113,6 +142,7 @@ func TestBatchWriterAddTaggedAndIter(t *testing.T) {
 			i,
 			write.id,
 			write.tagIter,
+			write.encodedTags(t).Bytes(),
 			write.timestamp,
 			write.value,
 			write.unit,
@@ -131,6 +161,7 @@ func TestBatchWriterSetSeries(t *testing.T) {
 			i,
 			write.id,
 			write.tagIter,
+			write.encodedTags(t).Bytes(),
 			write.timestamp,
 			write.value,
 			write.unit,
@@ -254,7 +285,7 @@ func TestBatchWriterFinalizer(t *testing.T) {
 	)
 
 	writeBatch := NewWriteBatch(batchSize, namespace, finalizeFn)
-	writeBatch.SetFinalizeEncodedTagsFn(finalizeAnnotationFn)
+	writeBatch.SetFinalizeEncodedTagsFn(finalizeEncodedTagsFn)
 	writeBatch.SetFinalizeAnnotationFn(finalizeAnnotationFn)
 
 	for i, write := range writes {
@@ -262,6 +293,7 @@ func TestBatchWriterFinalizer(t *testing.T) {
 			i,
 			write.id,
 			write.tagIter,
+			write.encodedTags(t).Bytes(),
 			write.timestamp,
 			write.value,
 			write.unit,

From d0efa4ac4b61884234cae44a8d77e4704996ba41 Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Wed, 21 Aug 2019 18:13:46 -0400
Subject: [PATCH 8/9] Use n1 type nodes

---
 scripts/vagrant/Vagrantfile | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/scripts/vagrant/Vagrantfile b/scripts/vagrant/Vagrantfile
index d714b51b6b..e7ad88c111 100644
--- a/scripts/vagrant/Vagrantfile
+++ b/scripts/vagrant/Vagrantfile
@@ -31,11 +31,11 @@ Vagrant.configure("2") do |config|
 
   config.vm.provider :google do |google, override|
     # Normal benchmarks:
-    google.machine_type = "n2-standard-16"
+    # google.machine_type = "n1-standard-16"
     google.preemptible = false
 
-    # For heavy benchmarks, use pre-emptible n2-standard-64:
-    # google.machine_type = "n2-standard-64"
+    # For heavy benchmarks, use pre-emptible n1-standard-64:
+    google.machine_type = "n1-standard-64"
     # google.preemptible = true
     # google.auto_restart = false
     # google.on_host_maintenance = "TERMINATE"

From 6c44a91c8485b27182257ff4075f2b8492a65af9 Mon Sep 17 00:00:00 2001
From: Rob Skillington <rob.skillington@gmail.com>
Date: Wed, 21 Aug 2019 18:14:40 -0400
Subject: [PATCH 9/9] Use normal benchmark vagrantfile

---
 scripts/vagrant/Vagrantfile | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/scripts/vagrant/Vagrantfile b/scripts/vagrant/Vagrantfile
index e7ad88c111..8647f835ad 100644
--- a/scripts/vagrant/Vagrantfile
+++ b/scripts/vagrant/Vagrantfile
@@ -31,11 +31,11 @@ Vagrant.configure("2") do |config|
 
   config.vm.provider :google do |google, override|
     # Normal benchmarks:
-    # google.machine_type = "n1-standard-16"
+    google.machine_type = "n1-standard-16"
     google.preemptible = false
 
     # For heavy benchmarks, use pre-emptible n1-standard-64:
-    google.machine_type = "n1-standard-64"
+    # google.machine_type = "n1-standard-64"
     # google.preemptible = true
     # google.auto_restart = false
     # google.on_host_maintenance = "TERMINATE"