Skip to content

Commit 8f67e53

Browse files
roychyingroychying
committed
feat(gateway): Emit request lifecycle metrics for acceptance, terminal outcome, and end-to-end land time
1 parent b963ac3 commit 8f67e53

5 files changed

Lines changed: 158 additions & 5 deletions

File tree

‎service/submitqueue/gateway/server/main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -366,7 +366,7 @@ func run() error {
366366
// Create controllers and wrap them for gRPC. Every store is queue-scoped and
367367
// resolves through the factory adapter; land/cancel/log share one materializer.
368368
storageFty := storageFactory{backend: store}
369-
materializer := requestcore.NewMaterializer(storageFty)
369+
materializer := requestcore.NewMaterializer(storageFty, requestcore.WithMetricsScope(scope))
370370
pingController := controller.NewPingController(logger, scope)
371371
landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry)
372372
cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry)

‎submitqueue/gateway/core/request/BUILD.bazel‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,11 @@ go_library(
1010
importpath = "github.com/uber/submitqueue/submitqueue/gateway/core/request",
1111
visibility = ["//visibility:public"],
1212
deps = [
13+
"//platform/metrics:go_default_library",
1314
"//submitqueue/entity:go_default_library",
1415
"//submitqueue/extension/storage:go_default_library",
1516
"//submitqueue/gateway/extension/storage:go_default_library",
17+
"@com_github_uber_go_tally//:go_default_library",
1618
],
1719
)
1820

@@ -31,6 +33,7 @@ go_test(
3133
"//submitqueue/gateway/extension/storage/mock:go_default_library",
3234
"@com_github_stretchr_testify//assert:go_default_library",
3335
"@com_github_stretchr_testify//require:go_default_library",
36+
"@com_github_uber_go_tally//:go_default_library",
3437
"@org_uber_go_mock//gomock:go_default_library",
3538
],
3639
)

‎submitqueue/gateway/core/request/materializer.go‎

Lines changed: 53 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,10 @@ import (
1919
"errors"
2020
"fmt"
2121
"maps"
22+
"time"
2223

24+
"github.com/uber-go/tally"
25+
"github.com/uber/submitqueue/platform/metrics"
2326
"github.com/uber/submitqueue/submitqueue/entity"
2427
basestorage "github.com/uber/submitqueue/submitqueue/extension/storage"
2528
gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage"
@@ -30,12 +33,26 @@ import (
3033
// Every store it touches is queue-scoped, so each call resolves the aggregate once
3134
// from the queue carried on the log being persisted.
3235
type Materializer struct {
33-
stores gwstorage.Factory
36+
stores gwstorage.Factory
37+
metricsScope tally.Scope
38+
}
39+
40+
// MaterializerOption configures a Materializer.
41+
type MaterializerOption func(*Materializer)
42+
43+
// WithMetricsScope emits request lifecycle metrics on scope. Without it the
44+
// materializer emits none.
45+
func WithMetricsScope(scope tally.Scope) MaterializerOption {
46+
return func(m *Materializer) { m.metricsScope = scope }
3447
}
3548

3649
// NewMaterializer creates a request read-model materializer.
37-
func NewMaterializer(stores gwstorage.Factory) *Materializer {
38-
return &Materializer{stores: stores}
50+
func NewMaterializer(stores gwstorage.Factory, opts ...MaterializerOption) *Materializer {
51+
m := &Materializer{stores: stores, metricsScope: tally.NoopScope}
52+
for _, opt := range opts {
53+
opt(m)
54+
}
55+
return m
3956
}
4057

4158
// 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
7693
return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err)
7794
}
7895
updated.Version = newVersion
96+
m.recordLifecycleTransition(summary, updated)
7997
summary = updated
8098
}
8199

@@ -92,6 +110,38 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er
92110
}
93111
}
94112

