feat(stovepipe): tag controller metrics by queue

Summary:
Intent:
- Make controller metrics attributable to their logical queue.
- Keep queue tagging request-local without mutating shared controllers.

Changes:
- Pass queue as an ordinary metrics tag through primary and DLQ controller paths.
- Reuse the existing package-level metric helpers and controller Tally scopes.
- Keep pre-decode failures untagged and cover queue tags in tests.

Test Plan:
- Run targeted Bazel tests for platform metrics and the affected primary and DLQ controllers.

Revert Plan:
- Revert this commit to restore the prior 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

stovepipe/controller/build/build.go

@@ -88,24 +88,25 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		// Non-retryable: a malformed message will never succeed regardless of retries.
 		return fmt.Errorf("failed to deserialize build request: %w", err)
 	}
+	queueTag := metrics.NewTag("queue", br.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: br.GetQueueName()})
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1, queueTag)
 		// Non-retryable: a missing or unresolvable queue is a malformed message.
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", br.GetQueueName(), err)
 	}
 
 	request, err := c.loadRequest(ctx, store, br.Id)
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, queueTag)
 		return err
 	}
 
 	// The payload's queue must match the request's authoritative queue; a
 	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
 	if br.GetQueueName() != "" && br.GetQueueName() != request.Queue {
-		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1, queueTag)
 		return fmt.Errorf("payload queue %q does not match queue %q of request %s", br.GetQueueName(), request.Queue, request.ID)
 	}
 
@@ -123,7 +124,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 
 	// process decided the scope; build never re-derives incremental-vs-full.
 	if request.BuildStrategy == entity.BuildStrategyUnknown {
-		metrics.NamedCounter(c.metricsScope, _opName, "strategy_not_visible", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "strategy_not_visible", 1, queueTag)
 		return errs.NewRetryableError(fmt.Errorf("request %s has no build strategy yet", request.ID))
 	}
 	baseURI := ""

stovepipe/controller/build/build_test.go

@@ -52,6 +52,7 @@ type buildMocks struct {
 	runnerFactory *buildrunnermock.MockFactory
 	runner        *buildrunnermock.MockBuildRunner
 	publisher     *mqmock.MockPublisher
+	metricsScope  tally.TestScope
 }
 
 // staticStorageFactory resolves every queue to one fixed store aggregate.
@@ -63,12 +64,14 @@ func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { ret
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMocks) {
 	t.Helper()
 
+	scope := tally.NewTestScope("test", nil)
 	m := buildMocks{
 		reqStore:      storagemock.NewMockRequestStore(ctrl),
 		buildStore:    storagemock.NewMockBuildStore(ctrl),
 		runnerFactory: buildrunnermock.NewMockFactory(ctrl),
 		runner:        buildrunnermock.NewMockBuildRunner(ctrl),
 		publisher:     mqmock.NewMockPublisher(ctrl),
+		metricsScope:  scope,
 	}
 
 	store := storagemock.NewMockStorage(ctrl)
@@ -83,10 +86,23 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildMoc
 	})
 	require.NoError(t, err)
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
+	c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuild, "stovepipe-build")
 	return c, m
 }
 
+func TestProcessTagsMetricsWithQueue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+	m.reqStore.EXPECT().Get(gomock.Any(), testID).
+		Return(processingRequest(entity.BuildStrategyUnknown, ""), nil)
+	m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil)
+
+	require.Error(t, c.Process(context.Background(), delivery(t, ctrl, buildPayload(t, testID))))
+	counter, ok := m.metricsScope.Snapshot().Counters()["test.build_controller.build.strategy_not_visible+queue=monorepo/main"]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, counter.Value())
+}
+
 func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery {
 	t.Helper()
 	d := consumermock.NewMockDelivery(ctrl)
@@ -97,7 +113,7 @@ func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.De
 
 func buildPayload(t *testing.T, id string) []byte {
 	t.Helper()
-	b, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id})
+	b, err := stovepipemq.Marshal(&stovepipemq.BuildRequest{Id: id, QueueName: testQueue})
 	require.NoError(t, err)
 	return b
 }

stovepipe/controller/buildsignal/buildsignal.go

@@ -108,30 +108,31 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		// Non-retryable: a malformed message will never succeed regardless of retries.
 		return fmt.Errorf("failed to deserialize build signal: %w", err)
 	}
+	queueTag := metrics.NewTag("queue", sig.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: sig.GetQueueName()})
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_resolve_errors", 1, queueTag)
 		// Non-retryable: a missing or unresolvable queue is a malformed message.
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", sig.GetQueueName(), err)
 	}
 
 	build, err := c.loadBuild(ctx, store, sig.Id)
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, queueTag)
 		return err
 	}
 
 	request, err := c.loadRequest(ctx, store, build.RequestID)
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, queueTag)
 		return err
 	}
 
 	// The payload's queue must match the request's authoritative queue; a
 	// mismatch is a malformed message. Non-retryable — reject to the DLQ.
 	if sig.GetQueueName() != "" && sig.GetQueueName() != request.Queue {
-		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "queue_mismatch", 1, queueTag)
 		return fmt.Errorf("payload queue %q does not match queue %q of request %s", sig.GetQueueName(), request.Queue, request.ID)
 	}
 
