Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Optionally disable m3msg consumer metric scope #3802

Merged
merged 3 commits into from
Oct 1, 2021
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion src/msg/producer/config/writer.go
Original file line number Diff line number Diff line change
@@ -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 `yaml:"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 {
1 change: 1 addition & 0 deletions src/msg/producer/writer/consumer_service_writer.go
Original file line number Diff line number Diff line change
@@ -163,6 +163,7 @@ func initShardWriters(
m = newMessageWriterMetrics(
opts.InstrumentOptions().MetricsScope(),
opts.InstrumentOptions().TimerOptions(),
opts.WithoutConsumerScope(),
)
mPool messagePool
)
17 changes: 13 additions & 4 deletions src/msg/producer/writer/message_writer.go
Original file line number Diff line number Diff line change
@@ -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"),
20 changes: 18 additions & 2 deletions src/msg/producer/writer/message_writer_test.go
Original file line number Diff line number Diff line change
@@ -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) {
18 changes: 18 additions & 0 deletions src/msg/producer/writer/options.go
Original file line number Diff line number Diff line change
@@ -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 deployments 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
}