diff --git a/platform/extension/messagequeue/mysql/delivery_state_store.go b/platform/extension/messagequeue/mysql/delivery_state_store.go index ff7078ee3..712247c2f 100644 --- a/platform/extension/messagequeue/mysql/delivery_state_store.go +++ b/platform/extension/messagequeue/mysql/delivery_state_store.go @@ -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() @@ -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() @@ -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(` @@ -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() @@ -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() @@ -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 @@ -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 { diff --git a/platform/extension/messagequeue/mysql/offset_store.go b/platform/extension/messagequeue/mysql/offset_store.go index f26b3fb14..a6ed71b34 100644 --- a/platform/extension/messagequeue/mysql/offset_store.go +++ b/platform/extension/messagequeue/mysql/offset_store.go @@ -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) }() @@ -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) }() @@ -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) }() @@ -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 @@ -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) }() diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 3b02d995d..8140ca57a 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -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) }() @@ -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 @@ -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), ) } diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index a5542df31..b7d08e212 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -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") @@ -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") }