[dbnode] Add metric to count times when timeseries annotation changes (…

…#3565)

src/cmd/services/m3comparator/main/main.go

@@ -62,7 +62,10 @@ func init() {
 	checkedBytesPool = pool.NewCheckedBytesPool(buckets, nil, newBackingBytesPool)
 	checkedBytesPool.Init()
 
-	encodingOpts = encoding.NewOptions().SetEncoderPool(encoderPool).SetBytesPool(checkedBytesPool)
+	encodingOpts = encoding.NewOptions().
+		SetEncoderPool(encoderPool).
+		SetBytesPool(checkedBytesPool).
+		SetMetrics(encoding.NewMetrics(iOpts.MetricsScope()))
 
 	encoderPool.Init(func() encoding.Encoder {
 		return m3tsz.NewEncoder(0, nil, true, encodingOpts)

src/dbnode/encoding/m3tsz/timestamp_encoder.go

@@ -49,6 +49,8 @@ type TimestampEncoder struct {
 	timeUnitEncodedManually bool
 	// Only taken into account if using the WriteTime() API.
 	hasWrittenFirst bool
+
+	metrics encoding.TimestampEncoderMetrics
 }
 
 var emptyAnnotationChecksum = xxhash.Sum64(nil)
@@ -62,6 +64,7 @@ func NewTimestampEncoder(
 		PrevAnnotationChecksum: emptyAnnotationChecksum,
 		markerEncodingScheme:   opts.MarkerEncodingScheme(),
 		timeEncodingSchemes:    opts.TimeEncodingSchemes(),
+		metrics:                opts.Metrics().TimestampEncoder,
 	}
 }
 
@@ -182,6 +185,12 @@ func (enc *TimestampEncoder) writeAnnotation(stream encoding.OStream, ant ts.Ann
 	stream.WriteBytes(buf[:annotationLength])
 	stream.WriteBytes(ant)
 
+	if enc.PrevAnnotationChecksum != emptyAnnotationChecksum {
+		// NB: current assumption is that each time series should have a single annotation write per block
+		// and that annotations should be rewritten rarely. If this assumption changes, it might not be worth
+		// keeping this metric around.
+		enc.metrics.IncAnnotationRewritten()
+	}
 	enc.PrevAnnotationChecksum = checksum
 }
 

src/dbnode/encoding/metrics.go

@@ -0,0 +1,53 @@
+// Copyright (c) 2021 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 encoding
+
+import "github.com/uber-go/tally"
+
+// Metrics contains metrics for encoding.
+type Metrics struct {
+	TimestampEncoder TimestampEncoderMetrics
+}
+
+// NewMetrics returns new Metrics.
+func NewMetrics(scope tally.Scope) Metrics {
+	return Metrics{
+		TimestampEncoder: newTimestampEncoderMetrics(scope.SubScope("timestamp-encoder")),
+	}
+}
+
+// TimestampEncoderMetrics contains timestamp encoder metrics.
+type TimestampEncoderMetrics struct {
+	annotationRewritten tally.Counter
+}
+
+func newTimestampEncoderMetrics(scope tally.Scope) TimestampEncoderMetrics {
+	return TimestampEncoderMetrics{
+		annotationRewritten: scope.Counter("annotation-rewritten"),
+	}
+}
+
+// IncAnnotationRewritten increments annotation rewritten counter.
+func (m *TimestampEncoderMetrics) IncAnnotationRewritten() {
+	if m.annotationRewritten != nil {
+		m.annotationRewritten.Inc(1)
+	}
+}

src/dbnode/encoding/options.go

@@ -23,6 +23,7 @@ package encoding
 import (
 	"github.com/m3db/m3/src/dbnode/x/xio"
 	"github.com/m3db/m3/src/dbnode/x/xpool"
+	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
 	xtime "github.com/m3db/m3/src/x/time"
 )
@@ -51,6 +52,7 @@ type options struct {
 	byteFieldDictLRUSize    int
 	iStreamReaderSizeM3TSZ  int
 	iStreamReaderSizeProto  int
+	metrics                 Metrics
 }
 
 func newOptions() Options {
@@ -61,6 +63,7 @@ func newOptions() Options {
 		byteFieldDictLRUSize:   defaultByteFieldDictLRUSize,
 		iStreamReaderSizeM3TSZ: defaultIStreamReaderSizeM3TSZ,
 		iStreamReaderSizeProto: defaultIStreamReaderSizeProto,
+		metrics:                NewMetrics(instrument.NewOptions().MetricsScope()),
 	}
 }
 
@@ -178,3 +181,13 @@ func (o *options) SetIStreamReaderSizeProto(value int) Options {
 func (o *options) IStreamReaderSizeProto() int {
 	return o.iStreamReaderSizeProto
 }
+
+func (o *options) SetMetrics(value Metrics) Options {
+	opts := *o
+	opts.metrics = value
+	return &opts
+}
+
+func (o *options) Metrics() Metrics {
+	return o.metrics
+}

src/dbnode/encoding/types.go

@@ -165,6 +165,12 @@ type Options interface {
 	// SetIStreamReaderSizeProto returns the IStream bufio reader size
 	// for proto encoding iteration.
 	IStreamReaderSizeProto() int
+
+	// SetMetrics sets the encoding metrics.
+	SetMetrics(value Metrics) Options
+
+	// Metrics returns the encoding metrics.
+	Metrics() Metrics
 }
 
 // Iterator is the generic interface for iterating over encoded data.

src/dbnode/server/server.go

@@ -1681,7 +1681,8 @@ func withEncodingAndPoolingOptions(
 		SetReaderIteratorPool(iteratorPool).
 		SetBytesPool(bytesPool).
 		SetSegmentReaderPool(segmentReaderPool).
-		SetCheckedBytesWrapperPool(bytesWrapperPool)
+		SetCheckedBytesWrapperPool(bytesWrapperPool).
+		SetMetrics(encoding.NewMetrics(scope))
 
 	encoderPool.Init(func() encoding.Encoder {
 		if cfg.Proto != nil && cfg.Proto.Enabled {

src/dbnode/storage/block/options.go

@@ -28,6 +28,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/x/xio"
 	"github.com/m3db/m3/src/x/clock"
 	"github.com/m3db/m3/src/x/context"
+	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
 	xsync "github.com/m3db/m3/src/x/sync"
 )
@@ -59,6 +60,7 @@ type options struct {
 
 // NewOptions creates new database block options
 func NewOptions() Options {
+	iOpts := instrument.NewOptions()
 	bytesPool := pool.NewCheckedBytesPool(nil, nil, func(s []pool.Bucket) pool.BytesPool {
 		return pool.NewBytesPool(s, nil)
 	})
@@ -86,7 +88,8 @@ func NewOptions() Options {
 		SetBytesPool(bytesPool).
 		SetEncoderPool(encoderPool).
 		SetReaderIteratorPool(readerIteratorPool).
-		SetSegmentReaderPool(segmentReaderPool)
+		SetSegmentReaderPool(segmentReaderPool).
+		SetMetrics(encoding.NewMetrics(iOpts.MetricsScope()))
 
 	o.encoderPool.Init(func() encoding.Encoder {
 		return m3tsz.NewEncoder(timeZero, nil, m3tsz.DefaultIntOptimizationEnabled, encodingOpts)

src/dbnode/storage/options.go

@@ -493,7 +493,8 @@ func (o *options) SetEncodingM3TSZPooled() Options {
 		SetBytesPool(bytesPool).
 		SetEncoderPool(encoderPool).
 		SetReaderIteratorPool(readerIteratorPool).
-		SetSegmentReaderPool(segmentReaderPool)
+		SetSegmentReaderPool(segmentReaderPool).
+		SetMetrics(encoding.NewMetrics(opts.InstrumentOptions().MetricsScope()))
 
 	// initialize encoder pool
 	encoderPool.Init(func() encoding.Encoder {