Skip to content

Commit e441df5

Browse files
mnoah1mnoah1
committed
refactor(stovepipe): persist accepted log through materializer
1 parent aa93da1 commit e441df5

4 files changed

Lines changed: 40 additions & 40 deletions

File tree

‎service/stovepipe/server/main.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -303,7 +303,7 @@ func run() error {
303303
brf := fakeBuildRunnerFactory{}
304304

305305
storageFty := storageFactory{backend: store}
306-
requestLog := requestlog.NewRecorder(scope)
306+
materializer := requestlog.NewMaterializer(scope)
307307
primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, storageFty, registry, sourceControl, brf, hookResolver{})
308308
if err != nil {
309309
return err
@@ -338,7 +338,7 @@ func run() error {
338338
newInMemoryCounterFactory(),
339339
sourceControl,
340340
storageFty,
341-
requestLog,
341+
materializer,
342342
registry,
343343
)
344344
srv := &StovepipeServer{

‎stovepipe/controller/BUILD.bazel‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ go_test(
4141
"//platform/extension/counter/mock:go_default_library",
4242
"//platform/extension/messagequeue/mock:go_default_library",
4343
"//stovepipe/core/messagequeue:go_default_library",
44+
"//stovepipe/core/requestlog:go_default_library",
4445
"//stovepipe/core/requestlog/mock:go_default_library",
4546
"//stovepipe/entity:go_default_library",
4647
"//stovepipe/extension/sourcecontrol:go_default_library",

‎stovepipe/controller/ingest.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ type IngestController struct {
6161
counters counter.Factory
6262
sourceControl sourcecontrol.Factory
6363
stores storage.Factory
64-
requestLog requestlog.Recorder
64+
materializer requestlog.Materializer
6565
registry consumer.TopicRegistry
6666
}
6767

@@ -73,7 +73,7 @@ func NewIngestController(
7373
counters counter.Factory,
7474
sourceControl sourcecontrol.Factory,
7575
stores storage.Factory,
76-
requestLog requestlog.Recorder,
76+
materializer requestlog.Materializer,
7777
registry consumer.TopicRegistry,
7878
) *IngestController {
7979
return &IngestController{
@@ -82,7 +82,7 @@ func NewIngestController(
8282
counters: counters,
8383
sourceControl: sourceControl,
8484
stores: stores,
85-
requestLog: requestLog,
85+
materializer: materializer,
8686
registry: registry,
8787
}
8888
}
@@ -141,7 +141,8 @@ func (c *IngestController) Ingest(ctx context.Context, req entity.IngestRequest)
141141
}
142142

143143
if request.State == entity.RequestStateAccepted {
144-
if err := c.requestLog.RecordRequestState(ctx, store.GetRequestLogStore(), request, entity.RequestOutcomeReasonUnknown); err != nil {
144+
log := requestlog.NewRequestStateLog(request, entity.RequestOutcomeReasonUnknown)
145+
if err := c.materializer.PersistLog(ctx, store, log); err != nil {
145146
return entity.IngestResult{}, fmt.Errorf("failed to record accepted state for request %s: %w", id, err)
146147
}
147148
}

‎stovepipe/controller/ingest_test.go‎

Lines changed: 32 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import (
2828
countermock "github.com/uber/submitqueue/platform/extension/counter/mock"
2929
mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
3030
stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue"
31+
"github.com/uber/submitqueue/stovepipe/core/requestlog"
3132
requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock"
3233
"github.com/uber/submitqueue/stovepipe/entity"
3334
"github.com/uber/submitqueue/stovepipe/extension/sourcecontrol"
@@ -45,15 +46,15 @@ const (
4546

4647
// ingestMocks bundles the mocks an Ingest test case wires expectations on.
4748
type ingestMocks struct {
48-
counter *countermock.MockCounter
49-
factory *scmock.MockFactory
50-
sc *scmock.MockSourceControl
51-
reqStore *storagemock.MockRequestStore
52-
uriStore *storagemock.MockRequestURIStore
53-
logStore *storagemock.MockRequestLogStore
54-
queueStore *storagemock.MockQueueStore
55-
requestLog *requestlogmock.MockRecorder
56-
publisher *mqmock.MockPublisher
49+
counter *countermock.MockCounter
50+
factory *scmock.MockFactory
51+
sc *scmock.MockSourceControl
52+
reqStore *storagemock.MockRequestStore
53+
uriStore *storagemock.MockRequestURIStore
54+
queueStore *storagemock.MockQueueStore
55+
store *storagemock.MockStorage
56+
materializer *requestlogmock.MockMaterializer
57+
publisher *mqmock.MockPublisher
5758
}
5859

5960
// staticStorageFactory resolves every queue to one fixed store aggregate.
@@ -72,21 +73,20 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
7273
t.Helper()
7374

7475
m := ingestMocks{
75-
counter: countermock.NewMockCounter(ctrl),
76-
factory: scmock.NewMockFactory(ctrl),
77-
sc: scmock.NewMockSourceControl(ctrl),
78-
reqStore: storagemock.NewMockRequestStore(ctrl),
79-
uriStore: storagemock.NewMockRequestURIStore(ctrl),
80-
logStore: storagemock.NewMockRequestLogStore(ctrl),
81-
queueStore: storagemock.NewMockQueueStore(ctrl),
82-
requestLog: requestlogmock.NewMockRecorder(ctrl),
83-
publisher: mqmock.NewMockPublisher(ctrl),
76+
counter: countermock.NewMockCounter(ctrl),
77+
factory: scmock.NewMockFactory(ctrl),
78+
sc: scmock.NewMockSourceControl(ctrl),
79+
reqStore: storagemock.NewMockRequestStore(ctrl),
80+
uriStore: storagemock.NewMockRequestURIStore(ctrl),
81+
queueStore: storagemock.NewMockQueueStore(ctrl),
82+
materializer: requestlogmock.NewMockMaterializer(ctrl),
83+
publisher: mqmock.NewMockPublisher(ctrl),
8484
}
8585

8686
store := storagemock.NewMockStorage(ctrl)
87+
m.store = store
8788
store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes()
8889
store.EXPECT().GetRequestURIStore().Return(m.uriStore).AnyTimes()
89-
store.EXPECT().GetRequestLogStore().Return(m.logStore).AnyTimes()
9090
store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes()
9191

9292
queue := mqmock.NewMockQueue(ctrl)
@@ -97,7 +97,7 @@ func newIngestController(t *testing.T, ctrl *gomock.Controller) (*IngestControll
9797
})
9898
require.NoError(t, err)
9999

100-
c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, m.requestLog, registry)
100+
c := NewIngestController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), staticCounterFactory{counter: m.counter}, m.factory, staticStorageFactory{store: store}, m.materializer, registry)
101101
return c, m
102102
}
103103

@@ -111,12 +111,11 @@ func acceptedRequest(id string) entity.Request {
111111
}
112112
}
113113

