From c8306ad0f48b7eca9821d75e4ba9cd19f2892725 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Wed, 22 Jul 2026 22:36:48 +0000 Subject: [PATCH 1/2] feat: report classified controller metrics --- platform/consumer/README.md | 2 + platform/consumer/consumer.go | 56 +++++++++++---- platform/consumer/consumer_test.go | 107 ++++++++++++++++++++++++++--- 3 files changed, 139 insertions(+), 26 deletions(-) diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 68b2603c..676395e5 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 `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. + ## Lifecycle 1. **Register** controllers before starting. diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index f5d4c773..56778dd1 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -18,6 +18,7 @@ import ( "context" "errors" "fmt" + "strconv" "sync" "time" @@ -363,27 +364,28 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d 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) - 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) + } + + // 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) + + 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) { @@ -485,6 +487,30 @@ 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" + } + + origin := "infra" + if errs.IsUserError(err) { + origin = "user" + } + + 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))), + } +} + // 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..8ef02747 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,46 @@ 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", + "error_origin": "infra", + "retryable": "true", + "dependency": "true", + }, + 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", + "error_origin": "infra", + "retryable": "true", + "dependency": "false", + }, expectAckCount: false, }, } @@ -465,7 +498,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", @@ -500,11 +533,10 @@ func TestConsumer_ObservabilityTags(t *testing.T) { if strings.Contains(histogram.Name(), "controller_latency") { 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.NotContains(t, tags, "success") } } assert.True(t, foundLatency, "Should have controller_latency metric") @@ -526,6 +558,59 @@ func TestConsumer_ObservabilityTags(t *testing.T) { } } +func TestControllerResultTags(t *testing.T) { + tests := []struct { + name string + err error + expected map[string]string + }{ + { + name: "success", + expected: map[string]string{"result": "success"}, + }, + { + name: "user error", + err: errs.NewUserError(fmt.Errorf("invalid request")), + expected: map[string]string{ + "result": "error", + "error_origin": "user", + "retryable": "false", + "dependency": "false", + }, + }, + { + name: "retryable dependency error", + err: errs.NewRetryableDependencyError(fmt.Errorf("database unavailable")), + expected: map[string]string{ + "result": "error", + "error_origin": "infra", + "retryable": "true", + "dependency": "true", + }, + }, + { + name: "cancellation", + err: errs.NewRetryableError(context.Canceled), + expected: map[string]string{ + "result": "cancel", + "error_origin": "infra", + "retryable": "true", + "dependency": "false", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + actual := make(map[string]string) + for _, tag := range controllerResultTags(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() From 100ff138c96c82aff9bc3e75751fd30c98ca0ca2 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Wed, 22 Jul 2026 23:05:43 +0000 Subject: [PATCH 2/2] refactor: use operation metrics for controller results --- platform/consumer/README.md | 2 +- platform/consumer/consumer.go | 52 +++++++++----------------- platform/consumer/consumer_test.go | 59 ++++++++++++++++-------------- platform/metrics/README.md | 12 +++++- platform/metrics/metrics.go | 9 +++-- platform/metrics/metrics_test.go | 19 ++++++++++ 6 files changed, 85 insertions(+), 68 deletions(-) diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 676395e5..65f9d373 100644 --- a/platform/consumer/README.md +++ b/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 diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 56778dd1..24bc7f24 100644 --- a/platform/consumer/consumer.go +++ b/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), } } diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 8ef02747..296bb86c 100644 --- a/platform/consumer/consumer_test.go +++ b/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,44 +565,42 @@ 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", }, }, } @@ -603,7 +608,7 @@ func TestControllerResultTags(t *testing.T) { 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) 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)