113+
// recordLifecycleTransition counts a request's public lifecycle milestones from
114+
// one successful summary update.
115+
//
116+
// Each milestone is recorded on the update that crosses it, never on a winning
117+
// log that merely restates it: an orchestrator retry re-publishes a terminal log
118+
// with the same request version and a newer timestamp, which wins logWins and
119+
// rewrites the summary again. Comparing the summary before and after the update
120+
// is what keeps each request counted once.
121+
func (m *Materializer) recordLifecycleTransition(before, after entity.RequestSummary) {
122+
queueTag := metrics.NewTag("queue", after.Queue)
123+
if before.Status == entity.RequestStatusAccepting && after.Status != entity.RequestStatusAccepting {
124+
metrics.NamedCounter(m.metricsScope, "request_lifecycle", "accepted", 1, queueTag)
125+
}
126+
if isTerminalStatus(before.Status) || !isTerminalStatus(after.Status) {
127+
return
128+
}
129+
statusTag := metrics.NewTag("status", string(after.Status))
130+
metrics.NamedCounter(m.metricsScope, "request_lifecycle", "terminated", 1,
131+
queueTag, statusTag, metrics.NewTag("from_status", string(before.Status)))
132+
if before.ReceivedAtMs <= 0 {
133+
return
134+
}
135+
// Clock skew between the gateway and the orchestrator can put the terminal timestamp before receipt.
136+
elapsed := max(time.Duration(after.StatusTimestampMs-before.ReceivedAtMs)*time.Millisecond, 0)
137+
metrics.NamedHistogram(m.metricsScope, "request_lifecycle", "end_to_end_latency", metrics.LandLatencyBuckets,
138+
queueTag, statusTag).RecordDuration(elapsed)
139+
}
140+
141+
func isTerminalStatus(status entity.RequestStatus) bool {
142+
return entity.IsRequestStateTerminal(entity.RequestState(status))
143+
}
144+
95145
// ensureRequestURIMappings retries every immutable mapping, including after partial activation.
96146
func ensureRequestURIMappings(ctx context.Context, uris basestorage.RequestURIStore, summary entity.RequestSummary) error {
97147
for _, changeURI := range summary.ChangeURIs {

‎submitqueue/gateway/core/request/materializer_test.go‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,103 @@ func TestMaterializer_PersistLog(t *testing.T) {
6565
})
6666
}
6767