114-
func expectRecordAccepted(m ingestMocks, id string) {
115-
m.requestLog.EXPECT().RecordRequestState(
114+
func expectMaterializeAccepted(m ingestMocks, id string) {
115+
m.materializer.EXPECT().PersistLog(
116116
gomock.Any(),
117-
m.logStore,
118-
acceptedRequest(id),
119-
entity.RequestOutcomeReasonUnknown,
117+
m.store,
118+
requestlog.NewRequestStateLog(acceptedRequest(id), entity.RequestOutcomeReasonUnknown),
120119
).Return(nil)
121120
}
122121

@@ -178,7 +177,7 @@ func TestIngestController_Ingest(t *testing.T) {
178177
m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(nil)
179178
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/7").Return(entity.Request{}, storage.ErrNotFound)
180179
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
181-
expectRecordAccepted(m, "request/monorepo/main/7")
180+
expectMaterializeAccepted(m, "request/monorepo/main/7")
182181
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
183182
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
184183
},
@@ -191,7 +190,7 @@ func TestIngestController_Ingest(t *testing.T) {
191190
expectResolve(m)
192191
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
193192
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
194-
expectRecordAccepted(m, "request/monorepo/main/3")
193+
expectMaterializeAccepted(m, "request/monorepo/main/3")
195194
expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
196195
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
197196
},
@@ -205,7 +204,7 @@ func TestIngestController_Ingest(t *testing.T) {
205204
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
206205
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(entity.Request{}, storage.ErrNotFound)
207206
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
208-
expectRecordAccepted(m, "request/monorepo/main/3")
207+
expectMaterializeAccepted(m, "request/monorepo/main/3")
209208
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/3")
210209
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
211210
},
@@ -221,7 +220,7 @@ func TestIngestController_Ingest(t *testing.T) {
221220
m.uriStore.EXPECT().Create(gomock.Any(), testURI, "request/monorepo/main/7").Return(storage.ErrAlreadyExists)
222221
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
223222
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
224-
expectRecordAccepted(m, "request/monorepo/main/3")
223+
expectMaterializeAccepted(m, "request/monorepo/main/3")
225224
expectAdvanceLatestRequestIDNoOp(m, testQueue, "request/monorepo/main/3")
226225
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(nil)
227226
},
@@ -286,7 +285,7 @@ func TestIngestController_Ingest(t *testing.T) {
286285
m.uriStore.EXPECT().Create(gomock.Any(), testURI, gomock.Any()).Return(nil)
287286
m.reqStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.Request{}, storage.ErrNotFound)
288287
m.reqStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil)
289-
expectRecordAccepted(m, "request/monorepo/main/7")
288+
expectMaterializeAccepted(m, "request/monorepo/main/7")
290289
expectAdvanceLatestRequestID(m, testQueue, "request/monorepo/main/7")
291290
m.publisher.EXPECT().Publish(gomock.Any(), "process", gomock.Any()).Return(errors.New("queue down"))
292291
},
@@ -299,11 +298,10 @@ func TestIngestController_Ingest(t *testing.T) {
299298
expectResolve(m)
300299
m.uriStore.EXPECT().GetIDByURI(gomock.Any(), testURI).Return("request/monorepo/main/3", nil)
301300
m.reqStore.EXPECT().Get(gomock.Any(), "request/monorepo/main/3").Return(acceptedRequest("request/monorepo/main/3"), nil)
302-
m.requestLog.EXPECT().RecordRequestState(
301+
m.materializer.EXPECT().PersistLog(
303302
gomock.Any(),
304-
m.logStore,
305-
acceptedRequest("request/monorepo/main/3"),
306-
entity.RequestOutcomeReasonUnknown,
303+
m.store,
304+
requestlog.NewRequestStateLog(acceptedRequest("request/monorepo/main/3"), entity.RequestOutcomeReasonUnknown),
307305
).Return(errors.New("log unavailable"))
308306
},
309307
wantErr: true,

0 commit comments

Comments
 (0)