Optionally disable m3msg consumer metric scope

src/msg/producer/config/writer.go

@@ -107,6 +107,9 @@ type WriterConfiguration struct {
 	// IgnoreCutoffCutover allows producing writes ignoring cutoff/cutover timestamp.
 	// Must be in sync with AggregatorConfiguration.WritesIgnoreCutoffCutover.
 	IgnoreCutoffCutover bool `yaml:"ignoreCutoffCutover"`
+	// WithoutConsumerScope drops the consumer tag from the metrics. For large m3msg deployments the consumer tag can
+	// add a lot of cardinality to the metrics.
+	WithoutConsumerScope bool `ymal:"withoutConsumerScope"`
 }
 
 // NewOptions creates writer options.
@@ -118,7 +121,8 @@ func (c *WriterConfiguration) NewOptions(
 	opts := writer.NewOptions().
 		SetTopicName(c.TopicName).
 		SetPlacementOptions(c.PlacementOptions.NewOptions()).
-		SetInstrumentOptions(iOpts)
+		SetInstrumentOptions(iOpts).
+		SetWithoutConsumerScope(c.WithoutConsumerScope)
 
 	kvOpts, err := c.TopicServiceOverride.NewOverrideOptions()
 	if err != nil {

src/msg/producer/writer/consumer_service_writer.go

@@ -163,6 +163,7 @@ func initShardWriters(
 		m   = newMessageWriterMetrics(
 			opts.InstrumentOptions().MetricsScope(),
 			opts.InstrumentOptions().TimerOptions(),
+			opts.WithoutConsumerScope(),
 		)
 		mPool messagePool
 	)

src/msg/producer/writer/message_writer.go

@@ -99,6 +99,7 @@ type messageWriter interface {
 }
 
 type messageWriterMetrics struct {
+	withoutConsumerScope     bool
 	scope                    tally.Scope
 	opts                     instrument.TimerOptions
 	writeSuccess             tally.Counter
@@ -127,23 +128,31 @@ type messageWriterMetrics struct {
 }
 
 func (m messageWriterMetrics) withConsumer(consumer string) messageWriterMetrics {
-	return newMessageWriterMetricsWithConsumer(m.scope, m.opts, consumer)
+	if m.withoutConsumerScope {
+		return m
+	}
+	return newMessageWriterMetricsWithConsumer(m.scope, m.opts, consumer, false)
 }
 
 func newMessageWriterMetrics(
 	scope tally.Scope,
 	opts instrument.TimerOptions,
+	withoutConsumerScope bool,
 ) messageWriterMetrics {
-	return newMessageWriterMetricsWithConsumer(scope, opts, "unknown")
+	return newMessageWriterMetricsWithConsumer(scope, opts, "unknown", withoutConsumerScope)
 }
 
 func newMessageWriterMetricsWithConsumer(
 	scope tally.Scope,
 	opts instrument.TimerOptions,
 	consumer string,
-) messageWriterMetrics {
-	consumerScope := scope.Tagged(map[string]string{"consumer": consumer})
+	withoutConsumerScope bool) messageWriterMetrics {
+	consumerScope := scope
+	if !withoutConsumerScope {
+		consumerScope = scope.Tagged(map[string]string{"consumer": consumer})
+	}
 	return messageWriterMetrics{
+		withoutConsumerScope:  withoutConsumerScope,
 		scope:                 scope,
 		opts:                  opts,
 		writeSuccess:          consumerScope.Counter("write-success"),

src/msg/producer/writer/message_writer_test.go

@@ -855,6 +855,22 @@ func TestMessageWriterQueueFullScanOnWriteErrors(t *testing.T) {
 	require.Equal(t, int64(1), counters["message-processed+consumer=c1,result=drop"].Value())
 }
 
+func TestMessageWriter_WithoutConsumerScope(t *testing.T) {
+	ctrl := xtest.NewController(t)
+	defer ctrl.Finish()
+
+	opts := testOptions().SetMessageQueueScanBatchSize(1)
+	scope := tally.NewTestScope("", nil)
+	metrics := newMessageWriterMetrics(scope, instrument.TimerOptions{}, true)
+	w := newMessageWriter(200, nil, opts, metrics).(*messageWriterImpl)
+	w.AddConsumerWriter(newConsumerWriter("bad", nil, opts, testConsumerWriterMetrics()))
+
+	snapshot := scope.Snapshot()
+	counters := snapshot.Counters()
+	require.Nil(t, counters["message-processed+consumer=c1,result=write"])
+	require.NotNil(t, counters["message-processed+result=write"])
+}
+
 func isEmptyWithLock(h *acks) bool {
 	h.Lock()
 	defer h.Unlock()
@@ -868,11 +884,11 @@ func testMessagePool(opts Options) messagePool {
 }
 
 func testMessageWriterMetrics() messageWriterMetrics {
-	return newMessageWriterMetrics(tally.NoopScope, instrument.TimerOptions{})
+	return newMessageWriterMetrics(tally.NoopScope, instrument.TimerOptions{}, false)
 }
 
 func testMessageWriterMetricsWithScope(scope tally.TestScope) messageWriterMetrics {
-	return newMessageWriterMetrics(scope, instrument.TimerOptions{})
+	return newMessageWriterMetrics(scope, instrument.TimerOptions{}, false)
 }
 
 func validateMessages(t *testing.T, msgs []*producer.RefCountedMessage, w *messageWriterImpl) {

src/msg/producer/writer/options.go

@@ -363,6 +363,13 @@ type Options interface {
 
 	// SetIgnoreCutoffCutover sets a flag controlling whether cutoff/cutover timestamps are ignored.
 	SetIgnoreCutoffCutover(value bool) Options
+
+	// WithoutConsumerScope disables the consumer scope for metrics. For large m3msg deploymentssrc/msg/producer/writer/message_writer.go the consumer
+	// scope can add a lot of cardinality to the metrics.
+	WithoutConsumerScope() bool
+
+	// SetWithoutConsumerScope sets the value for WithoutConsumerScope.
+	SetWithoutConsumerScope(value bool) Options
 }
 
 type writerOptions struct {
@@ -385,6 +392,7 @@ type writerOptions struct {
 	cOpts                             ConnectionOptions
 	iOpts                             instrument.Options
 	ignoreCutoffCutover               bool
+	withoutConsumerScope              bool
 }
 
 // NewOptions creates Options.
@@ -596,3 +604,13 @@ func (opts *writerOptions) SetIgnoreCutoffCutover(value bool) Options {
 	o.ignoreCutoffCutover = value
 	return &o
 }
+
+func (opts *writerOptions) WithoutConsumerScope() bool {
+	return opts.withoutConsumerScope
+}
+
+func (opts *writerOptions) SetWithoutConsumerScope(value bool) Options {
+	o := *opts
+	o.withoutConsumerScope = value
+	return &o
+}