refactor: use operation metrics for controller results

Author: sbalabanov-zz

platform/consumer/README.md

@@ -118,7 +118,7 @@ func (c *MyController) Process(ctx context.Context, delivery consumer.Delivery)
 
 When the consumer is wired with `errs.AlwaysRetryableProcessor` (DLQ reconciliation), the framework overrides this: every non-nil error is forced retryable so the DLQ message comes back for another attempt. See `submitqueue/orchestrator/controller/dlq/README.md`.
 
-The consumer records `process.controller_latency` after the error processor runs. Every series has `result=success|error|cancel`; error and cancellation series also include `error_origin=user|infra`, `retryable=true|false`, and `dependency=true|false`. These dimensions therefore describe the processed error that drives ack, nack, or reject behavior rather than the controller's raw return value.
+The consumer records controller operations with `process.start` and `process.finish`. The finish histogram records both latency and completion count with `result=success|error|cancel`; error and cancellation series also include `origin=infra|infra_retryable|user` and `dependency=yes|no`. These dimensions are added after error processing, so they describe the classified error that drives ack, nack, or reject behavior rather than the controller's raw return value. The lifecycle histogram count replaces separate received, processed, and controller-error counters.
 
 ## Lifecycle
 

platform/consumer/consumer.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strconv"
 	"sync"
 	"time"
 
@@ -342,9 +341,6 @@ 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"
 
-	start := time.Now()
-	metrics.NamedCounter(controllerScope, opName, "messages_received", 1)
-
 	msg := delivery.Message()
 	topicKey := controller.TopicKey()
 
@@ -360,27 +356,25 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	wrapped := &deliveryWrapper{delivery: delivery}
 
 	// Call controller with wrapped delivery
+	start := time.Now()
+	op := metrics.Begin(controllerScope, opName, metrics.LongLatencyBuckets)
 	err := controller.Process(ctx, wrapped)
 
 	elapsed := time.Since(start)
 
+	var completionTags []metrics.Tag
 	if err != nil {
 		// Single explicit classification pass through the configured
 		// ErrorProcessor. Primary consumers use a classifier-based processor
 		// (preserves controller framework wraps); DLQ consumers use the
 		// always-retryable processor (forces redelivery on any error).
 		err = m.processor.Process(err)
+		completionTags = controllerClassificationTags(err)
 	}
 
-	// Record controller latency only after classification so error dimensions
-	// reflect the verdict used for ack/nack/reject behavior.
-	metrics.NamedHistogram(
-		controllerScope,
-		opName,
-		"controller_latency",
-		metrics.LongLatencyBuckets,
-		controllerResultTags(err)...,
-	).RecordDuration(elapsed)
+	// Complete only after classification so the finish histogram carries the
+	// verdict used for ack/nack/reject behavior.
+	op.Complete(err, completionTags...)
 
 	if err != nil {
 		// By convention, Controller can only return context.Canceled if it is
@@ -399,8 +393,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 				"elapsed_ms", elapsed.Milliseconds(),
 			)
 
-			metrics.NamedCounter(controllerScope, opName, "non_retryable_errors", 1)
-
 			// Reject moves to DLQ (or acks if DLQ disabled)
 			if rejectErr := delivery.Reject(ctx, err.Error()); rejectErr != nil {
 				m.logger.Errorw("failed to reject non-retryable message",
@@ -431,8 +423,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 			"elapsed_ms", elapsed.Milliseconds(),
 		)
 
-		metrics.NamedCounter(controllerScope, opName, "controller_errors", 1)
-
 		// Nack with no delay - let visibility timeout handle retry delay
 		nackStart := time.Now()
 		if nackErr := delivery.Nack(ctx, 0); nackErr != nil {
@@ -470,7 +460,6 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 		return
 	}
 
-	metrics.NamedCounter(controllerScope, opName, "messages_processed", 1)
 	metrics.NamedCounter(controllerScope, opName, "ack_count", 1)
 	metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets,
 		metrics.NewTag("operation", "ack"),
@@ -487,27 +476,22 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d
 	)
 }
 
