feat(stovepipe): tag controller metrics by queue

platform/base/messagequeue/BUILD.bazel

@@ -2,14 +2,20 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["message.go"],
+    srcs = [
+        "context.go",
+        "message.go",
+    ],
     importpath = "github.com/uber/submitqueue/platform/base/messagequeue",
     visibility = ["//visibility:public"],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["message_test.go"],
+    srcs = [
+        "context_test.go",
+        "message_test.go",
+    ],
     embed = [":go_default_library"],
     deps = ["@com_github_stretchr_testify//assert:go_default_library"],
 )

platform/base/messagequeue/context.go

@@ -0,0 +1,38 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package messagequeue
+
+import (
+	"context"
+)
+
+// MetadataKeyQueueName carries the queue name independently of the
+// transport partition key. Producers set it on Message.Metadata so consumers
+// can attribute work before decoding the payload.
+const MetadataKeyQueueName = "queue_name"
+
+type queueNameContextKey struct{}
+
+// WithQueueName returns a child context containing the queue name of the
+// delivered message.
+func WithQueueName(ctx context.Context, queueName string) context.Context {
+	return context.WithValue(ctx, queueNameContextKey{}, queueName)
+}
+
+// QueueName returns the delivered message's queue name from ctx.
+func QueueName(ctx context.Context) (string, bool) {
+	queueName, ok := ctx.Value(queueNameContextKey{}).(string)
+	return queueName, ok
+}

platform/base/messagequeue/context_test.go

@@ -0,0 +1,33 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package messagequeue
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueueNameContext(t *testing.T) {
+	ctx := WithQueueName(context.Background(), "monorepo/main")
+
+	queueName, ok := QueueName(ctx)
+	assert.True(t, ok)
+	assert.Equal(t, "monorepo/main", queueName)
+
+	_, ok = QueueName(context.Background())
+	assert.False(t, ok)
+}

platform/consumer/BUILD.bazel

