Skip to content

fix(messagequeue): drop the partition key from queue metric tags #593

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.

Sign up for GitHub

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

Merged
mnoah1 merged 2 commits into from
Aug 14, 2026
Merged

fix(messagequeue): drop the partition key from queue metric tags #593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b257e14
fix(messagequeue): drop the partition key from queue metric tags
mnoah1 Aug 14, 2026
e6aa79b
Merge branch 'main' into mnoah1/queue-metric-partition-key
mnoah1 Aug 14, 2026
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
21 changes: 7 additions & 14 deletions platform/extension/messagequeue/mysql/delivery_state_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,7 @@ func newDeliveryStateStore(db *sql.DB, logger *zap.SugaredLogger, scope tally.Sc
func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (_ int, retErr error) {
op := metrics.Begin(s.scope, "mark_delivered", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

now := time.Now().UnixMilli()
Expand Down Expand Up @@ -97,8 +96,7 @@ func (s *sqldeliveryStateStore) MarkDelivered(ctx context.Context, consumerGroup
func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, visibilityTimeoutMs int64) (retErr error) {
op := metrics.Begin(s.scope, "extend_visibility", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

now := time.Now().UnixMilli()
Expand Down Expand Up @@ -131,8 +129,7 @@ func (s *sqldeliveryStateStore) ExtendVisibility(ctx context.Context, consumerGr
func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) {
op := metrics.Begin(s.scope, "mark_acked", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

_, err := s.db.ExecContext(ctx, fmt.Sprintf(`
Expand All @@ -155,8 +152,7 @@ func (s *sqldeliveryStateStore) MarkAcked(ctx context.Context, consumerGroup, to
func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (retErr error) {
op := metrics.Begin(s.scope, "mark_nacked", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

invisibleUntil := time.Now().UnixMilli()
Expand Down Expand Up @@ -185,8 +181,7 @@ func (s *sqldeliveryStateStore) MarkNacked(ctx context.Context, consumerGroup, t
func (s *sqldeliveryStateStore) MarkPostponed(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64, delayMs int64) (retErr error) {
op := metrics.Begin(s.scope, "mark_postponed", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

now := time.Now().UnixMilli()
Expand Down Expand Up @@ -214,8 +209,7 @@ func (s *sqldeliveryStateStore) MarkPostponed(ctx context.Context, consumerGroup
func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGroup, topic, partitionKey string, offset int64) (_ DeliveryState, _ bool, retErr error) {
op := metrics.Begin(s.scope, "get_delivery_state", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

var state DeliveryState
Expand All @@ -241,8 +235,7 @@ func (s *sqldeliveryStateStore) GetDeliveryState(ctx context.Context, consumerGr
func (s *sqldeliveryStateStore) AdvanceWatermark(ctx context.Context, consumerGroup, topic, partitionKey string, currentWatermark int64, offsets []int64) (_ int64, retErr error) {
op := metrics.Begin(s.scope, "advance_watermark", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("consumer_group", consumerGroup),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

if len(offsets) == 0 {
Expand Down
7 changes: 1 addition & 6 deletions platform/extension/messagequeue/mysql/offset_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,6 @@ func newOffsetStore(db *sql.DB, scope tally.Scope) offsetStore {
func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partitionKey string, consumerGroup string) (retErr error) {
op := metrics.Begin(s.scope, "initialize", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("partition_key", partitionKey),
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

Expand All @@ -65,7 +64,6 @@ func (s *sqloffsetStore) Initialize(ctx context.Context, topic string, partition
func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, partitionKey string, consumerGroup string) (_ int64, retErr error) {
op := metrics.Begin(s.scope, "get_acked_offset", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("partition_key", partitionKey),
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

Expand All @@ -90,7 +88,6 @@ func (s *sqloffsetStore) GetAckedOffset(ctx context.Context, topic string, parti
func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, partitionKey string, offset int64, consumerGroup string) (retErr error) {
op := metrics.Begin(s.scope, "update_acked_offset", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("partition_key", partitionKey),
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

Expand All @@ -113,8 +110,7 @@ func (s *sqloffsetStore) UpdateAckedOffset(ctx context.Context, topic string, pa
// for a topic+partition. Returns (0, false, nil) if no offset rows exist.
func (s *sqloffsetStore) GetMinAckedOffset(ctx context.Context, topic string, partitionKey string) (_ int64, _ bool, retErr error) {
op := metrics.Begin(s.scope, "get_min_acked_offset", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("partition_key", partitionKey))
metrics.NewTag("topic", topic))
defer func() { op.Complete(retErr) }()

var minOffset int64
Expand All @@ -138,7 +134,6 @@ func (s *sqloffsetStore) GetMinAckedOffset(ctx context.Context, topic string, pa
func (s *sqloffsetStore) DeleteOffset(ctx context.Context, topic string, partitionKey string, consumerGroup string) (retErr error) {
op := metrics.Begin(s.scope, "delete_offset", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", topic),
metrics.NewTag("partition_key", partitionKey),
metrics.NewTag("consumer_group", consumerGroup))
defer func() { op.Complete(retErr) }()

Expand Down
3 changes: 0 additions & 3 deletions platform/extension/messagequeue/mysql/subscriber.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1019,7 +1019,6 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) {

op := metrics.Begin(s.scope, "poll", metrics.StorageLatencyBuckets,
metrics.NewTag("topic", sub.topic),
metrics.NewTag("partition_key", partitionKey),
)
defer func() { op.Complete(retErr) }()

Expand Down Expand Up @@ -1115,7 +1114,6 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) {
messageAge := time.Duration(time.Now().UnixMilli()-row.PublishedAt) * time.Millisecond
metrics.NamedHistogram(s.scope, "poll", "message_age", metrics.LongLatencyBuckets,
metrics.NewTag("topic", sub.topic),
metrics.NewTag("partition_key", partitionKey),
).RecordDuration(messageAge)

// Create delivery ID from offset
Expand Down Expand Up @@ -1219,7 +1217,6 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) {
if messageCount > 0 {
metrics.NamedCounter(s.scope, "poll", "messages_delivered", int64(messageCount),
metrics.NewTag("topic", sub.topic),
metrics.NewTag("partition_key", partitionKey),
)
}

Expand Down
7 changes: 5 additions & 2 deletions platform/extension/messagequeue/mysql/subscriber_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -683,13 +683,16 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
// Verify offset was initialized only once
assert.True(t, w.offsetInitialized)

// The partition key is deliberately absent: topics partitioned by an entity
// ID mint a key per request, batch or build, and tally never reclaims the
// subscope a tag value creates. Partition identity stays in the logs.
snapshot := metricsScope.Snapshot()
var foundStart bool
for _, counter := range snapshot.Counters() {
if counter.Name() == "test.subscriber.poll.start" {
foundStart = true
assert.Equal(t, "test_topic", counter.Tags()["topic"])
assert.Equal(t, "part-1", counter.Tags()["partition_key"])
assert.NotContains(t, counter.Tags(), "partition_key")
}
}
assert.True(t, foundStart, "expected poll.start counter")
Expand All @@ -700,7 +703,7 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
foundFinish = true
assert.Equal(t, "success", histogram.Tags()["result"])
assert.Equal(t, "test_topic", histogram.Tags()["topic"])
assert.Equal(t, "part-1", histogram.Tags()["partition_key"])
assert.NotContains(t, histogram.Tags(), "partition_key")
}
assert.NotContains(t, histogram.Name(), "poll.latency")
}
Expand Down
Loading