From bd1b97c536011fea00d3d964074d6ce775c13703 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 14 Aug 2026 18:18:47 +0000 Subject: [PATCH] feat(consumer): expose concrete topic in metrics Tag controller metrics with both the stable logical topic key and the configured backend topic so observability can correlate consumer and queue series. --- platform/consumer/consumer.go | 5 +++-- platform/consumer/consumer_test.go | 16 ++++++++++++++-- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index f3dcdbb4d..c7888d264 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -225,7 +225,7 @@ func (m *consumer) subscribe(ctx context.Context, controller Controller) error { m.subscriptions[topicKey] = sub // Spawn consumption goroutine - go m.consumeLoop(controllerCtx, controller, deliveryChan, done, config.BatchSize) + go m.consumeLoop(controllerCtx, controller, topicName, deliveryChan, done, config.BatchSize) m.logger.Infow("controller started", "controller", controller.Name(), @@ -258,13 +258,14 @@ func (m *consumer) subscribe(ctx context.Context, controller Controller) error { // Any messages buffered in partition channels but not processed before ctx // cancellation are safe to drop — the queue's visibility timeout will make // them visible again for redelivery (at-least-once semantics). -func (m *consumer) consumeLoop(ctx context.Context, controller Controller, deliveryChan <-chan extqueue.Delivery, done chan struct{}, batchSize int) { +func (m *consumer) consumeLoop(ctx context.Context, controller Controller, topicName string, deliveryChan <-chan extqueue.Delivery, done chan struct{}, batchSize int) { defer close(done) topicKey := controller.TopicKey() controllerScope := m.metricsScope.Tagged(map[string]string{ "controller": controller.Name(), + "topic": topicName, "topic_key": topicKey.String(), }) diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 6e2eb5d63..0480db1d7 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -642,6 +642,8 @@ func TestConsumer_Stop(t *testing.T) { } func TestConsumer_ObservabilityTags(t *testing.T) { + const topicName = "configured-test-topic" + tests := []struct { name string handlerError error @@ -712,7 +714,15 @@ func TestConsumer_ObservabilityTags(t *testing.T) { mockQ := queuemock.NewMockQueue(ctrl) mockQ.EXPECT().Subscriber().Return(mockSub) - reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") + reg, err := NewTopicRegistry([]TopicConfig{ + { + Key: testTopicKeyStart, + Name: topicName, + Queue: mockQ, + Subscription: extqueue.DefaultSubscriptionConfig("test-worker", "test-group"), + }, + }) + require.NoError(t, err) testC := New(logger, testScope, reg, tt.processor, consumergatenoop.New()) @@ -723,7 +733,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { }, ) - err := testC.Register(handler) + err = testC.Register(handler) require.NoError(t, err) ctx, cancel := context.WithCancel(context.Background()) @@ -753,6 +763,8 @@ func TestConsumer_ObservabilityTags(t *testing.T) { for key, value := range tt.expectedTags { assert.Equal(t, value, tags[key]) } + assert.Equal(t, topicName, tags["topic"]) + assert.Equal(t, testTopicKeyStart.String(), tags["topic_key"]) } } assert.True(t, foundLatency, "Should have process.finish metric")