@@ -36,6 +36,7 @@ go_test(
         "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/messagequeue:go_default_library",
         "//platform/extension/messagequeue/mock:go_default_library",
+        "//platform/metrics:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",

platform/consumer/consumer.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 	"github.com/uber/submitqueue/platform/errs"
 	"github.com/uber/submitqueue/platform/extension/consumergate"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
@@ -371,6 +372,13 @@ func (m *consumer) processPartition(ctx context.Context, controller Controller,
 func (m *consumer) processDelivery(ctx context.Context, controller Controller, delivery extqueue.Delivery, controllerScope tally.Scope) {
 	const opName = "process"
 
+	msg := delivery.Message()
+	queueName := msg.Metadata[entityqueue.MetadataKeyQueueName]
+	ctx = entityqueue.WithQueueName(ctx, queueName)
+	if queueName != "" {
+		ctx = metrics.WithContextTags(ctx, metrics.NewTag("queue", queueName))
+	}
+
 	// Consumer gate: a delivery whose gate is closed is recorded as parked and
 	// postponed (barrier + re-check on redelivery); a false return also covers
 	// shutdown-while-checking, where the delivery is left in flight so its
@@ -380,7 +388,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		return
 	}
 
-	msg := delivery.Message()
 	topicKey := controller.TopicKey()
 
 	m.logger.Debugw("processing delivery",
@@ -396,7 +403,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 
 	// Call controller with wrapped delivery
 	start := time.Now()
-	op := metrics.Begin(controllerScope, opName, metrics.LongLatencyBuckets)
+	op := metrics.Begin(controllerScope, opName, metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...)
 	err := controller.Process(ctx, wrapped)
 
 	elapsed := time.Since(start)
@@ -419,7 +426,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		// A failure outcome wins over a recorded hold — a hold is only honored
 		// on success, so retry accounting and dead-lettering stay meaningful.
 		if wrapped.held {
-			metrics.NamedCounter(controllerScope, opName, "hold_ignored", 1)
+			metrics.NamedCounter(controllerScope, opName, "hold_ignored", 1, metrics.TagsFromContext(ctx)...)
 			m.logger.Warnw("hold recorded but controller returned error, failure outcome wins",
 				"controller", controller.Name(),
 				"topic_key", topicKey,
@@ -451,7 +458,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			)
 
 			// Reject moves to DLQ (or acks if DLQ disabled)
-			rejectOp := metrics.Begin(controllerScope, "reject", metrics.StorageLatencyBuckets)
+			rejectOp := metrics.Begin(controllerScope, "reject", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
 			rejectErr := delivery.Reject(ctx, controllerFailure)
 			rejectOp.Complete(rejectErr)
 			if rejectErr != nil {
@@ -485,7 +492,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		// Nack requeues immediately - the visibility timeout spaces retries.
 		// The failure travels with it so that the attempt which finally spends
 		// the retry budget can dead-letter saying why.
-		nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets)
+		nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
 		nackErr := delivery.Nack(ctx, controllerFailure)
 		nackOp.Complete(nackErr)
 		if nackErr != nil {
@@ -505,7 +512,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	// the visibility timeout lapses into a normal redelivery, so the hold
 	// loop's liveness never depends on this write succeeding.
 	if wrapped.held {
-		postponeOp := metrics.Begin(controllerScope, "postpone", metrics.StorageLatencyBuckets)
+		postponeOp := metrics.Begin(controllerScope, "postpone", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
 		postponeErr := delivery.Postpone(ctx, wrapped.holdDelayMs)
 		postponeOp.Complete(postponeErr)
 		if postponeErr != nil {
@@ -530,7 +537,7 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	}
 
 	// Controller succeeded - ack message
-	ackOp := metrics.Begin(controllerScope, "ack", metrics.StorageLatencyBuckets)
+	ackOp := metrics.Begin(controllerScope, "ack", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
 	ackErr := delivery.Ack(ctx)
 	ackOp.Complete(ackErr)
 	if ackErr != nil {
@@ -578,7 +585,7 @@ func (m *consumer) checkGate(ctx context.Context, controller Controller, deliver
 			// into a normal redelivery.
 			return false
 		}
-		metrics.NamedCounter(scope, opName, "enter_errors", 1)
+		metrics.NamedCounter(scope, opName, "enter_errors", 1, metrics.TagsFromContext(ctx)...)
 		m.logger.Errorw("gate check failed, failing open",
 			"consumer_group", consumerGroup,
 			"topic", topic,
@@ -600,7 +607,7 @@ func (m *consumer) checkGate(ctx context.Context, controller Controller, deliver
 		// earlier re-check, the gate has opened and the record must go so
 		// observers see an empty parked set. A no-op when never parked.
 		if unparkErr := entry.Unpark(ctx, descriptor); unparkErr != nil {
-			metrics.NamedCounter(scope, opName, "unpark_errors", 1)
+			metrics.NamedCounter(scope, opName, "unpark_errors", 1, metrics.TagsFromContext(ctx)...)
 			m.logger.Warnw("failed to remove parked record on admit",
 				"consumer_group", consumerGroup,
 				"topic", topic,
@@ -615,7 +622,7 @@ func (m *consumer) checkGate(ctx context.Context, controller Controller, deliver
 	// partition waits behind it (barrier) and the gate is re-checked on
 	// redelivery without burning retry budget.
 	if parkErr := entry.Park(ctx, descriptor); parkErr != nil {
-		metrics.NamedCounter(scope, opName, "park_errors", 1)
+		metrics.NamedCounter(scope, opName, "park_errors", 1, metrics.TagsFromContext(ctx)...)
 		m.logger.Warnw("failed to write parked record, postponing anyway",
 			"consumer_group", consumerGroup,
 			"topic", topic,
@@ -624,8 +631,8 @@ func (m *consumer) checkGate(ctx context.Context, controller Controller, deliver
 		)
 	}
 
-	metrics.NamedCounter(scope, opName, "parked", 1)
-	postponeOp := metrics.Begin(scope, "postpone", metrics.StorageLatencyBuckets)
+	metrics.NamedCounter(scope, opName, "parked", 1, metrics.TagsFromContext(ctx)...)
+	postponeOp := metrics.Begin(scope, "postpone", metrics.StorageLatencyBuckets, metrics.TagsFromContext(ctx)...)
 	postponeErr := delivery.Postpone(ctx, m.gateRecheckDelayMs)
 	postponeOp.Complete(postponeErr)
 	if postponeErr != nil {

platform/consumer/consumer_test.go

@@ -33,6 +33,7 @@ import (
 	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
 	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
 	queuemock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	"github.com/uber/submitqueue/platform/metrics"
 	"go.uber.org/mock/gomock"
 	"go.uber.org/zap/zaptest"
 )
@@ -286,10 +287,14 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) {
 	c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handledMsg := ""
+	handledQueue := ""
+	var handledTags []metrics.Tag
 	handler := &testController{}
 	setupController(handler, "test-handler", testTopicKeyStart, "test-group",
 		func(ctx context.Context, delivery Delivery) error {
 			handledMsg = delivery.Message().ID
+			handledQueue, _ = entityqueue.QueueName(ctx)
+			handledTags = metrics.TagsFromContext(ctx)
 			return nil
 		},
 	)
@@ -303,14 +308,18 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) {
 	err = c.Start(ctx)
 	require.NoError(t, err)
 
-	msg := entityqueue.NewMessage("test-msg-1", []byte("payload"), "partition1", nil)
+	msg := entityqueue.NewMessage("test-msg-1", []byte("payload"), "partition1", map[string]string{
+		entityqueue.MetadataKeyQueueName: "monorepo/main",
+	})
 	mockDel := queuemock.NewMockDelivery(ctrl)
 	done := setupDelivery(mockDel, msg, nil, nil)
 
 	deliveryChan <- mockDel
 	<-done
 
 	assert.Equal(t, "test-msg-1", handledMsg)
+	assert.Equal(t, "monorepo/main", handledQueue)
+	assert.Equal(t, []metrics.Tag{metrics.NewTag("queue", "monorepo/main")}, handledTags)
 
 	err = c.Stop(30000)
 	require.NoError(t, err)

platform/metrics/metrics.go

@@ -30,11 +30,36 @@ type Tag struct {
 	Value string
 }
 
+type contextTagsKey struct{}
+
 // NewTag creates a Tag with the given key and value.
 func NewTag(key, value string) Tag {
 	return Tag{Key: key, Value: value}
 }
 
+// WithContextTags returns a child context carrying the supplied metric tags.
+// Tags already carried by ctx are preserved, and both inputs are copied.
+func WithContextTags(ctx context.Context, tags ...Tag) context.Context {
+	existing, _ := ctx.Value(contextTagsKey{}).([]Tag)
+	contextTags := make([]Tag, 0, len(existing)+len(tags))
+	contextTags = append(contextTags, existing...)
+	contextTags = append(contextTags, tags...)
+	return context.WithValue(ctx, contextTagsKey{}, contextTags)
+}
+
+// TagsFromContext appends metric tags explicitly carried by ctx. Additional
+// tags are preserved, and context-derived tags win when the same key is present.
+func TagsFromContext(ctx context.Context, tags ...Tag) []Tag {
+	stored, _ := ctx.Value(contextTagsKey{}).([]Tag)
+	if len(stored) == 0 {
+		return tags
+	}
+
+	contextTags := make([]Tag, 0, len(tags)+len(stored))
+	contextTags = append(contextTags, tags...)
+	return append(contextTags, stored...)
+}
+
 // Common duration bucket sets for latency histograms. Operations differ widely
 // in expected latency, so there is no single default — pick the set whose range
 // matches the operation and pass it to Begin or NamedHistogram. Buckets

platform/metrics/metrics_test.go

@@ -158,6 +158,22 @@ func TestNamedGauge(t *testing.T) {
 	assert.Equal(t, float64(42), g.Value())
 }
 
+func TestTagsFromContext(t *testing.T) {
+	contextTags := []Tag{NewTag("queue", "monorepo/main")}
+	ctx := WithContextTags(context.Background(), contextTags...)
+	contextTags[0] = NewTag("queue", "changed")
+	ctx = WithContextTags(ctx, NewTag("controller", "build"))
+
+	tags := TagsFromContext(ctx, NewTag("result", "success"), NewTag("queue", "wrong"))
+	assert.Equal(t, []Tag{
+		NewTag("result", "success"),
+		NewTag("queue", "wrong"),
+		NewTag("queue", "monorepo/main"),
+		NewTag("controller", "build"),
+	}, tags)
+	assert.Empty(t, TagsFromContext(context.Background()))
+}
+
 func TestLatencyBuckets_Sorted(t *testing.T) {
 	sets := map[string]tally.DurationBuckets{
 		"FastLatencyBuckets":    FastLatencyBuckets,

platform/publish/publish.go

@@ -26,6 +26,7 @@ package publish
 import (
 	"context"
 	"fmt"
+	"maps"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -34,7 +35,8 @@ import (
 	"github.com/uber/submitqueue/platform/consumer"
 )
 
-// Message publishes payload to the topic registered for key.
+// Message publishes payload to the topic registered for key. Allowlisted
+// delivery context is propagated as message metadata.
 //
 // msgID selects the dedup behavior, so the caller must choose it deliberately.
 // The queue deduplicates on (topic, partition key, message ID) against every
@@ -53,6 +55,8 @@ func Message(ctx context.Context, registry consumer.TopicRegistry, key consumer.
 // MessageWithMetadata is Message with side-band message metadata (headers/attributes)
 // attached to the delivery. Use it to carry diagnostic context that is not part of
 // the payload — the backend persists and redelivers metadata alongside the message.
+// Allowlisted delivery context, currently only the queue name, is propagated unless
+// the caller supplies that metadata key explicitly.
 func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, key consumer.TopicKey, msgID string, payload []byte, partitionKey string, metadata map[string]string) error {
 	q, ok := registry.Queue(key)
 	if !ok {
@@ -63,10 +67,26 @@ func MessageWithMetadata(ctx context.Context, registry consumer.TopicRegistry, k
 		return fmt.Errorf("no topic name registered for topic key %s", key)
 	}
 
-	msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadata)
+	msg := entityqueue.NewMessage(msgID, payload, partitionKey, metadataFromContext(ctx, metadata))
 	return q.Publisher().Publish(ctx, topicName, msg)
 }
 
+func metadataFromContext(ctx context.Context, metadata map[string]string) map[string]string {
+	metadata = maps.Clone(metadata)
+	if _, exists := metadata[entityqueue.MetadataKeyQueueName]; exists {
+		return metadata
+	}
+	queueName, ok := entityqueue.QueueName(ctx)
+	if !ok || queueName == "" {
+		return metadata
+	}
+	if metadata == nil {
+		metadata = make(map[string]string)
+	}
+	metadata[entityqueue.MetadataKeyQueueName] = queueName
+	return metadata
+}
+
 // IntentID names the occasion to publish rather than the entity published
 // about: entityID says what the message concerns, and cause says why this
 // particular message exists.