feat(stovepipe): tag controller metrics by queue

Scope metrics from decoded pipeline messages to their logical queue so operators can filter controller health without splitting controller instances.

Author: mnoah1

stovepipe/controller/build/build.go

@@ -88,6 +88,7 @@ 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)
 	}
+	c = c.forQueue(br.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: br.GetQueueName()})
 	if err != nil {
@@ -192,3 +193,12 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) forQueue(queue string) *Controller {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}

stovepipe/controller/buildsignal/buildsignal.go

@@ -108,6 +108,7 @@ 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)
 	}
+	c = c.forQueue(sig.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: sig.GetQueueName()})
 	if err != nil {
@@ -376,3 +377,12 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) forQueue(queue string) *Controller {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}

stovepipe/controller/dlq/buildsignal.go

@@ -95,6 +95,7 @@ func (c *BuildSignalController) Process(ctx context.Context, delivery consumer.D
 		// without saying so.
 		return fmt.Errorf("failed to decode dlq payload: %w", err)
 	}
+	c = c.forQueue(sig.GetQueueName())
 	if sig.Id == "" {
 		metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "empty_id_errors", 1)
 		return fmt.Errorf("dlq payload decoded to empty build id")
@@ -165,3 +166,12 @@ func (c *BuildSignalController) TopicKey() consumer.TopicKey {
 func (c *BuildSignalController) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *BuildSignalController) forQueue(queue string) *BuildSignalController {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}

stovepipe/controller/dlq/request.go

@@ -83,6 +83,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 		// non-terminal.
 		return fmt.Errorf("failed to decode dlq payload: %w", err)
 	}
+	c = c.forQueue(pr.GetQueueName())
 	if pr.Id == "" {
 		metrics.NamedCounter(c.metricsScope, _opName, "empty_id_errors", 1)
 		return fmt.Errorf("dlq payload decoded to empty request id")
@@ -126,3 +127,12 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) forQueue(queue string) *Controller {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}

stovepipe/controller/ingest.go

@@ -94,7 +94,11 @@ func NewIngestController(
 func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest) (result entity.IngestResult, retErr error) {
 	const opName = "ingest"
 
-	op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets)
+	metricsScope := c.metricsScope
+	if req.Queue != "" {
+		metricsScope = metricsScope.Tagged(map[string]string{"queue": req.Queue})
+	}
+	op := metrics.Begin(metricsScope, opName, metrics.LongLatencyBuckets)
 	defer func() { op.Complete(retErr) }()
 
 	if req.Queue == "" {

stovepipe/controller/process/process.go

@@ -91,6 +91,7 @@ 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 process request: %w", err)
 	}
+	c = c.forQueue(pr.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: pr.GetQueueName()})
 	if err != nil {
@@ -489,3 +490,12 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) forQueue(queue string) *Controller {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}

stovepipe/controller/process/process_test.go

@@ -109,7 +109,7 @@ func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) *consumermo
 
 func processPayload(t *testing.T, id string) []byte {
 	t.Helper()
-	b, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id})
+	b, err := stovepipemq.Marshal(&stovepipemq.ProcessRequest{Id: id, QueueName: testQueue})
 	require.NoError(t, err)
 	return b
 }
@@ -313,7 +313,7 @@ func TestProcessEmitsAdmittedStrategyMetric(t *testing.T) {
 
 	require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))))
 
-	counter, ok := scope.Snapshot().Counters()["test.process_controller.process.admitted+strategy=full"]
+	counter, ok := scope.Snapshot().Counters()["test.process_controller.process.admitted+queue=monorepo/main,strategy=full"]
 	require.True(t, ok)
 	assert.Equal(t, int64(1), counter.Value())
 }
@@ -338,7 +338,7 @@ func TestProcessEmitsSourceControlResolutionMetric(t *testing.T) {
 
 	require.Error(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))))
 
-	counter, ok := scope.Snapshot().Counters()["test.process_controller.process.source_control_errors+stage=resolve"]
+	counter, ok := scope.Snapshot().Counters()["test.process_controller.process.source_control_errors+queue=monorepo/main,stage=resolve"]
 	require.True(t, ok)
 	assert.Equal(t, int64(1), counter.Value())
 }

stovepipe/controller/record/record.go

@@ -99,6 +99,7 @@ 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 record: %w", err)
 	}
+	c = c.forQueue(rec.GetQueueName())
 
 	store, err := c.stores.For(storage.Config{QueueName: rec.GetQueueName()})
 	if err != nil {
@@ -489,3 +490,12 @@ func (c *Controller) TopicKey() consumer.TopicKey {
 func (c *Controller) ConsumerGroup() string {
 	return c.consumerGroup
 }
+
+func (c *Controller) forQueue(queue string) *Controller {
+	if queue == "" {
+		return c
+	}
+	scoped := *c
+	scoped.metricsScope = c.metricsScope.Tagged(map[string]string{"queue": queue})
+	return &scoped
+}