feat(stovepipe): tag controller metrics by queue

Summary:
Attribute Stovepipe controller metrics to the logical queue carried with each message.

Intent:
- Make primary and DLQ controller metrics filterable by queue.
- Keep message-derived metric context request-local and narrowly scoped.

Changes:
- Publish queue_name in Message.Metadata at each Stovepipe message handoff.
- Copy only queue_name from delivered metadata into context.
- Add metrics.TagsFromContext to append allowlisted context tags and migrate affected metrics.
- Preserve payload queue fields as the authority for storage and validation, and omit the tag for older messages without metadata.

Test Plan:
- ./tool/bazel test //platform/base/messagequeue:go_default_test //platform/consumer:go_default_test //platform/metrics:go_default_test //stovepipe/controller:go_default_test //stovepipe/controller/build:go_default_test //stovepipe/controller/buildsignal:go_default_test //stovepipe/controller/dlq:go_default_test //stovepipe/controller/process:go_default_test //stovepipe/controller/record:go_default_test

Revert Plan:
- Revert this commit to restore the prior untagged controller metrics.

---

<sub>Generated by the 🪄 [pr-create](https://sg.uberinternal.com/code.uber.internal/uber-code/devexp-agent-marketplace/-/blob/claude-code/plugins/dev/uber-dev/skills/pr-create/SKILL.md) skill in devexp-agent-marketplace</sub>

Author: mnoah1

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 logical 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 logical 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 logical 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/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,9 @@ 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()
+	ctx = entityqueue.WithQueueName(ctx, msg.Metadata[entityqueue.MetadataKeyQueueName])
+
 	// 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 +384,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 +399,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 +422,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 +454,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 +488,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 +508,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 +533,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 +581,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 +603,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 +618,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 +627,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

@@ -286,10 +286,12 @@ func TestConsumer_ProcessDelivery_Success(t *testing.T) {
 	c := New(logger, tally.NoopScope, reg, errs.NewClassifierProcessor(), consumergatenoop.New())
 
 	handledMsg := ""
+	handledQueue := ""
 	handler := &testController{}
 	setupController(handler, "test-handler", testTopicKeyStart, "test-group",
 		func(ctx context.Context, delivery Delivery) error {
 			handledMsg = delivery.Message().ID
+			handledQueue, _ = entityqueue.QueueName(ctx)
 			return nil
 		},
 	)
@@ -303,14 +305,17 @@ 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)
 
 	err = c.Stop(30000)
 	require.NoError(t, err)

platform/metrics/BUILD.bazel

@@ -5,14 +5,18 @@ go_library(
     srcs = ["metrics.go"],
     importpath = "github.com/uber/submitqueue/platform/metrics",
     visibility = ["//visibility:public"],
-    deps = ["@com_github_uber_go_tally//:go_default_library"],
+    deps = [
+        "//platform/base/messagequeue:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
+    ],
 )
 
 go_test(
     name = "go_default_test",
     srcs = ["metrics_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
     ],

platform/metrics/metrics.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 )
 
 // Tag is a key-value pair attached to a metric for dimensional filtering.
@@ -35,6 +36,21 @@ func NewTag(key, value string) Tag {
 	return Tag{Key: key, Value: value}
 }
 
+// TagsFromContext appends the allowlisted metric tags carried by ctx. Additional
+// tags are preserved, and context-derived tags win when the same key is present.
+// Messages published before a context value was introduced return only the
+// additional tags.
+func TagsFromContext(ctx context.Context, tags ...Tag) []Tag {
+	queueName, ok := entityqueue.QueueName(ctx)
+	if !ok || queueName == "" {
+		return tags
+	}
+
+	contextTags := make([]Tag, 0, len(tags)+1)
+	contextTags = append(contextTags, tags...)
+	return append(contextTags, NewTag("queue", queueName))
+}
+
 // 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

@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/uber-go/tally"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
 )
 
 func TestBegin_EmitsStart(t *testing.T) {
@@ -158,6 +159,19 @@ func TestNamedGauge(t *testing.T) {
 	assert.Equal(t, float64(42), g.Value())
 }
 
+func TestTagsFromContext(t *testing.T) {
+	ctx := entityqueue.WithQueueName(context.Background(), "monorepo/main")
+
+	tags := TagsFromContext(ctx, NewTag("result", "success"), NewTag("queue", "wrong"))
+	assert.Equal(t, []Tag{
+		NewTag("result", "success"),
+		NewTag("queue", "wrong"),
+		NewTag("queue", "monorepo/main"),
+	}, tags)
+	assert.Empty(t, TagsFromContext(context.Background()))
+	assert.Empty(t, TagsFromContext(entityqueue.WithQueueName(context.Background(), "")))
+}
+
 func TestLatencyBuckets_Sorted(t *testing.T) {
 	sets := map[string]tally.DurationBuckets{
 		"FastLatencyBuckets":    FastLatencyBuckets,

stovepipe/controller/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//api/stovepipe/protopb:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/extension/counter:go_default_library",
@@ -33,6 +34,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//api/stovepipe/protopb:go_default_library",
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/extension/counter:go_default_library",
         "//platform/extension/counter/mock:go_default_library",

stovepipe/controller/build/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/build",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/errs:go_default_library",
         "//platform/metrics:go_default_library",