-func controllerResultTags(err error) []metrics.Tag {
-	result := "success"
-	if err == nil {
-		return []metrics.Tag{metrics.NewTag("result", result)}
-	}
-
-	result = "error"
-	if errors.Is(err, context.Canceled) {
-		result = "cancel"
-	}
-
+func controllerClassificationTags(err error) []metrics.Tag {
 	origin := "infra"
-	if errs.IsUserError(err) {
+	if errs.IsRetryable(err) {
+		origin = "infra_retryable"
+	} else if errs.IsUserError(err) {
 		origin = "user"
 	}
 
+	dependency := "no"
+	if errs.IsDependencyError(err) {
+		dependency = "yes"
+	}
+
 	return []metrics.Tag{
-		metrics.NewTag("result", result),
-		metrics.NewTag("error_origin", origin),
-		metrics.NewTag("retryable", strconv.FormatBool(errs.IsRetryable(err))),
-		metrics.NewTag("dependency", strconv.FormatBool(errs.IsDependencyError(err))),
+		metrics.NewTag("origin", origin),
+		metrics.NewTag("dependency", dependency),
 	}
 }
 

platform/consumer/consumer_test.go

@@ -459,10 +459,9 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 				verdict: errs.InfraDependencyRetryable,
 			}),
 			expectedTags: map[string]string{
-				"result":       "error",
-				"error_origin": "infra",
-				"retryable":    "true",
-				"dependency":   "true",
+				"result":     "error",
+				"origin":     "infra_retryable",
+				"dependency": "yes",
 			},
 			expectAckCount: false,
 		},
@@ -474,10 +473,9 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 				verdict: errs.InfraRetryable,
 			}),
 			expectedTags: map[string]string{
-				"result":       "cancel",
-				"error_origin": "infra",
-				"retryable":    "true",
-				"dependency":   "false",
+				"result":     "cancel",
+				"origin":     "infra_retryable",
+				"dependency": "no",
 			},
 			expectAckCount: false,
 		},