@@ -167,7 +168,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 	}
 
 	if effective.IsTerminal() {
-		if err := c.finishRequest(ctx, store, &request, effective); err != nil {
+		if err := c.finishRequest(ctx, queueTag, store, &request, effective); err != nil {
 			return err
 		}
 		if err := c.publishRecord(ctx, request.ID, request.Queue); err != nil {
@@ -208,18 +209,18 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 // the request non-terminal, so redelivery re-runs both steps and decrements again
 // — transiently over-admitting by one until releaseBuildSlot's zero clamp
 // reconverges, which is the failure mode this pipeline prefers.
-func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, status entity.BuildStatus) error {
+func (c *Controller) finishRequest(ctx context.Context, queueTag metrics.Tag, store storage.Storage, request *entity.Request, status entity.BuildStatus) error {
 	if request.State.HasBuildOutcome() {
 		return nil
 	}
 
-	if err := c.releaseBuildSlot(ctx, store, request.Queue); err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
+	if err := c.releaseBuildSlot(ctx, queueTag, store, request.Queue); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, queueTag)
 		return err
 	}
 
-	if err := c.markOutcome(ctx, store, request, outcomeState(status)); err != nil {
-		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1)
+	if err := c.markOutcome(ctx, queueTag, store, request, outcomeState(status)); err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, queueTag)
 		return err
 	}
 	return nil
@@ -243,7 +244,7 @@ func outcomeState(status entity.BuildStatus) entity.RequestState {
 // conflicts. First writer wins: once any outcome is recorded a later caller leaves it
 // alone, so duplicate builds for one request (which build.md accepts) cannot flip the
 // verdict back and forth.
-func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, request *entity.Request, state entity.RequestState) error {
+func (c *Controller) markOutcome(ctx context.Context, queueTag metrics.Tag, store storage.Storage, request *entity.Request, state entity.RequestState) error {
 	reqStore := store.GetRequestStore()
 
 	for {
@@ -268,6 +269,7 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req
 		updated.Version = newVersion
 		*request = updated
 		metrics.NamedCounter(c.metricsScope, _opName, "outcomes", 1,
+			queueTag,
 			metrics.NewTag("state", string(state)),
 		)
 		return nil
@@ -279,7 +281,7 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req
 // (preserving concurrent updates), clamps at zero, and retries on version conflicts.
 // Unlike process's unwind-path release this is not best-effort: the caller must not
 // mark the request terminal if the slot was not freed, so a hard failure is returned.
-func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) error {
+func (c *Controller) releaseBuildSlot(ctx context.Context, queueTag metrics.Tag, store storage.Storage, queueName string) error {
 	queueStore := store.GetQueueStore()
 
 	for {
@@ -300,7 +302,7 @@ func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage
 			}
 			return fmt.Errorf("failed to release build slot for queue %s: %w", queueName, err)
 		}
-		metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1)
+		metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1, queueTag)
 		return nil
 	}
 }

stovepipe/controller/buildsignal/buildsignal_test.go

@@ -52,6 +52,7 @@ type buildsignalMocks struct {
 	runnerFactory *buildrunnermock.MockFactory
 	runner        *buildrunnermock.MockBuildRunner
 	publisher     *mqmock.MockPublisher
+	metricsScope  tally.TestScope
 }
 
 // staticStorageFactory resolves every queue to one fixed store aggregate.
@@ -63,13 +64,15 @@ func (f staticStorageFactory) For(storage.Config) (storage.Storage, error) { ret
 func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsignalMocks) {
 	t.Helper()
 
+	scope := tally.NewTestScope("test", nil)
 	m := buildsignalMocks{
 		reqStore:      storagemock.NewMockRequestStore(ctrl),
 		buildStore:    storagemock.NewMockBuildStore(ctrl),
 		queueStore:    storagemock.NewMockQueueStore(ctrl),
 		runnerFactory: buildrunnermock.NewMockFactory(ctrl),
 		runner:        buildrunnermock.NewMockBuildRunner(ctrl),
 		publisher:     mqmock.NewMockPublisher(ctrl),
+		metricsScope:  scope,
 	}
 
 	store := storagemock.NewMockStorage(ctrl)
@@ -86,10 +89,21 @@ func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, buildsig
 	})
 	require.NoError(t, err)
 
-	c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
+	c := NewController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: store}, m.runnerFactory, registry, stovepipemq.TopicKeyBuildSignal, "stovepipe-buildsignal")
 	return c, m
 }
 
