From febb1d2704ef2046642b9c6d6d0a9521e9a34a37 Mon Sep 17 00:00:00 2001 From: sergeyb Date: Thu, 23 Jul 2026 00:21:06 +0000 Subject: [PATCH] refactor: standardize metrics instrumentation --- platform/consumer/README.md | 2 + platform/consumer/consumer.go | 44 ++++----- platform/consumer/consumer_test.go | 99 ++++++++++++------- platform/consumer/controller.go | 1 + .../messagequeue/mysql/subscriber.go | 14 +-- .../messagequeue/mysql/subscriber_test.go | 28 +++++- platform/metrics/README.md | 6 +- runway/controller/merge/merge.go | 5 +- .../mergeconflictcheck/mergeconflictcheck.go | 5 +- stovepipe/controller/build/build.go | 5 +- .../controller/buildsignal/buildsignal.go | 5 +- stovepipe/controller/dlq/request.go | 6 +- stovepipe/controller/process/process.go | 5 +- submitqueue/gateway/controller/cancel.go | 15 ++- submitqueue/gateway/controller/cancel_test.go | 23 ++++- submitqueue/gateway/controller/land.go | 1 - submitqueue/gateway/controller/log/log.go | 5 +- submitqueue/gateway/controller/ping.go | 13 ++- submitqueue/gateway/controller/ping_test.go | 26 ++++- .../orchestrator/controller/batch/batch.go | 15 ++- .../orchestrator/controller/build/build.go | 5 +- .../controller/buildsignal/buildsignal.go | 5 +- .../controller/cancel/BUILD.bazel | 1 + .../orchestrator/controller/cancel/cancel.go | 35 +++---- .../controller/conclude/conclude.go | 5 +- .../orchestrator/controller/dlq/batch.go | 6 +- .../controller/dlq/buildsignal.go | 5 +- .../orchestrator/controller/dlq/log.go | 9 +- .../controller/dlq/mergeconflictsignal.go | 6 +- .../controller/dlq/mergesignal.go | 6 +- .../orchestrator/controller/dlq/request.go | 6 +- .../orchestrator/controller/merge/merge.go | 5 +- .../mergeconflictsignal.go | 5 +- .../controller/mergesignal/mergesignal.go | 5 +- .../orchestrator/controller/score/score.go | 17 ++-- .../controller/speculate/speculate.go | 7 +- .../orchestrator/controller/start/start.go | 5 +- .../controller/validate/validate.go | 5 +- 38 files changed, 231 insertions(+), 230 deletions(-) diff --git a/platform/consumer/README.md b/platform/consumer/README.md index 65f9d373..6bd067db 100644 --- a/platform/consumer/README.md +++ b/platform/consumer/README.md @@ -120,6 +120,8 @@ When the consumer is wired with `errs.AlwaysRetryableProcessor` (DLQ reconciliat 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. +The consumer also owns lifecycle metrics for the resulting `ack`, `nack`, or `reject` transport operation. Queue controllers should emit only domain-specific event counters; they must not duplicate the consumer-owned `process` lifecycle metrics. + ## Lifecycle 1. **Register** controllers before starting. diff --git a/platform/consumer/consumer.go b/platform/consumer/consumer.go index 24bc7f24..2f684e73 100644 --- a/platform/consumer/consumer.go +++ b/platform/consumer/consumer.go @@ -83,12 +83,14 @@ type activeSubscription struct { // consumers (per-node classifier walk that preserves controller-attached // framework wraps), or errs.AlwaysRetryableProcessor for narrowly-scoped // consumers such as DLQ reconciliation that must redeliver on any failure. -// processor must not be nil; callers that genuinely want no transformation -// can pass errs.NewClassifierProcessor() with no classifiers. +// scope is used as provided so wiring can distinguish primary and DLQ consumers +// without introducing duplicate consumer sub-scopes. processor must not be nil; +// callers that genuinely want no transformation can pass +// errs.NewClassifierProcessor() with no classifiers. func New(logger *zap.SugaredLogger, scope tally.Scope, registry TopicRegistry, processor errs.ErrorProcessor) Consumer { return &consumer{ logger: logger, - metricsScope: scope.SubScope("consumer"), + metricsScope: scope, registry: registry, processor: processor, subscriptions: make(map[TopicKey]*activeSubscription), @@ -394,14 +396,16 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d ) // Reject moves to DLQ (or acks if DLQ disabled) - if rejectErr := delivery.Reject(ctx, err.Error()); rejectErr != nil { + rejectOp := metrics.Begin(controllerScope, "reject", metrics.StorageLatencyBuckets) + rejectErr := delivery.Reject(ctx, err.Error()) + rejectOp.Complete(rejectErr) + if rejectErr != nil { m.logger.Errorw("failed to reject non-retryable message", "controller", controller.Name(), "topic_key", controller.TopicKey(), "message_id", msg.ID, "error", rejectErr, ) - metrics.NamedCounter(controllerScope, opName, "reject_errors", 1) } return } @@ -424,48 +428,34 @@ func (m *consumer) processDelivery(ctx context.Context, controller Controller, d ) // Nack with no delay - let visibility timeout handle retry delay - nackStart := time.Now() - if nackErr := delivery.Nack(ctx, 0); nackErr != nil { + nackOp := metrics.Begin(controllerScope, "nack", metrics.StorageLatencyBuckets) + nackErr := delivery.Nack(ctx, 0) + nackOp.Complete(nackErr) + if nackErr != nil { m.logger.Errorw("failed to nack message", "controller", controller.Name(), "topic_key", topicKey, "message_id", msg.ID, "error", nackErr, ) - metrics.NamedCounter(controllerScope, opName, "nack_errors", 1) - } else { - metrics.NamedCounter(controllerScope, opName, "nack_count", 1) - metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets, - metrics.NewTag("operation", "nack"), - metrics.NewTag("success", "true"), - ).RecordDuration(time.Since(nackStart)) } return } // Controller succeeded - ack message - ackStart := time.Now() - if ackErr := delivery.Ack(ctx); ackErr != nil { + ackOp := metrics.Begin(controllerScope, "ack", metrics.StorageLatencyBuckets) + ackErr := delivery.Ack(ctx) + ackOp.Complete(ackErr) + if ackErr != nil { m.logger.Errorw("failed to ack message", "controller", controller.Name(), "topic_key", topicKey, "message_id", msg.ID, "error", ackErr, ) - metrics.NamedCounter(controllerScope, opName, "ack_errors", 1) - metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets, - metrics.NewTag("operation", "ack"), - metrics.NewTag("success", "false"), - ).RecordDuration(time.Since(ackStart)) return } - metrics.NamedCounter(controllerScope, opName, "ack_count", 1) - metrics.NamedHistogram(controllerScope, opName, "ack_nack_latency", metrics.StorageLatencyBuckets, - metrics.NewTag("operation", "ack"), - metrics.NewTag("success", "true"), - ).RecordDuration(time.Since(ackStart)) - m.logger.Debugw("message processed successfully", "controller", controller.Name(), "topic_key", topicKey, diff --git a/platform/consumer/consumer_test.go b/platform/consumer/consumer_test.go index 296bb86c..e141ce08 100644 --- a/platform/consumer/consumer_test.go +++ b/platform/consumer/consumer_test.go @@ -117,6 +117,10 @@ func setupDelivery(del *queuemock.MockDelivery, msg entityqueue.Message, ackErr, close(done) return nackErr }).MaxTimes(1) + del.EXPECT().Reject(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, reason string) error { + close(done) + return nil + }).MaxTimes(1) return done } @@ -436,20 +440,20 @@ func TestConsumer_Stop(t *testing.T) { func TestConsumer_ObservabilityTags(t *testing.T) { tests := []struct { - name string - handlerError error - nackError error - processor errs.ErrorProcessor - expectedTags map[string]string - expectAckCount bool + name string + handlerError error + nackError error + processor errs.ErrorProcessor + expectedTags map[string]string + transport string }{ { - name: "success with ack", - handlerError: nil, - nackError: nil, - processor: errs.NewClassifierProcessor(), - expectedTags: map[string]string{"result": "success"}, - expectAckCount: true, + name: "success with ack", + handlerError: nil, + nackError: nil, + processor: errs.NewClassifierProcessor(), + expectedTags: map[string]string{"result": "success"}, + transport: "ack", }, { name: "classified failure with nack", @@ -463,7 +467,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { "origin": "infra_retryable", "dependency": "yes", }, - expectAckCount: false, + transport: "nack", }, { name: "classified cancellation with nack", @@ -477,7 +481,18 @@ func TestConsumer_ObservabilityTags(t *testing.T) { "origin": "infra_retryable", "dependency": "no", }, - expectAckCount: false, + transport: "nack", + }, + { + name: "user failure with reject", + handlerError: errs.NewUserError(fmt.Errorf("invalid request")), + processor: errs.NewClassifierProcessor(), + expectedTags: map[string]string{ + "result": "error", + "origin": "user", + "dependency": "no", + }, + transport: "reject", }, } @@ -528,6 +543,7 @@ func TestConsumer_ObservabilityTags(t *testing.T) { var foundLatency bool for _, histogram := range histograms { + assert.NotContains(t, histogram.Name(), "consumer.consumer") if strings.Contains(histogram.Name(), "process.finish") { foundLatency = true tags := histogram.Tags() @@ -538,27 +554,32 @@ func TestConsumer_ObservabilityTags(t *testing.T) { } assert.True(t, foundLatency, "Should have process.finish metric") + var foundTransport bool + for _, histogram := range histograms { + if strings.Contains(histogram.Name(), tt.transport+".finish") { + foundTransport = true + assert.Equal(t, "success", histogram.Tags()["result"]) + } + assert.NotContains(t, histogram.Name(), "ack_nack_latency") + } + assert.True(t, foundTransport, "Should have %s.finish metric", tt.transport) + counters := snapshot.Counters() for _, duplicate := range []string{ "messages_received", "messages_processed", "non_retryable_errors", "controller_errors", + "ack_count", + "ack_errors", + "nack_count", + "nack_errors", + "reject_errors", } { for _, counter := range counters { assert.NotContains(t, counter.Name(), duplicate) } } - if tt.expectAckCount { - var foundAck bool - for _, counter := range counters { - if strings.Contains(counter.Name(), "ack_count") { - foundAck = true - assert.Greater(t, counter.Value(), int64(0)) - } - } - assert.True(t, foundAck, "Should have ack_count metric") - } _ = testC.Stop(30000) }) @@ -616,7 +637,7 @@ func TestControllerClassificationTags(t *testing.T) { } } -func TestConsumer_AckNackLatencyTracking(t *testing.T) { +func TestConsumer_AckLifecycleMetrics(t *testing.T) { ctrl := gomock.NewController(t) logger := zaptest.NewLogger(t).Sugar() scope := tally.NewTestScope("consumer", nil) @@ -654,14 +675,21 @@ func TestConsumer_AckNackLatencyTracking(t *testing.T) { <-done snapshot := scope.Snapshot() - assert.NotEmpty(t, snapshot.Histograms(), "Should have histogram metrics for latency tracking") - assert.NotEmpty(t, snapshot.Counters(), "Should have counter metrics") + histograms := snapshot.Histograms() + var foundAck bool + for _, histogram := range histograms { + if strings.Contains(histogram.Name(), "ack.finish") { + foundAck = true + assert.Equal(t, "success", histogram.Tags()["result"]) + } + } + assert.True(t, foundAck, "Should have successful ack.finish metric") err = c.Stop(30000) require.NoError(t, err) } -func TestConsumer_ErrorMetrics(t *testing.T) { +func TestConsumer_NackLifecycleMetrics(t *testing.T) { ctrl := gomock.NewController(t) logger := zaptest.NewLogger(t).Sugar() scope := tally.NewTestScope("consumer", nil) @@ -701,16 +729,15 @@ func TestConsumer_ErrorMetrics(t *testing.T) { <-done snapshot := scope.Snapshot() - counters := snapshot.Counters() - - var hasErrorMetrics bool - for _, counter := range counters { - if strings.Contains(counter.Name(), "errors") { - hasErrorMetrics = true - break + histograms := snapshot.Histograms() + var foundNackError bool + for _, histogram := range histograms { + if strings.Contains(histogram.Name(), "nack.finish") { + foundNackError = true + assert.Equal(t, "error", histogram.Tags()["result"]) } } - assert.True(t, hasErrorMetrics, "Should track error metrics") + assert.True(t, foundNackError, "Should have failed nack.finish metric") err = c.Stop(30000) require.NoError(t, err) diff --git a/platform/consumer/controller.go b/platform/consumer/controller.go index 13b33e55..0b31b05c 100644 --- a/platform/consumer/controller.go +++ b/platform/consumer/controller.go @@ -89,6 +89,7 @@ func (d *deliveryWrapper) Metadata() map[string]string { // The Controller interface enables clean separation of concerns: // - Controller focuses on business logic (deserialize, process, return error status) // - Consumer handles infrastructure (subscription, ack/nack, metrics, lifecycle) +// Controllers may emit domain event counters, but must not duplicate the consumer-owned Process lifecycle metrics. // The implementation of the controller should be idempotent and stateless. The controller is expected to be retried for the same message multiple times and should process side effects gracefully. // The implementation must be thread-safe. type Controller interface { diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 514bcced..ffc80532 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/platform/extension/messagequeue/mysql/subscriber.go @@ -747,13 +747,18 @@ func (w *partitionWorker) run(ctx context.Context) { // Partition leasing guarantees a single writer, so the TOCTOU gap between // GetDeliveryState and MarkDelivered cannot cause incorrect behavior — no other // worker can mutate the same (consumer_group, topic, partition_key, offset). -func (w *partitionWorker) pollAndDeliver(ctx context.Context) error { - start := time.Now() +func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) { s := w.subscriber sub := w.sub cfg := sub.config partitionKey := w.partitionKey + op := metrics.Begin(s.scope, "poll", metrics.StorageLatencyBuckets, + metrics.NewTag("topic", sub.topic), + metrics.NewTag("partition_key", partitionKey), + ) + defer func() { op.Complete(retErr) }() + // Initialize offset for this partition once per worker lifetime if !w.offsetInitialized { if err := s.offsetStore.Initialize(ctx, sub.topic, partitionKey, cfg.ConsumerGroup); err != nil { @@ -912,15 +917,10 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) error { // Record poll metrics if messageCount > 0 { - elapsed := time.Since(start) metrics.NamedCounter(s.scope, "poll", "messages_delivered", int64(messageCount), metrics.NewTag("topic", sub.topic), metrics.NewTag("partition_key", partitionKey), ) - metrics.NamedHistogram(s.scope, "poll", "latency", metrics.StorageLatencyBuckets, - metrics.NewTag("topic", sub.topic), - metrics.NewTag("partition_key", partitionKey), - ).RecordDuration(elapsed) } return nil diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index baa2fb6b..d5ab8c3f 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/platform/extension/messagequeue/mysql/subscriber_test.go @@ -460,10 +460,11 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) { mockOffsetStore := NewMockoffsetStore(ctrl) mockLeaseStore := NewMockpartitionLeaseStore(ctrl) mockDeliveryState := NewMockdeliveryStateStore(ctrl) + metricsScope := tally.NewTestScope("test", nil) s := NewSubscriber( zaptest.NewLogger(t).Sugar(), - tally.NoopScope, + metricsScope, mockMessageStore, mockOffsetStore, mockLeaseStore, @@ -512,7 +513,7 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) { done: make(chan struct{}), } - w.pollAndDeliver(ctx) + require.NoError(t, w.pollAndDeliver(ctx)) // Verify message was delivered select { @@ -524,6 +525,29 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) { // Verify offset was initialized only once assert.True(t, w.offsetInitialized) + + snapshot := metricsScope.Snapshot() + var foundStart bool + for _, counter := range snapshot.Counters() { + if counter.Name() == "test.subscriber.poll.start" { + foundStart = true + assert.Equal(t, "test_topic", counter.Tags()["topic"]) + assert.Equal(t, "part-1", counter.Tags()["partition_key"]) + } + } + assert.True(t, foundStart, "expected poll.start counter") + + var foundFinish bool + for _, histogram := range snapshot.Histograms() { + if histogram.Name() == "test.subscriber.poll.finish" { + foundFinish = true + assert.Equal(t, "success", histogram.Tags()["result"]) + assert.Equal(t, "test_topic", histogram.Tags()["topic"]) + assert.Equal(t, "part-1", histogram.Tags()["partition_key"]) + } + assert.NotContains(t, histogram.Name(), "poll.latency") + } + assert.True(t, foundFinish, "expected poll.finish histogram") } // TestSubscriber_StopAllWorkers tests that all workers are stopped gracefully. diff --git a/platform/metrics/README.md b/platform/metrics/README.md index 530388ba..f3161ac6 100644 --- a/platform/metrics/README.md +++ b/platform/metrics/README.md @@ -57,11 +57,7 @@ h := metrics.NamedHistogram(c.scope, "process", "duration", metrics.FastLatencyB h.RecordDuration(elapsed) ``` -Use `tally.Scope` directly for gauges: - -```go -c.scope.SubScope("consumer").Gauge("pending_messages").Update(float64(len(pending))) -``` +Do not emit gauges or timers. Represent operation latency and completion count with lifecycle histograms, and represent instantaneous quantities as sampled histogram values when needed. ### Why histograms, not timers diff --git a/runway/controller/merge/merge.go b/runway/controller/merge/merge.go index 5e065f21..f89a9f0a 100644 --- a/runway/controller/merge/merge.go +++ b/runway/controller/merge/merge.go @@ -76,12 +76,9 @@ func NewController(p Params) *Controller { // Process deserializes the merge request, performs the committing merge, and // publishes the result. Returns nil to ack, or an error to nack. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() request := &runwaymq.MergeRequest{} diff --git a/runway/controller/mergeconflictcheck/mergeconflictcheck.go b/runway/controller/mergeconflictcheck/mergeconflictcheck.go index fb53c62d..4cdd2436 100644 --- a/runway/controller/mergeconflictcheck/mergeconflictcheck.go +++ b/runway/controller/mergeconflictcheck/mergeconflictcheck.go @@ -76,12 +76,9 @@ func NewController(p Params) *Controller { // Process deserializes the merge request, performs a dry-run merge check, and // publishes the result. Returns nil to ack, or an error to nack. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() request := &runwaymq.MergeRequest{} diff --git a/stovepipe/controller/build/build.go b/stovepipe/controller/build/build.go index 557ece2d..b488d93d 100644 --- a/stovepipe/controller/build/build.go +++ b/stovepipe/controller/build/build.go @@ -79,10 +79,7 @@ func NewController( // Process reloads the request referenced by the delivery, triggers a build for // its decided scope, and publishes the build id to buildsignal. Returns nil to // ack (success) or an error to nack (retry) / reject (DLQ). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() br := &stovepipemq.BuildRequest{} diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index 3f8fb0a3..eb65d43b 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -94,10 +94,7 @@ func NewController( // the latest status, persists a real transition, and either reschedules a // poll or, once terminal, publishes the build id to record. Returns nil to // ack (success) or an error to nack (retry) / reject (DLQ). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() sig := &stovepipemq.BuildSignal{} diff --git a/stovepipe/controller/dlq/request.go b/stovepipe/controller/dlq/request.go index 9484bb91..262fd996 100644 --- a/stovepipe/controller/dlq/request.go +++ b/stovepipe/controller/dlq/request.go @@ -66,10 +66,7 @@ func NewController( // (success) or an error to nack (retry) — pair this controller only with a consumer // wired with errs.AlwaysRetryableProcessor so a transient reconcile failure retries // instead of dead-lettering the DLQ message itself. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() pr := &stovepipemq.ProcessRequest{} @@ -105,7 +102,6 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return err } - metrics.NamedCounter(c.metricsScope, _opName, "reconciled", 1) return nil } diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 0fbf33ed..a1931735 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -83,10 +83,7 @@ func NewController( // Process reloads the request referenced by the delivery, coalesces older heads, // and admits the latest when a slot is open. Returns nil to ack (success) or an error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, _opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() pr := &stovepipemq.ProcessRequest{} diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index 25b9cf23..8a19848a 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/submitqueue/gateway/controller/cancel.go @@ -17,12 +17,12 @@ package controller import ( "context" "fmt" - "time" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/platform/metrics" requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -64,13 +64,11 @@ func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store sto // completion before the cancel propagates may still land. The RequestStatusCancelling // entry written here records the user's intent; the terminal outcome is reflected by a // later RequestStatusCancelled (orchestrator side) or RequestStatusLanded entry. -func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest) error { - start := time.Now() - defer func() { - c.metricsScope.Timer("cancel_request_latency").Record(time.Since(start)) - }() +func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest) (retErr error) { + const opName = "cancel" - c.metricsScope.Counter("cancel_request_count").Inc(1) + op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets) + defer func() { op.Complete(retErr) }() if req.ID == "" { return fmt.Errorf("requires the request to have a sqid specified: %w", ErrInvalidRequest) @@ -84,7 +82,7 @@ func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest) // Verify the sqid exists before recording intent or publishing. if _, err := c.requestSummaryStore.Get(ctx, req.ID); err != nil { if storage.IsNotFound(err) { - c.metricsScope.Counter("cancel_request_not_found").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "not_found", 1) return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) } return fmt.Errorf("failed to look up request summary for sqid=%s: %w", req.ID, err) @@ -110,7 +108,6 @@ func (c *CancelController) Cancel(ctx context.Context, req entity.CancelRequest) "sqid", req.ID, "topic_key", topickey.TopicKeyCancel, ) - c.metricsScope.Counter("cancel_publish_success").Inc(1) return nil } diff --git a/submitqueue/gateway/controller/cancel_test.go b/submitqueue/gateway/controller/cancel_test.go index 1e4161ca..e0c76751 100644 --- a/submitqueue/gateway/controller/cancel_test.go +++ b/submitqueue/gateway/controller/cancel_test.go @@ -82,13 +82,34 @@ func TestNewCancelController(t *testing.T) { func TestCancel_HappyPath(t *testing.T) { ctrl := gomock.NewController(t) + scope := tally.NewTestScope("gateway", nil) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageFixture(ctrl, "test-queue/42").storage, newCancelTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewCancelController(zap.NewNop().Sugar(), scope, newCancelStorageFixture(ctrl, "test-queue/42").storage, newCancelTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() err := controller.Cancel(ctx, testCancelRequest("test-queue/42", "user changed their mind")) require.NoError(t, err) + + snapshot := scope.Snapshot() + var foundStart bool + for _, counter := range snapshot.Counters() { + if counter.Name() == "gateway.cancel.start" { + foundStart = true + } + assert.NotContains(t, counter.Name(), "cancel_request_count") + assert.NotContains(t, counter.Name(), "cancel_publish_success") + } + assert.True(t, foundStart) + + var foundFinish bool + for _, histogram := range snapshot.Histograms() { + if histogram.Name() == "gateway.cancel.finish" { + foundFinish = true + assert.Equal(t, "success", histogram.Tags()["result"]) + } + } + assert.True(t, foundFinish) } func TestCancel_ReturnsErrorOnEmptySqid(t *testing.T) { diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 1352388c..878427ca 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -174,7 +174,6 @@ func (c *LandController) Land(ctx context.Context, req entity.LandRequest) (resu "sqid", req.ID, "topic_key", topickey.TopicKeyStart, ) - metrics.NamedCounter(c.metricsScope, opName, "publish_success", 1) return entity.LandResult{ID: req.ID}, nil } diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go index 5deb70e7..a37ba75f 100644 --- a/submitqueue/gateway/controller/log/log.go +++ b/submitqueue/gateway/controller/log/log.go @@ -65,12 +65,9 @@ func NewController( // Process processes a log delivery from the queue. // Deserializes the request log entry and persists it to storage. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.StorageLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() // Deserialize request log entry diff --git a/submitqueue/gateway/controller/ping.go b/submitqueue/gateway/controller/ping.go index 99c6c2f2..47027248 100644 --- a/submitqueue/gateway/controller/ping.go +++ b/submitqueue/gateway/controller/ping.go @@ -21,6 +21,7 @@ import ( "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/metrics" "go.uber.org/zap" ) @@ -39,20 +40,18 @@ func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController { } // Ping handles the ping request and returns a response -func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error) { - start := time.Now() - defer func() { - c.metricsScope.Timer("ping_latency").Record(time.Since(start)) - }() +func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) { + const opName = "ping" - c.metricsScope.Counter("ping_requests_total").Inc(1) + op := metrics.Begin(c.metricsScope, opName, metrics.FastLatencyBuckets) + defer func() { op.Complete(retErr) }() message := "pong!" isEcho := false if req.Message != "" { message = "echo: " + req.Message isEcho = true - c.metricsScope.Counter("echo_requests_total").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "echo_requests", 1) } hostname, _ := os.Hostname() diff --git a/submitqueue/gateway/controller/ping_test.go b/submitqueue/gateway/controller/ping_test.go index 5591df45..e7b413df 100644 --- a/submitqueue/gateway/controller/ping_test.go +++ b/submitqueue/gateway/controller/ping_test.go @@ -43,7 +43,8 @@ func TestPing_DefaultMessage(t *testing.T) { } func TestPing_CustomMessage(t *testing.T) { - controller := NewPingController(zap.NewNop(), tally.NoopScope) + scope := tally.NewTestScope("gateway", nil) + controller := NewPingController(zap.NewNop(), scope) ctx := context.Background() testCases := []struct { @@ -65,6 +66,29 @@ func TestPing_CustomMessage(t *testing.T) { assert.Equal(t, tc.expected, resp.Message) }) } + + snapshot := scope.Snapshot() + var foundStart, foundEcho bool + for _, counter := range snapshot.Counters() { + switch counter.Name() { + case "gateway.ping.start": + foundStart = true + case "gateway.ping.echo_requests": + foundEcho = true + } + assert.NotContains(t, counter.Name(), "requests_total") + } + assert.True(t, foundStart) + assert.True(t, foundEcho) + + var foundFinish bool + for _, histogram := range snapshot.Histograms() { + if histogram.Name() == "gateway.ping.finish" { + foundFinish = true + assert.Equal(t, "success", histogram.Tags()["result"]) + } + } + assert.True(t, foundFinish) } func TestPing_ServiceName(t *testing.T) { diff --git a/submitqueue/orchestrator/controller/batch/batch.go b/submitqueue/orchestrator/controller/batch/batch.go index 8e25435c..38b007e7 100644 --- a/submitqueue/orchestrator/controller/batch/batch.go +++ b/submitqueue/orchestrator/controller/batch/batch.go @@ -49,6 +49,8 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +const opName = "process" + // NewController creates a new batch controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -75,12 +77,7 @@ func NewController( // Process processes a batch delivery from the queue. // Deserializes the request, groups into batch, and publishes to the score topic. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - const opName = "process" - - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() // Deserialize request ID from payload @@ -110,7 +107,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // terminal state, or the cancel controller has recorded a cancellation intent // (RequestStateCancelling). A halted request must never spawn a new batch. if entity.IsRequestStateHalted(request.State) { - c.metricsScope.Counter("skipped_halted").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "skipped_halted", 1) c.logger.Infow("skipping batch for halted request", "request_id", request.ID, "state", string(request.State), @@ -259,7 +256,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // the message: there is nothing for us to do, and retrying would not help // since the new state of R is now visible to the cancel pipeline. if errors.Is(err, storage.ErrVersionMismatch) { - c.metricsScope.Counter("request_claim_lost_race").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_claim_lost_race", 1) c.logger.Infow("abandoning batch creation; request advanced concurrently (likely cancel)", "request_id", request.ID, "request_version", request.Version, @@ -267,7 +264,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r ) return nil } - c.metricsScope.Counter("request_claim_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_claim_errors", 1) return fmt.Errorf("failed to claim request %s for batch %s: %w", request.ID, batch.ID, err) } request.Version = newRequestVersion diff --git a/submitqueue/orchestrator/controller/build/build.go b/submitqueue/orchestrator/controller/build/build.go index e9170ed2..a2fc396e 100644 --- a/submitqueue/orchestrator/controller/build/build.go +++ b/submitqueue/orchestrator/controller/build/build.go @@ -70,12 +70,9 @@ func NewController( // Process processes a build delivery from the queue. // Deserializes the batch, triggers a build, and publishes a build entity to the build signal topic. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() // Deserialize batch ID from payload diff --git a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go index a30dfddf..f2064164 100644 --- a/submitqueue/orchestrator/controller/buildsignal/buildsignal.go +++ b/submitqueue/orchestrator/controller/buildsignal/buildsignal.go @@ -101,12 +101,9 @@ func NewController( // so a transient enqueue blip nacks and replays (up to MaxAttempts) rather // than silently stalling the build, then still falls through to DLQ if it // persists. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() buildID, err := entity.BuildIDFromBytes(msg.Payload) diff --git a/submitqueue/orchestrator/controller/cancel/BUILD.bazel b/submitqueue/orchestrator/controller/cancel/BUILD.bazel index 770a2989..66f99b3f 100644 --- a/submitqueue/orchestrator/controller/cancel/BUILD.bazel +++ b/submitqueue/orchestrator/controller/cancel/BUILD.bazel @@ -8,6 +8,7 @@ go_library( deps = [ "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", + "//platform/metrics:go_default_library", "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", diff --git a/submitqueue/orchestrator/controller/cancel/cancel.go b/submitqueue/orchestrator/controller/cancel/cancel.go index c83f1ab2..652c4f10 100644 --- a/submitqueue/orchestrator/controller/cancel/cancel.go +++ b/submitqueue/orchestrator/controller/cancel/cancel.go @@ -61,6 +61,7 @@ import ( "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/metrics" corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" @@ -81,6 +82,8 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +const opName = "process" + // NewController creates a new cancel controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -103,19 +106,17 @@ func NewController( // Process processes a cancel delivery from the queue. // Returns nil to ack (success), or error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { - c.metricsScope.Counter("received").Inc(1) - msg := delivery.Message() cancelReq, err := entity.CancelRequestFromBytes(msg.Payload) if err != nil { - c.metricsScope.Counter("deserialize_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "deserialize_errors", 1) return fmt.Errorf("failed to deserialize cancel request: %w", err) } request, err := c.store.GetRequestStore().Get(ctx, cancelReq.ID) if err != nil { - c.metricsScope.Counter("storage_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) return fmt.Errorf("failed to get request %s: %w", cancelReq.ID, err) } @@ -130,7 +131,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er ) if entity.IsRequestStateTerminal(request.State) { - c.metricsScope.Counter("already_terminal").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "already_terminal", 1) return nil } @@ -167,17 +168,17 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er func (c *Controller) markCancelling(ctx context.Context, request entity.Request) (entity.Request, error) { if request.State == entity.RequestStateCancelling { // Idempotent re-delivery: prior pass already recorded intent. - c.metricsScope.Counter("already_cancelling").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "already_cancelling", 1) return request, nil } newVersion := request.Version + 1 if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newVersion, entity.RequestStateCancelling); err != nil { - c.metricsScope.Counter("request_update_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_update_errors", 1) return entity.Request{}, fmt.Errorf("failed to mark request %s as cancelling: %w", request.ID, err) } request.Version = newVersion request.State = entity.RequestStateCancelling - c.metricsScope.Counter("request_cancelling").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_cancelling", 1) return request, nil } @@ -194,7 +195,7 @@ func (c *Controller) findActiveBatch(ctx context.Context, request entity.Request // TODO: Scans all the batches in flight - make it more efficient? active, err := c.store.GetBatchStore().GetByQueueAndStates(ctx, request.Queue, entity.ActiveBatchStates()) if err != nil { - c.metricsScope.Counter("batch_store_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "batch_store_errors", 1) return entity.Batch{}, false, fmt.Errorf("failed to get active batches for queue=%s: %w", request.Queue, err) } @@ -218,7 +219,7 @@ func (c *Controller) findActiveBatch(ctx context.Context, request entity.Request func (c *Controller) cancelRequest(ctx context.Context, request entity.Request, reason string) error { newVersion := request.Version + 1 if err := c.store.GetRequestStore().UpdateState(ctx, request.ID, request.Version, newVersion, entity.RequestStateCancelled); err != nil { - c.metricsScope.Counter("request_update_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_update_errors", 1) return fmt.Errorf("failed to cancel request %s: %w", request.ID, err) } @@ -228,7 +229,7 @@ func (c *Controller) cancelRequest(ctx context.Context, request entity.Request, } logEntry := entity.NewRequestLog(request.ID, entity.RequestStatusCancelled, newVersion, "", metadata) if err := corerequest.PublishLog(ctx, c.registry, logEntry, request.ID); err != nil { - c.metricsScope.Counter("log_publish_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "log_publish_errors", 1) return fmt.Errorf("failed to publish cancel log for request %s: %w", request.ID, err) } @@ -236,7 +237,7 @@ func (c *Controller) cancelRequest(ctx context.Context, request entity.Request, "request_id", request.ID, "queue", request.Queue, ) - c.metricsScope.Counter("request_cancelled").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "request_cancelled", 1) return nil } @@ -266,7 +267,7 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error if batch.State != entity.BatchStateCancelling { newVersion := batch.Version + 1 if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateCancelling); err != nil { - c.metricsScope.Counter("batch_update_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "batch_update_errors", 1) // storage.ErrVersionMismatch here means the batch advanced concurrently // (e.g. speculate / merge progressed). Returned as-is because the // sentinel is intrinsically retryable; the re-fetch will see the new state @@ -276,17 +277,17 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error } batch.Version = newVersion batch.State = entity.BatchStateCancelling - c.metricsScope.Counter("batch_cancelling").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "batch_cancelling", 1) } else { - c.metricsScope.Counter("batch_already_cancelling").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "batch_already_cancelling", 1) } if err := c.publishBatchID(ctx, topickey.TopicKeySpeculate, batch.ID, batch.Queue); err != nil { - c.metricsScope.Counter("publish_errors").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1) return fmt.Errorf("failed to hand off cancelled batch %s to speculate: %w", batch.ID, err) } - c.metricsScope.Counter("batch_handed_off").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "batch_handed_off", 1) return nil } diff --git a/submitqueue/orchestrator/controller/conclude/conclude.go b/submitqueue/orchestrator/controller/conclude/conclude.go index f5c36e4a..4e451a5b 100644 --- a/submitqueue/orchestrator/controller/conclude/conclude.go +++ b/submitqueue/orchestrator/controller/conclude/conclude.go @@ -64,10 +64,7 @@ func NewController( // Process processes a conclude delivery from the queue. // Deserializes the batch and completes the pipeline processing. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() // Deserialize batch ID from payload diff --git a/submitqueue/orchestrator/controller/dlq/batch.go b/submitqueue/orchestrator/controller/dlq/batch.go index c91b03bb..59167e1c 100644 --- a/submitqueue/orchestrator/controller/dlq/batch.go +++ b/submitqueue/orchestrator/controller/dlq/batch.go @@ -70,12 +70,9 @@ func NewDLQBatchController( } // Process reconciles a single DLQ delivery for a batch-scoped topic. -func (c *batchController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *batchController) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() bid, err := entity.BatchIDFromBytes(msg.Payload) @@ -102,7 +99,6 @@ func (c *batchController) Process(ctx context.Context, delivery consumer.Deliver return err } - metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1) return nil } diff --git a/submitqueue/orchestrator/controller/dlq/buildsignal.go b/submitqueue/orchestrator/controller/dlq/buildsignal.go index aa8e690e..b45c2255 100644 --- a/submitqueue/orchestrator/controller/dlq/buildsignal.go +++ b/submitqueue/orchestrator/controller/dlq/buildsignal.go @@ -68,12 +68,9 @@ func NewDLQBuildSignalController( } // Process reconciles a single DLQ delivery for the buildsignal topic. -func (c *buildSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *buildSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() bid, err := entity.BuildIDFromBytes(msg.Payload) diff --git a/submitqueue/orchestrator/controller/dlq/log.go b/submitqueue/orchestrator/controller/dlq/log.go index 7b02b8db..e8261891 100644 --- a/submitqueue/orchestrator/controller/dlq/log.go +++ b/submitqueue/orchestrator/controller/dlq/log.go @@ -19,7 +19,6 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/metrics" "go.uber.org/zap" ) @@ -57,12 +56,7 @@ func NewDLQLogController( } // Process records that a log message landed in the DLQ and acks it. -func (c *logController) Process(_ context.Context, delivery consumer.Delivery) (retErr error) { - const opName = "process" - - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *logController) Process(_ context.Context, delivery consumer.Delivery) error { msg := delivery.Message() dmeta := delivery.Metadata() c.logger.Warnw("log message dropped to dlq", @@ -72,7 +66,6 @@ func (c *logController) Process(_ context.Context, delivery consumer.Delivery) ( "dlq_failure_count", dmeta["dlq.failure_count"], "dlq_last_error", dmeta["dlq.last_error"], ) - metrics.NamedCounter(c.metricsScope, opName, "dropped", 1) return nil } diff --git a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go index 52489d91..34d922d7 100644 --- a/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergeconflictsignal.go @@ -64,12 +64,9 @@ func NewDLQMergeConflictSignalController( } // Process reconciles a single DLQ delivery for the mergeconflictsignal topic. -func (c *mergeConflictSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *mergeConflictSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() result := &runwaymq.MergeResult{} @@ -92,7 +89,6 @@ func (c *mergeConflictSignalController) Process(ctx context.Context, delivery co return err } - metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1) return nil } diff --git a/submitqueue/orchestrator/controller/dlq/mergesignal.go b/submitqueue/orchestrator/controller/dlq/mergesignal.go index 155efa55..1eb5aec7 100644 --- a/submitqueue/orchestrator/controller/dlq/mergesignal.go +++ b/submitqueue/orchestrator/controller/dlq/mergesignal.go @@ -63,12 +63,9 @@ func NewDLQMergeSignalController( } // Process reconciles a single DLQ delivery for the mergesignal topic. -func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() result := &runwaymq.MergeResult{} @@ -91,7 +88,6 @@ func (c *mergeSignalController) Process(ctx context.Context, delivery consumer.D return err } - metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1) return nil } diff --git a/submitqueue/orchestrator/controller/dlq/request.go b/submitqueue/orchestrator/controller/dlq/request.go index ee8df0fc..615346c4 100644 --- a/submitqueue/orchestrator/controller/dlq/request.go +++ b/submitqueue/orchestrator/controller/dlq/request.go @@ -106,12 +106,9 @@ func NewDLQRequestController( } // Process reconciles a single DLQ delivery for a request-scoped topic. -func (c *requestController) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *requestController) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() requestID, err := c.decode(msg.Payload) @@ -142,7 +139,6 @@ func (c *requestController) Process(ctx context.Context, delivery consumer.Deliv return err } - metrics.NamedCounter(c.metricsScope, opName, "reconciled", 1) return nil } diff --git a/submitqueue/orchestrator/controller/merge/merge.go b/submitqueue/orchestrator/controller/merge/merge.go index 74c52323..494030cb 100644 --- a/submitqueue/orchestrator/controller/merge/merge.go +++ b/submitqueue/orchestrator/controller/merge/merge.go @@ -90,12 +90,9 @@ func NewController( // (reject to DLQ). The publish to runway is retryable — it is the hand-off that // keeps the merge alive, so a transient enqueue blip should replay rather than // strand the batch. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() bid, err := entity.BatchIDFromBytes(msg.Payload) diff --git a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go index 5007b726..54b192ae 100644 --- a/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go +++ b/submitqueue/orchestrator/controller/mergeconflictsignal/mergeconflictsignal.go @@ -76,12 +76,9 @@ func NewController( // infrastructure faults — deserialize, storage, the terminal transition, and the // batch publish — return an error and reject to the DLQ, where the request is // reconciled to Error. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() // The runway result carries full data (it crosses the service boundary). Its diff --git a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go index 2d619bdd..cccff266 100644 --- a/submitqueue/orchestrator/controller/mergesignal/mergesignal.go +++ b/submitqueue/orchestrator/controller/mergesignal/mergesignal.go @@ -77,12 +77,9 @@ func NewController( // infrastructure faults — deserialize, storage, the state transition, and the // fan-out publishes — return an error and reject to the DLQ, where the batch is // reconciled to Failed. -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() // The runway result carries full data (it crosses the service boundary). Its diff --git a/submitqueue/orchestrator/controller/score/score.go b/submitqueue/orchestrator/controller/score/score.go index 055a2638..591a0b44 100644 --- a/submitqueue/orchestrator/controller/score/score.go +++ b/submitqueue/orchestrator/controller/score/score.go @@ -47,6 +47,8 @@ type Controller struct { // Verify Controller implements consumer.Controller interface at compile time. var _ consumer.Controller = (*Controller)(nil) +const opName = "process" + // NewController creates a new score controller for the orchestrator. func NewController( logger *zap.SugaredLogger, @@ -73,12 +75,7 @@ func NewController( // persists the minimum score, publishes request log entries, // and publishes to the speculate topic. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - const opName = "process" - - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() // Deserialize batch ID from payload @@ -109,7 +106,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // controller has handed the batch off to speculate, which owns the // terminal write and downstream fanout. if batch.State == entity.BatchStateCancelling { - c.metricsScope.Counter("skipped_cancelling").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "skipped_cancelling", 1) c.logger.Infow("skipping score for cancelling batch", "batch_id", batch.ID, ) @@ -119,7 +116,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // Score owns no terminal-state recovery. The controller that wrote the // terminal state owns its remaining fanout. if batch.State.IsTerminal() { - c.metricsScope.Counter("skipped_terminal").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "skipped_terminal", 1) c.logger.Infow("skipping score for terminal batch", "batch_id", batch.ID, "state", string(batch.State), @@ -168,11 +165,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r // published to speculate before its acknowledgement was recorded. // Downstream processing has already advanced the batch, so this stale // delivery is satisfied and must not regress the batch to Scored. - c.metricsScope.Counter("skipped_downstream").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "skipped_downstream", 1) return nil default: - c.metricsScope.Counter("unexpected_state").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "unexpected_state", 1) return fmt.Errorf("unexpected batch state %q for batch %s", batch.State, batch.ID) } diff --git a/submitqueue/orchestrator/controller/speculate/speculate.go b/submitqueue/orchestrator/controller/speculate/speculate.go index 4a92885c..2d46978f 100644 --- a/submitqueue/orchestrator/controller/speculate/speculate.go +++ b/submitqueue/orchestrator/controller/speculate/speculate.go @@ -89,10 +89,7 @@ func NewController( // Process advances a batch one step along the naive happy-path. // Returns nil to ack (success), or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() bid, err := entity.BatchIDFromBytes(msg.Payload) @@ -196,7 +193,7 @@ func (c *Controller) tryFinalize(ctx context.Context, batch entity.Batch) error case entity.BatchStateCancelled: // Out-of-the-way: the cancelled batch will never land, so it can // no longer conflict. Drop it from the chain and continue. - c.metricsScope.Counter("dependency_cancelled_skipped").Inc(1) + metrics.NamedCounter(c.metricsScope, opName, "dependency_cancelled_skipped", 1) c.logger.Infow("dependency cancelled; dropping from speculation chain", "batch_id", batch.ID, "dependency_id", d.ID, diff --git a/submitqueue/orchestrator/controller/start/start.go b/submitqueue/orchestrator/controller/start/start.go index a6796406..a37e4244 100644 --- a/submitqueue/orchestrator/controller/start/start.go +++ b/submitqueue/orchestrator/controller/start/start.go @@ -70,12 +70,9 @@ func NewController( // Process processes a request delivery from the queue. // Persists the request and publishes to validate. Returns nil to ack (success), // or error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { const opName = "process" - op := metrics.Begin(c.metricsScope, opName, metrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - msg := delivery.Message() landRequest, err := entity.LandRequestFromBytes(msg.Payload) diff --git a/submitqueue/orchestrator/controller/validate/validate.go b/submitqueue/orchestrator/controller/validate/validate.go index 7cafc95c..e0176168 100644 --- a/submitqueue/orchestrator/controller/validate/validate.go +++ b/submitqueue/orchestrator/controller/validate/validate.go @@ -88,10 +88,7 @@ func NewController( // Runs duplicate detection, change metadata fetch, and change claiming, then kicks off the // asynchronous merge-conflict check by publishing the full check request to runway. // Returns nil to ack (success or non-retryable rejection), error to nack (retry). -func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { - op := coremetrics.Begin(c.metricsScope, "process", coremetrics.LongLatencyBuckets) - defer func() { op.Complete(retErr) }() - +func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) error { msg := delivery.Message() // Deserialize request ID from payload