diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 68b2603c..65f9d373 100644 --- a/platform/consumer/README.md +++ b/platform/consumer/README.md @@ -118,6 +118,8 @@ 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 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 1. **Register** controllers before starting. diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index f5d4c773..24bc7f24 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -341,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() @@ -359,31 +356,30 @@ 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) - // By convention, Controller can only return context.Canceled if it is cancelled by the context, i.e. when consumer is stopped or application is shutting down - isCanceled := errors.Is(err, context.Canceled) - - // Track latency with success/failure tags - successTag := "true" - if err != nil { - if isCanceled { - successTag = "cancel" - } else { - successTag = "false" - } - } - - metrics.NamedHistogram(controllerScope, opName, "controller_latency", metrics.LongLatencyBuckets, metrics.NewTag("success", successTag)).RecordDuration(elapsed) - + 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) + } + + // 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 + // cancelled by the processing context during shutdown. + isCanceled := errors.Is(err, context.Canceled) // Check if the error is non-retryable (poison pill message) if !errs.IsRetryable(err) { @@ -397,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", @@ -429,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 { @@ -468,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"), @@ -485,6 +476,25 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d ) } +func controllerClassificationTags(err error) []metrics.Tag { + origin := "infra" + 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("origin", origin), + metrics.NewTag("dependency", dependency), + } +} + // Stop gracefully shuts down all handlers with the specified timeout. // Cancels all subscription contexts and waits for consumption goroutines to finish. // timeoutMs is the maximum time in milliseconds to wait for graceful shutdown. diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 186bd1fa..296bb86c 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -47,6 +47,14 @@ type testController struct { processFunc func(context.Context, Delivery) error } +type testClassifier struct { + verdict errs.Verdict +} + +func (c testClassifier) Classify(error) errs.Verdict { + return c.verdict +} + func (c *testController) Process(ctx context.Context, delivery Delivery) error { if c.processFunc == nil { return nil @@ -431,21 +439,44 @@ func TestConsumer_ObservabilityTags(t *testing.T) { name string handlerError error nackError error - expectSuccess bool + processor errs.ErrorProcessor + expectedTags map[string]string expectAckCount bool }{ { name: "success with ack", handlerError: nil, nackError: nil, - expectSuccess: true, + processor: errs.NewClassifierProcessor(), + expectedTags: map[string]string{"result": "success"}, expectAckCount: true, }, { - name: "failure with nack", - handlerError: errs.NewRetryableError(fmt.Errorf("handler failed")), - nackError: nil, - expectSuccess: false, + name: "classified failure with nack", + handlerError: fmt.Errorf("handler failed"), + nackError: nil, + processor: errs.NewClassifierProcessor(testClassifier{ + verdict: errs.InfraDependencyRetryable, + }), + expectedTags: map[string]string{ + "result": "error", + "origin": "infra_retryable", + "dependency": "yes", + }, + expectAckCount: false, + }, + { + name: "classified cancellation with nack", + handlerError: context.Canceled, + nackError: nil, + processor: errs.NewClassifierProcessor(testClassifier{ + verdict: errs.InfraRetryable, + }), + expectedTags: map[string]string{ + "result": "cancel", + "origin": "infra_retryable", + "dependency": "no", + }, expectAckCount: false, }, } @@ -465,7 +496,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { reg := newRegistry(t, mockQ, testTopicKeyStart, "test-group") - testC := New(logger, testScope, reg, errs.NewClassifierProcessor()) + testC := New(logger, testScope, reg, tt.processor) handler := &testController{} setupController(handler, "test-handler", testTopicKeyStart, "test-group", @@ -497,19 +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() - if tt.expectSuccess { - assert.Equal(t, "true", tags["success"]) - } else { - assert.Equal(t, "false", tags["success"]) + for key, value := range tt.expectedTags { + assert.Equal(t, value, tags[key]) } } } - 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 { @@ -526,6 +565,57 @@ func TestConsumer_ObservabilityTags(t *testing.T) { } } +func TestControllerClassificationTags(t *testing.T) { + tests := []struct { + name string + err error + expected map[string]string + }{ + { + 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{ + "origin": "user", + "dependency": "no", + }, + }, + { + name: "retryable dependency error", + err: errs.NewRetryableDependencyError(fmt.Errorf("database unavailable")), + expected: map[string]string{ + "origin": "infra_retryable", + "dependency": "yes", + }, + }, + { + name: "cancellation", + err: errs.NewRetryableError(context.Canceled), + expected: map[string]string{ + "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 controllerClassificationTags(tt.err) { + actual[tag.Key] = tag.Value + } + assert.Equal(t, tt.expected, actual) + }) + } +} + func TestConsumer_AckNackLatencyTracking(t *testing.T) { ctrl := gomock.NewController(t) logger := zaptest.NewLogger(t).Sugar() diff --git a/platform/metrics/README.md b/platform/metrics/README.md index 4ea6f268..530388ba 100644 --- a/platform/metrics/README.md +++ b/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: diff --git a/platform/metrics/metrics.go b/platform/metrics/metrics.go index 0e7a568b..e2bf0db9 100644 --- a/platform/metrics/metrics.go +++ b/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)) diff --git a/platform/metrics/metrics_test.go b/platform/metrics/metrics_test.go index aa83302b..4e00bf3c 100644 --- a/platform/metrics/metrics_test.go +++ b/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)