+func TestProcessTagsMetricsWithQueue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	c, m := newController(t, ctrl)
+	m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(entity.Build{}, assert.AnError)
+
+	require.Error(t, c.Process(context.Background(), delivery(t, ctrl, buildSignalPayload(t, testBuildID))))
+	counter, ok := m.metricsScope.Snapshot().Counters()["test.buildsignal_controller.buildsignal.storage_errors+queue=monorepo/main"]
+	require.True(t, ok)
+	assert.EqualValues(t, 1, counter.Value())
+}
+
 func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermock.MockDelivery {
 	t.Helper()
 	d := consumermock.NewMockDelivery(ctrl)
@@ -100,7 +114,7 @@ func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermo
 
 func buildSignalPayload(t *testing.T, id string) []byte {
 	t.Helper()
-	b, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: id})
+	b, err := stovepipemq.Marshal(&stovepipemq.BuildSignal{Id: id, QueueName: testQueue})
 	require.NoError(t, err)
 	return b
 }

stovepipe/controller/dlq/build.go

@@ -65,14 +65,15 @@ func (c *buildController) Process(ctx context.Context, delivery consumer.Deliver
 		metrics.NamedCounter(c.metricsScope, _buildOpName, "deserialize_errors", 1)
 		return fmt.Errorf("failed to decode dlq payload: %w", err)
 	}
+	queueTag := metrics.NewTag("queue", buildRequest.GetQueueName())
 	if buildRequest.Id == "" {
-		metrics.NamedCounter(c.metricsScope, _buildOpName, "empty_id_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _buildOpName, "empty_id_errors", 1, queueTag)
 		return fmt.Errorf("build dlq payload decoded to empty request id")
 	}
 
 	store, err := c.stores.For(storage.Config{QueueName: buildRequest.GetQueueName()})
 	if err != nil {
-		metrics.NamedCounter(c.metricsScope, _buildOpName, "storage_resolve_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _buildOpName, "storage_resolve_errors", 1, queueTag)
 		return fmt.Errorf("failed to resolve storage for queue %q: %w", buildRequest.GetQueueName(), err)
 	}
 
@@ -86,10 +87,10 @@ func (c *buildController) Process(ctx context.Context, delivery consumer.Deliver
 	)
 
 	if err := failRequest(ctx, store, c.logger, buildRequest.Id); err != nil {
-		metrics.NamedCounter(c.metricsScope, _buildOpName, "reconcile_errors", 1)
+		metrics.NamedCounter(c.metricsScope, _buildOpName, "reconcile_errors", 1, queueTag)
 		return err
 	}
-	metrics.NamedCounter(c.metricsScope, _buildOpName, "reconciled", 1)
+	metrics.NamedCounter(c.metricsScope, _buildOpName, "reconciled", 1, queueTag)
 	return nil
 }
 

stovepipe/controller/dlq/build_test.go

@@ -32,15 +32,17 @@ import (
 func newBuildController(t *testing.T, ctrl *gomock.Controller) (consumer.Controller, dlqMocks) {
 	t.Helper()
 
+	scope := tally.NewTestScope("test", nil)
 	m := dlqMocks{
-		reqStore:   storagemock.NewMockRequestStore(ctrl),
-		queueStore: storagemock.NewMockQueueStore(ctrl),
+		reqStore:     storagemock.NewMockRequestStore(ctrl),
+		queueStore:   storagemock.NewMockQueueStore(ctrl),
+		metricsScope: scope,
 	}
 	store := storagemock.NewMockStorage(ctrl)
 	store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
 	store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
 
-	c := NewDLQBuildController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq")
+	c := NewDLQBuildController(zap.NewNop().Sugar(), scope, staticStorageFactory{store: store}, TopicKey(stovepipemq.TopicKeyBuild), "stovepipe-build-dlq")
 	return c, m
 }
 
@@ -53,10 +55,11 @@ func buildPayload(t *testing.T, id string) []byte {
 
 func TestBuildProcess(t *testing.T) {
 	tests := []struct {
-		name    string
-		payload []byte
-		setup   func(m dlqMocks)
-		wantErr bool
+		name       string
+		payload    []byte
+		setup      func(m dlqMocks)
+		wantErr    bool
+		wantMetric string
 	}{
 		{
 			name: "processing request is failed",
@@ -68,9 +71,15 @@ func TestBuildProcess(t *testing.T) {
 				updated.State = entity.RequestStateFailed
 				m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil)
 			},
+			wantMetric: "test.build_dlq_controller.build_dlq.reconciled+queue=monorepo/main",
 		},
 		{name: "malformed payload is returned", payload: []byte("not-a-proto"), wantErr: true},
-		{name: "empty request id is returned", payload: buildPayload(t, ""), wantErr: true},
+		{
+			name:       "empty request id is returned",
+			payload:    buildPayload(t, ""),
+			wantErr:    true,
+			wantMetric: "test.build_dlq_controller.build_dlq.empty_id_errors+queue=monorepo/main",
+		},
 	}
 
 	for _, tt := range tests {
@@ -86,6 +95,11 @@ func TestBuildProcess(t *testing.T) {
 				payload = buildPayload(t, testID)
 			}
 			err := controller.Process(context.Background(), delivery(t, ctrl, payload))
+			if tt.wantMetric != "" {
+				counter, ok := mocks.metricsScope.Snapshot().Counters()[tt.wantMetric]
+				require.True(t, ok)
+				assert.EqualValues(t, 1, counter.Value())
+			}
 			if tt.wantErr {
 				require.Error(t, err)
 				return