68+
func TestMaterializer_RecordsLifecycleMetrics(t *testing.T) {
69+
started := testRequestSummary()
70+
started.Status, started.RequestVersion, started.StatusTimestampMs = entity.RequestStatusStarted, 1, 15
71+
landed := started
72+
landed.Status, landed.RequestVersion, landed.StatusTimestampMs = entity.RequestStatusLanded, 2, 20
73+
unreceived := started
74+
unreceived.ReceivedAtMs = 0
75+
statusLog := func(status entity.RequestStatus, version int32, timestampMs int64) entity.RequestLog {
76+
return entity.RequestLog{Queue: "q", RequestID: "1", Type: entity.RequestLogTypeStatus, Status: status, RequestVersion: version, TimestampMs: timestampMs}
77+
}
78+
79+
tests := []struct {
80+
name string
81+
before entity.RequestSummary
82+
log entity.RequestLog
83+
wantUpdate bool
84+
wantCounters map[string]int64
85+
wantHistograms []string
86+
}{
87+
{
88+
name: "accepted log moves the request out of accepting",
89+
before: testRequestSummary(),
90+
log: statusLog(entity.RequestStatusAccepted, 0, 20),
91+
wantUpdate: true,
92+
wantCounters: map[string]int64{"request_lifecycle.accepted+queue=q": 1},
93+
},
94+
{
95+
name: "a later status overtaking the accepted log still counts acceptance once",
96+
before: testRequestSummary(),
97+
log: statusLog(entity.RequestStatusStarted, 1, 20),
98+
wantUpdate: true,
99+
wantCounters: map[string]int64{"request_lifecycle.accepted+queue=q": 1},
100+
},
101+
{
102+
name: "terminal log finishes the request and records its land time",
103+
before: started,
104+
log: statusLog(entity.RequestStatusLanded, 2, 20),
105+
wantUpdate: true,
106+
wantCounters: map[string]int64{"request_lifecycle.terminated+from_status=started,queue=q,status=landed": 1},
107+
wantHistograms: []string{"request_lifecycle.end_to_end_latency+queue=q,status=landed"},
108+
},
109+
{
110+
name: "missing receipt time counts the finish without a land time",
111+
before: unreceived,
112+
log: statusLog(entity.RequestStatusError, 2, 20),
113+
wantUpdate: true,
114+
wantCounters: map[string]int64{"request_lifecycle.terminated+from_status=started,queue=q,status=error": 1},
115+
},
116+
{
117+
name: "re-published terminal log with a newer timestamp rewrites the summary without counting again",
118+
before: landed,
119+
log: statusLog(entity.RequestStatusLanded, 2, 30),
120+
wantUpdate: true,
121+
},
122+
{
123+
name: "identical terminal redelivery leaves the summary and the metrics alone",
124+
before: landed,
125+
log: statusLog(entity.RequestStatusLanded, 2, 20),
126+
},
127+
{
128+
name: "cancelling is not a finished status",
129+
before: started,
130+
log: statusLog(entity.RequestStatusCancelling, 0, 20),
131+
wantUpdate: true,
132+
},
133+
}
134+
for _, tt := range tests {
135+
t.Run(tt.name, func(t *testing.T) {
136+
f := newMaterializerReceiptFixture(gomock.NewController(t))
137+
f.logs.EXPECT().Insert(gomock.Any(), tt.log).Return(nil)
138+
f.summaries.EXPECT().Get(gomock.Any(), tt.before.RequestID).Return(tt.before, nil)
139+
if tt.wantUpdate {
140+
f.summaries.EXPECT().Update(gomock.Any(), gomock.Any(), tt.before.Version, tt.before.Version+1).Return(nil)
141+
}
142+
f.uris.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
143+
f.receipts.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
144+
145+
require.NoError(t, f.materializer.PersistLog(context.Background(), tt.log))
146+
147+
snapshot := f.metrics.Snapshot()
148+
counters := map[string]int64{}
149+
for key, counter := range snapshot.Counters() {
150+
counters[key] = counter.Value()
151+
}
152+
if tt.wantCounters == nil {
153+
tt.wantCounters = map[string]int64{}
154+
}
155+
assert.Equal(t, tt.wantCounters, counters)
156+
histograms := []string{}
157+
for key := range snapshot.Histograms() {
158+
histograms = append(histograms, key)
159+
}
160+
assert.ElementsMatch(t, tt.wantHistograms, histograms)
161+
})
162+
}
163+
}
164+
68165
func TestMaterializer_PersistenceFailures(t *testing.T) {
69166
writeErr := errors.New("storage unavailable")
70167
tests := []struct {

‎submitqueue/gateway/core/request/request_receipt_test.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"testing"
2121

2222
"github.com/stretchr/testify/require"
23+
"github.com/uber-go/tally"
2324
"github.com/uber/submitqueue/submitqueue/entity"
2425
"github.com/uber/submitqueue/submitqueue/extension/storage"
2526
storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock"
@@ -33,13 +34,15 @@ type materializerReceiptFixture struct {
3334
uris *storagemock.MockRequestURIStore
3435
logs *storagemock.MockRequestLogStore
3536
receipts *storagemock.MockRequestReceiptStore
37+
metrics tally.TestScope
3638
}
3739

3840
func newMaterializerReceiptFixture(ctrl *gomock.Controller) materializerReceiptFixture {
3941
f := materializerReceiptFixture{
4042
summaries: storagemock.NewMockRequestSummaryStore(ctrl),
4143
uris: storagemock.NewMockRequestURIStore(ctrl), logs: storagemock.NewMockRequestLogStore(ctrl),
4244
receipts: storagemock.NewMockRequestReceiptStore(ctrl),
45+
metrics: tally.NewTestScope("", nil),
4346
}
4447
stores := gwstoragemock.NewMockStorage(ctrl)
4548
stores.EXPECT().GetRequestSummaryStore().Return(f.summaries).AnyTimes()
@@ -48,7 +51,7 @@ func newMaterializerReceiptFixture(ctrl *gomock.Controller) materializerReceiptF
4851
stores.EXPECT().GetRequestReceiptStore().Return(f.receipts).AnyTimes()
4952
factory := gwstoragemock.NewMockFactory(ctrl)
5053
factory.EXPECT().For(gomock.Any()).Return(stores, nil).AnyTimes()
51-
f.materializer = NewMaterializer(factory)
54+
f.materializer = NewMaterializer(factory, WithMetricsScope(f.metrics))
5255
return f
5356
}
5457

0 commit comments

Comments
 (0)