@@ -530,18 +528,27 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 
 			var foundLatency bool
 			for _, histogram := range histograms {
-				if strings.Contains(histogram.Name(), "controller_latency") {
+				if strings.Contains(histogram.Name(), "process.finish") {
 					foundLatency = true
 					tags := histogram.Tags()
 					for key, value := range tt.expectedTags {
 						assert.Equal(t, value, tags[key])
 					}
-					assert.NotContains(t, tags, "success")
 				}
 			}
-			assert.True(t, foundLatency, "Should have controller_latency metric")
+			assert.True(t, foundLatency, "Should have process.finish metric")
 
 			counters := snapshot.Counters()
+			for _, duplicate := range []string{
+				"messages_received",
+				"messages_processed",
+				"non_retryable_errors",
+				"controller_errors",
+			} {
+				for _, counter := range counters {
+					assert.NotContains(t, counter.Name(), duplicate)
+				}
+			}
 			if tt.expectAckCount {
 				var foundAck bool
 				for _, counter := range counters {
@@ -558,52 +565,50 @@ func TestConsumer_ObservabilityTags(t *testing.T) {
 	}
 }
 
-func TestControllerResultTags(t *testing.T) {
+func TestControllerClassificationTags(t *testing.T) {
 	tests := []struct {
 		name     string
 		err      error
 		expected map[string]string
 	}{
 		{
-			name:     "success",
-			expected: map[string]string{"result": "success"},
+			name: "infra error",
+			err:  fmt.Errorf("failed"),
+			expected: map[string]string{
+				"origin":     "infra",
+				"dependency": "no",
+			},
 		},
 		{
 			name: "user error",
 			err:  errs.NewUserError(fmt.Errorf("invalid request")),
 			expected: map[string]string{
-				"result":       "error",
-				"error_origin": "user",
-				"retryable":    "false",
-				"dependency":   "false",
+				"origin":     "user",
+				"dependency": "no",
 			},
 		},
 		{
 			name: "retryable dependency error",
 			err:  errs.NewRetryableDependencyError(fmt.Errorf("database unavailable")),
 			expected: map[string]string{
-				"result":       "error",
-				"error_origin": "infra",
-				"retryable":    "true",
-				"dependency":   "true",
+				"origin":     "infra_retryable",
+				"dependency": "yes",
 			},
 		},
 		{
 			name: "cancellation",
 			err:  errs.NewRetryableError(context.Canceled),
 			expected: map[string]string{
-				"result":       "cancel",
-				"error_origin": "infra",
-				"retryable":    "true",
-				"dependency":   "false",
+				"origin":     "infra_retryable",
+				"dependency": "no",
 			},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			actual := make(map[string]string)
-			for _, tag := range controllerResultTags(tt.err) {
+			for _, tag := range controllerClassificationTags(tt.err) {
 				actual[tag.Key] = tag.Value
 			}
 			assert.Equal(t, tt.expected, actual)

platform/metrics/README.md

@@ -8,7 +8,7 @@ The `metrics` package provides reusable helpers for emitting counters and histog
 
 **Operation lifecycle** — `Begin` and `Complete` tie operation metrics together. `Begin` captures the start time and emits `{name}.start`; `Complete` records duration and count on `{name}.finish`.
 
-**Result tagging** — the finish histogram is tagged with `result=success`, `result=error`, or `result=cancel`. Cancellation is detected with `errors.Is(err, context.Canceled)`. Error classification tags are intentionally omitted because many call sites complete before classification occurs.
+**Result tagging** — the finish histogram is tagged with `result=success`, `result=error`, or `result=cancel`. Cancellation is detected with `errors.Is(err, context.Canceled)`. Callers that accumulate tags while the operation runs can pass them to `Complete`.
 
 **Consistent naming** — named helpers follow the `{name}.{sub}` sub-scope pattern, producing metric paths such as `process.start` and `publish.attempts`.
 
@@ -19,7 +19,7 @@ For any operation with a clear start and end, use `Begin` and `Complete`:
 | Function | Emits |
 |----------|-------|
 | `Begin(scope, name, buckets, ...tags)` | `{name}.start` counter +1 and returns an `Op` |
-| `op.Complete(err)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` |
+| `op.Complete(err, ...tags)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` and any completion tags |
 
 `buckets` is required at `Begin` because operations differ widely in expected latency. The finish histogram records both the duration distribution and the number of completed operations, so `Complete` does not emit a separate counter.
 
@@ -33,6 +33,14 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 }
 ```
 
+Tags passed to `Begin` apply to both lifecycle metrics. Tags known only after execution, such as an error classification, can be attached to the finish histogram:
+
+```go
+err := controller.Process(ctx, delivery)
+err = classifier.Process(err)
+op.Complete(err, metrics.NewTag("origin", "infra_retryable"))
+```
+
 ## Named Helpers
 
 For ad-hoc metrics that do not fit the operation lifecycle:

platform/metrics/metrics.go

@@ -141,9 +141,10 @@ func Begin(scope tally.Scope, name string, buckets tally.Buckets, tags ...Tag) O
 }
 
 // Complete records elapsed time on the {name}.finish histogram, tagged with
-// result=success|error|cancel. The histogram records both duration and count.
-// Cancellation is detected through the error chain.
-func (o Op) Complete(err error) {
+// result=success|error|cancel and any additional tags accumulated while the
+// operation ran. The histogram records both duration and count. Cancellation
+// is detected through the error chain.
+func (o Op) Complete(err error, tags ...Tag) {
 	result := "success"
 	if err != nil {
 		result = "error"
@@ -152,7 +153,7 @@ func (o Op) Complete(err error) {
 		}
 	}
 
-	o.scope.
+	tagged(o.scope, tags).
 		Tagged(map[string]string{"result": result}).
 		Histogram("finish", o.buckets).
 		RecordDuration(time.Since(o.start))

platform/metrics/metrics_test.go

@@ -106,6 +106,25 @@ func TestBegin_WithTags(t *testing.T) {
 	assert.True(t, ok, "expected tagged finish histogram, got keys: %v", histogramKeys(histograms))
 }
 
+func TestOp_CompleteWithTags(t *testing.T) {
+	scope := tally.NewTestScope("", nil)
+	op := Begin(scope, "process", FastLatencyBuckets)
+	op.Complete(
+		fmt.Errorf("failed"),
+		NewTag("origin", "infra_retryable"),
+		NewTag("dependency", "no"),
+	)
+
+	snapshot := scope.Snapshot()
+	counters := snapshot.Counters()
+	_, ok := counters["process.start+"]
+	assert.True(t, ok, "finish-only tags should not be applied to start")
+
+	histograms := snapshot.Histograms()
+	_, ok = histograms["process.finish+dependency=no,origin=infra_retryable,result=error"]
+	assert.True(t, ok, "expected finish-only tags, got keys: %v", histogramKeys(histograms))
+}
+
 func TestNamedCounter(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
 	NamedCounter(scope, "publish", "attempts", 5)