feat(metrics): add request lifecycle, queue-scoped stage, and batch shape metrics

platform/metrics/README.md

@@ -20,6 +20,7 @@ For any operation with a clear start and end, use `Begin` and `Complete`:
 |----------|-------|
 | `Begin(scope, name, buckets, ...tags)` | `{name}.start` counter +1 and returns an `Op` |
 | `op.Complete(err, ...tags)` | `{name}.finish` histogram tagged with `result=success\|error\|cancel` and any completion tags |
+| `op.CompleteAs(result, ...tags)` | `{name}.finish` histogram tagged with the given `result` and any completion tags, for outcomes the error alone does not classify (for example `result=not_found`) |
 
 `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.
 
@@ -79,13 +80,16 @@ metrics.NamedCounter(c.scope, "publish", "attempts", 1, metrics.NewTag("topic",
 
 ## Duration Buckets
 
-There is no default bucket set. The package exports four common sets:
+There is no default bucket set. The package exports five common sets:
 
 | Set | Range | Use for |
 |-----|-------|---------|
 | `FastLatencyBuckets` | ~100µs – 5s | Fast in-process work such as scoring, cache lookups, and CPU-bound operations |
 | `StorageLatencyBuckets` | ~1ms – 1m | Storage and message-queue round trips such as database reads, writes, publishing, and consuming |
 | `LongLatencyBuckets` | ~5ms – 4h | Long-running pipeline work and external calls such as builds, lands, pushes, and provider calls |
 | `ChangeAgeBuckets` | ~1m – 30d | Elapsed time measured from a source-control change's commit timestamp rather than from work this system started |
+| `LandLatencyBuckets` | ~1m – 24h | End-to-end request lifecycles, from accepting a land request to its terminal outcome |
 
 Pass one of these sets or a custom `tally.DurationBuckets` to `Begin` or `NamedHistogram`.
+
+For counts recorded with `RecordValue`, such as how many items a unit of work carries or depends on, use `CountBuckets` (0 – 144, Fibonacci-spaced, with zero in its own bucket).

platform/metrics/metrics.go

@@ -153,8 +153,40 @@ var (
 		14 * 24 * time.Hour,
 		30 * 24 * time.Hour,
 	}
+
+	// LandLatencyBuckets suits end-to-end request lifecycles, from accepting a
+	// land request to its terminal outcome. Resolution is densest between 5 and
+	// 60 minutes, where typical lands fall, and the tail reaches a day so that
+	// requests stuck overnight stay distinguishable instead of overflowing.
+	LandLatencyBuckets = tally.DurationBuckets{
+		1 * time.Minute,
+		2 * time.Minute,
+		3 * time.Minute,
+		5 * time.Minute,
+		7*time.Minute + 30*time.Second,
+		10 * time.Minute,
+		15 * time.Minute,
+		20 * time.Minute,
+		30 * time.Minute,
+		45 * time.Minute,
+		1 * time.Hour,
+		90 * time.Minute,
+		2 * time.Hour,
+		3 * time.Hour,
+		4 * time.Hour,
+		6 * time.Hour,
+		8 * time.Hour,
+		12 * time.Hour,
+		24 * time.Hour,
+	}
 )
 
+// CountBuckets suits small cardinalities recorded with RecordValue, such as how
+// many items a unit of work carries or depends on. Zero has its own bucket
+// because "none" is a meaningful answer, and the Fibonacci spacing keeps
+// resolution where counts usually fall while still separating outliers.
+var CountBuckets = tally.ValueBuckets{0, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144}
+
 // Op tracks the lifecycle of a named operation. It captures the start time on
 // creation, emits a {name}.start counter, and records the duration and result
 // on a {name}.finish histogram when Complete is called.
@@ -200,6 +232,13 @@ func (o Op) Complete(err error, tags ...Tag) {
 		}
 	}
 
+	o.CompleteAs(result, tags...)
+}
+
+// CompleteAs records elapsed time on the {name}.finish histogram like Complete,
+// but with a caller-chosen result tag. It serves outcomes that the error alone
+// does not classify, such as a lookup that found no record.
+func (o Op) CompleteAs(result string, tags ...Tag) {
 	tagged(o.scope, tags).
 		Tagged(map[string]string{"result": result}).
 		Histogram("finish", o.buckets).

platform/metrics/metrics_test.go

@@ -125,6 +125,46 @@ func TestOp_CompleteWithTags(t *testing.T) {
 	assert.True(t, ok, "expected finish-only tags, got keys: %v", histogramKeys(histograms))
 }
 
+func TestOp_CompleteAs(t *testing.T) {
+	tests := []struct {
+		name         string
+		result       string
+		tags         []Tag
+		histogramKey string
+	}{
+		{
+			name:         "records the given result",
+			result:       "not_found",
+			histogramKey: "process.finish+result=not_found",
+		},
+		{
+			name:         "records completion tags alongside the result",
+			result:       "not_found",
+			tags:         []Tag{NewTag("store", "build")},
+			histogramKey: "process.finish+result=not_found,store=build",
+		},
+		{
+			name:         "result tag wins over a completion tag of the same key",
+			result:       "not_found",
+			tags:         []Tag{NewTag("result", "success")},
+			histogramKey: "process.finish+result=not_found",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scope := tally.NewTestScope("", nil)
+			op := Begin(scope, "process", StorageLatencyBuckets)
+			op.CompleteAs(tt.result, tt.tags...)
+
+			histograms := scope.Snapshot().Histograms()
+			_, ok := histograms[tt.histogramKey]
+			assert.True(t, ok, "expected histogram key %s, got keys: %v", tt.histogramKey, histogramKeys(histograms))
+			assert.Len(t, histograms, 1)
+		})
+	}
+}
+
 func TestNamedCounter(t *testing.T) {
 	scope := tally.NewTestScope("", nil)
 	NamedCounter(scope, "publish", "attempts", 5)
@@ -180,6 +220,7 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 		"StorageLatencyBuckets": StorageLatencyBuckets,
 		"LongLatencyBuckets":    LongLatencyBuckets,
 		"ChangeAgeBuckets":      ChangeAgeBuckets,
+		"LandLatencyBuckets":    LandLatencyBuckets,
 	}
 	for name, buckets := range sets {
 		t.Run(name, func(t *testing.T) {
@@ -192,6 +233,13 @@ func TestLatencyBuckets_Sorted(t *testing.T) {
 	}
 }
 
+func TestCountBuckets_Sorted(t *testing.T) {
+	for i := 1; i < len(CountBuckets); i++ {
+		assert.Greater(t, CountBuckets[i], CountBuckets[i-1],
+			"CountBuckets[%d] (%v) must be greater than CountBuckets[%d] (%v)", i, CountBuckets[i], i-1, CountBuckets[i-1])
+	}
+}
+
 // counterKeys extracts map keys for error messages.
 func counterKeys(m map[string]tally.CounterSnapshot) []string {
 	keys := make([]string, 0, len(m))

service/submitqueue/gateway/server/main.go

@@ -366,7 +366,7 @@ func run() error {
 	// Create controllers and wrap them for gRPC. Every store is queue-scoped and
 	// resolves through the factory adapter; land/cancel/log share one materializer.
 	storageFty := storageFactory{backend: store}
-	materializer := requestcore.NewMaterializer(storageFty)
+	materializer := requestcore.NewMaterializer(storageFty, requestcore.WithMetricsScope(scope))
 	pingController := controller.NewPingController(logger, scope)
 	landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry)
 	cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry)

submitqueue/extension/storage/BUILD.bazel

@@ -22,6 +22,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//platform/errs:go_default_library",
+        "//platform/metrics:go_default_library",
         "//submitqueue/entity:go_default_library",
     ],
 )
@@ -32,7 +33,9 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//platform/errs:go_default_library",
+        "//platform/metrics:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
     ],
 )
 

submitqueue/extension/storage/storage.go

@@ -25,6 +25,7 @@ import (
 	"fmt"
 
 	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/metrics"
 )
 
 // ErrNotFound is returned by storage implementations when the requested record is not found in the database.
@@ -40,6 +41,22 @@ func WrapNotFound(err error) error {
 	return fmt.Errorf("%w: %w", ErrNotFound, err)
 }
 
+// MetricsResultNotFound is the result tag CompleteReadMetricsOp records for a missing record.
+const MetricsResultNotFound = "not_found"
+
+// CompleteReadMetricsOp finishes a store read's metrics. A read that finds no
+// record has answered the question rather than failed, so ErrNotFound is recorded
+// as result=not_found instead of result=error; every other outcome is classified
+// by metrics.Op.Complete. Writes finish with metrics.Op.Complete directly: a write
+// whose row is missing is an anomaly and stays counted as an error.
+func CompleteReadMetricsOp(op metrics.Op, err error) {
+	if IsNotFound(err) {
+		op.CompleteAs(MetricsResultNotFound)
+		return
+	}
+	op.Complete(err)
+}
+
 // ErrAlreadyExists is returned by storage implementations when attempting to create a record with an ID that already exists.
 var ErrAlreadyExists = errors.New("record already exists")
 

submitqueue/extension/storage/storage_test.go

@@ -15,11 +15,14 @@
 package storage
 
 import (
+	"database/sql"
 	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/uber-go/tally"
 	"github.com/uber/submitqueue/platform/errs"
+	"github.com/uber/submitqueue/platform/metrics"
 )
 
 func TestErrVersionMismatchClassification(t *testing.T) {
@@ -30,3 +33,28 @@ func TestErrVersionMismatchClassification(t *testing.T) {
 	assert.False(t, errs.IsUserError(err))
 	assert.False(t, errs.IsDependencyError(err))
 }
+
+func TestCompleteReadMetricsOp(t *testing.T) {
+	tests := []struct {
+		name   string
+		err    error
+		result string
+	}{
+		{name: "success", result: "success"},
+		{name: "wrapped not found", err: WrapNotFound(sql.ErrNoRows), result: MetricsResultNotFound},
+		{name: "not found wrapped by a caller", err: fmt.Errorf("get build: %w", ErrNotFound), result: MetricsResultNotFound},
+		{name: "other error", err: fmt.Errorf("connection refused"), result: "error"},
+		{name: "version mismatch", err: ErrVersionMismatch, result: "error"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			scope := tally.NewTestScope("", nil)
+			CompleteReadMetricsOp(metrics.Begin(scope, "get", metrics.StorageLatencyBuckets), tt.err)
+
+			histograms := scope.Snapshot().Histograms()
+			assert.Len(t, histograms, 1)
+			assert.Contains(t, histograms, "get.finish+result="+tt.result)
+		})
+	}
+}

submitqueue/gateway/core/request/BUILD.bazel

@@ -10,9 +10,11 @@ go_library(
     importpath = "github.com/uber/submitqueue/submitqueue/gateway/core/request",
     visibility = ["//visibility:public"],
     deps = [
+        "//platform/metrics:go_default_library",
         "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/gateway/extension/storage:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
     ],
 )
 
@@ -31,6 +33,7 @@ go_test(
         "//submitqueue/gateway/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
+        "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_mock//gomock:go_default_library",
     ],
 )

submitqueue/gateway/core/request/materializer.go

@@ -19,7 +19,10 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"time"
 
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/platform/metrics"
 	"github.com/uber/submitqueue/submitqueue/entity"
 	basestorage "github.com/uber/submitqueue/submitqueue/extension/storage"
 	gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
@@ -30,12 +33,26 @@ import (
 // Every store it touches is queue-scoped, so each call resolves the aggregate once
 // from the queue carried on the log being persisted.
 type Materializer struct {
-	stores gwstorage.Factory
+	stores       gwstorage.Factory
+	metricsScope tally.Scope
+}
+
+// MaterializerOption configures a Materializer.
+type MaterializerOption func(*Materializer)
+
+// WithMetricsScope emits request lifecycle metrics on scope. Without it the
+// materializer emits none.
+func WithMetricsScope(scope tally.Scope) MaterializerOption {
+	return func(m *Materializer) { m.metricsScope = scope }
 }
 
 // NewMaterializer creates a request read-model materializer.
-func NewMaterializer(stores gwstorage.Factory) *Materializer {
-	return &Materializer{stores: stores}
+func NewMaterializer(stores gwstorage.Factory, opts ...MaterializerOption) *Materializer {
+	m := &Materializer{stores: stores, metricsScope: tally.NoopScope}
+	for _, opt := range opts {
+		opt(m)
+	}
+	return m
 }
 
 // PersistLog appends one audit log and materializes its winning state.
@@ -76,6 +93,7 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 				return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err)
 			}
 			updated.Version = newVersion
+			m.recordLifecycleTransition(summary, updated)
 			summary = updated
 		}
 
@@ -92,6 +110,38 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
 	}
 }
 
+// recordLifecycleTransition counts a request's public lifecycle milestones from
+// one successful summary update.
+//
+// Each milestone is recorded on the update that crosses it, never on a winning
+// log that merely restates it: an orchestrator retry re-publishes a terminal log
+// with the same request version and a newer timestamp, which wins logWins and
+// rewrites the summary again. Comparing the summary before and after the update
+// is what keeps each request counted once.
+func (m *Materializer) recordLifecycleTransition(before, after entity.RequestSummary) {
+	queueTag := metrics.NewTag("queue", after.Queue)
+	if before.Status == entity.RequestStatusAccepting && after.Status != entity.RequestStatusAccepting {
+		metrics.NamedCounter(m.metricsScope, "request_lifecycle", "accepted", 1, queueTag)
+	}
+	if isTerminalStatus(before.Status) || !isTerminalStatus(after.Status) {
+		return
+	}
+	statusTag := metrics.NewTag("status", string(after.Status))
+	metrics.NamedCounter(m.metricsScope, "request_lifecycle", "terminated", 1,
+		queueTag, statusTag, metrics.NewTag("from_status", string(before.Status)))
+	if before.ReceivedAtMs <= 0 {
+		return
+	}
+	// Clock skew between the gateway and the orchestrator can put the terminal timestamp before receipt.
+	elapsed := max(time.Duration(after.StatusTimestampMs-before.ReceivedAtMs)*time.Millisecond, 0)
+	metrics.NamedHistogram(m.metricsScope, "request_lifecycle", "end_to_end_latency", metrics.LandLatencyBuckets,
+		queueTag, statusTag).RecordDuration(elapsed)
+}
+
+func isTerminalStatus(status entity.RequestStatus) bool {
+	return entity.IsRequestStateTerminal(entity.RequestState(status))
+}
+
 // ensureRequestURIMappings retries every immutable mapping, including after partial activation.
 func ensureRequestURIMappings(ctx context.Context, uris basestorage.RequestURIStore, summary entity.RequestSummary) error {
 	for _, changeURI := range summary.ChangeURIs {