feat(consumer): include configured topic name in queue consumer metrics

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(),
 	})
 

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")