diff --git a/Makefile b/Makefile index b15be5161..bf8253679 100644 --- a/Makefile +++ b/Makefile @@ -390,6 +390,10 @@ local-init-submitqueue-schemas: ## Manually apply all database schemas echo " - Applying $$(basename $$file)..."; \ docker exec -i $(SUBMITQUEUE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \ done + @for file in submitqueue/gateway/extension/storage/mysql/schema/*.sql; do \ + echo " - Applying $$(basename $$file)..."; \ + docker exec -i $(SUBMITQUEUE_LOCAL_PROJECT)-mysql-app-1 mysql -uroot -proot submitqueue < $$file 2>&1 | grep -v "Using a password" || true; \ + done @echo "Applying counter schema to mysql-app..." @for file in platform/extension/counter/mysql/schema/*.sql; do \ echo " - Applying $$(basename $$file)..."; \ @@ -579,7 +583,7 @@ local-stovepipe-stop: ## Stop the Stovepipe service mocks: ## Generate mock files using mockgen @echo "Generating mocks..." - @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... + @$(BAZEL) run @rules_go//go -- generate ./submitqueue/extension/storage/... ./submitqueue/gateway/extension/storage/... ./submitqueue/extension/buildrunner/... ./submitqueue/extension/changeprovider/... ./platform/extension/counter/... ./platform/extension/consumergate/... ./platform/extension/hook/... ./platform/extension/messagequeue/... ./submitqueue/extension/queueconfig/... ./runway/extension/merger/... ./submitqueue/extension/conflict/... ./submitqueue/extension/speculation/... ./submitqueue/extension/validator/... ./platform/consumer/... ./stovepipe/core/requestlog/... ./stovepipe/extension/storage/... ./stovepipe/extension/sourcecontrol/... @echo "Mocks generated successfully!" proto: ## Generate protobuf files from .proto definitions diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index ac641e0df..480f237df 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -28,10 +28,10 @@ go_library( "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/queueconfig/yaml:go_default_library", - "//submitqueue/extension/storage:go_default_library", - "//submitqueue/extension/storage/mysql:go_default_library", "//submitqueue/gateway/controller:go_default_library", "//submitqueue/gateway/controller/log:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage/mysql:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index e5df77b74..9b1a061da 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -45,10 +45,10 @@ import ( requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" - "github.com/uber/submitqueue/submitqueue/extension/storage" - mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" "github.com/uber/submitqueue/submitqueue/gateway/controller" logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + mysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql" "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -375,7 +375,7 @@ func run() error { // directly so reconciliation does not depend on another asynchronous path. logConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, errs.NewClassifierProcessor( - // Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) + // Storage (submitqueue/gateway/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) // both run on the same MySQL driver, so a single classifier covers // errors surfaced from either backend. genericerrs.Classifier, diff --git a/submitqueue/core/request/BUILD.bazel b/submitqueue/core/request/BUILD.bazel index 1d5db35e0..ad473bbe7 100644 --- a/submitqueue/core/request/BUILD.bazel +++ b/submitqueue/core/request/BUILD.bazel @@ -17,6 +17,7 @@ go_library( "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", ], ) @@ -38,6 +39,7 @@ go_test( "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", + "//submitqueue/gateway/extension/storage/mock:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", "@org_uber_go_mock//gomock:go_default_library", diff --git a/submitqueue/core/request/materializer.go b/submitqueue/core/request/materializer.go index d470ed416..ee99dd214 100644 --- a/submitqueue/core/request/materializer.go +++ b/submitqueue/core/request/materializer.go @@ -22,7 +22,8 @@ import ( "slices" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" ) // Materializer appends request logs and projects the winning public request state. @@ -70,7 +71,7 @@ func (m *Materializer) PersistLog(ctx context.Context, log entity.RequestLog) er updated.Metadata = cloneMetadata(log.Metadata) if err := summaries.Update(ctx, updated, oldVersion, newVersion); err != nil { - if errors.Is(err, storage.ErrVersionMismatch) { + if errors.Is(err, basestorage.ErrVersionMismatch) { continue } return fmt.Errorf("failed to update request summary request_id=%s: %w", log.RequestID, err) @@ -93,12 +94,12 @@ func (m *Materializer) repairPublicProjections(ctx context.Context, stores stora queueSummaries := stores.GetRequestQueueSummaryStore() for { current, err := queueSummaries.Get(ctx, desired.ReceivedAtMs, desired.RequestID) - if errors.Is(err, storage.ErrNotFound) { + if errors.Is(err, basestorage.ErrNotFound) { if err := m.createURIMappings(ctx, stores, authoritative); err != nil { return err } if err := queueSummaries.Create(ctx, desired); err != nil { - if errors.Is(err, storage.ErrAlreadyExists) { + if errors.Is(err, basestorage.ErrAlreadyExists) { continue } return fmt.Errorf("failed to recreate queue summary request_id=%s: %w", desired.RequestID, err) @@ -116,7 +117,7 @@ func (m *Materializer) repairPublicProjections(ctx context.Context, stores stora return nil } if err := queueSummaries.Update(ctx, desired, current.Version, desired.Version); err != nil { - if errors.Is(err, storage.ErrVersionMismatch) { + if errors.Is(err, basestorage.ErrVersionMismatch) { continue } return fmt.Errorf("failed to update queue summary request_id=%s: %w", desired.RequestID, err) @@ -134,7 +135,7 @@ func (m *Materializer) createURIMappings(ctx context.Context, stores storage.Sto ReceivedAtMs: summary.ReceivedAtMs, RequestID: summary.RequestID, } - if err := uris.Create(ctx, mapping); err != nil && !errors.Is(err, storage.ErrAlreadyExists) { + if err := uris.Create(ctx, mapping); err != nil && !errors.Is(err, basestorage.ErrAlreadyExists) { return fmt.Errorf("failed to create request URI mapping request_id=%s change_uri=%s: %w", summary.RequestID, changeURI, err) } } diff --git a/submitqueue/core/request/materializer_test.go b/submitqueue/core/request/materializer_test.go index ad51a4107..bf37bdc49 100644 --- a/submitqueue/core/request/materializer_test.go +++ b/submitqueue/core/request/materializer_test.go @@ -24,6 +24,7 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" "go.uber.org/mock/gomock" ) @@ -316,12 +317,12 @@ func materializerStores(ctrl *gomock.Controller) (*Materializer, *storagemock.Mo queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) uriStore := storagemock.NewMockRequestURIStore(ctrl) logStore := storagemock.NewMockRequestLogStore(ctrl) - queueScoped := storagemock.NewMockStorage(ctrl) + queueScoped := gwstoragemock.NewMockStorage(ctrl) queueScoped.EXPECT().GetRequestQueueSummaryStore().Return(queueStore).AnyTimes() queueScoped.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() queueScoped.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() queueScoped.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() - factory := storagemock.NewMockFactory(ctrl) + factory := gwstoragemock.NewMockFactory(ctrl) factory.EXPECT().For(gomock.Any()).Return(queueScoped, nil).AnyTimes() return NewMaterializer(factory), summaryStore, queueStore, uriStore, logStore } diff --git a/submitqueue/core/request/request.go b/submitqueue/core/request/request.go index 318f2f03f..bb0baeb1e 100644 --- a/submitqueue/core/request/request.go +++ b/submitqueue/core/request/request.go @@ -19,7 +19,7 @@ import ( "fmt" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) // CurrentState holds the current request status obtained from the request log. It is eventually consistent with the request status in the request store. It might take some time to converge, typically no more than a few seconds. diff --git a/submitqueue/extension/storage/BUILD.bazel b/submitqueue/extension/storage/BUILD.bazel index f7892d550..1e879b6be 100644 --- a/submitqueue/extension/storage/BUILD.bazel +++ b/submitqueue/extension/storage/BUILD.bazel @@ -35,3 +35,16 @@ go_test( "@com_github_stretchr_testify//assert:go_default_library", ], ) + +# The whole SubmitQueue storage schema. Each service owns the tables it reads, +# so a deployment that separates them depends on the service package it runs; +# one that colocates them — the end-to-end tests, the local compose stack — +# takes this and does not have to track which services exist. +filegroup( + name = "schema", + srcs = [ + "//submitqueue/extension/storage/mysql/schema", + "//submitqueue/gateway/extension/storage/mysql/schema", + ], + visibility = ["//visibility:public"], +) diff --git a/submitqueue/extension/storage/mock/storage_mock.go b/submitqueue/extension/storage/mock/storage_mock.go index e5ef7379a..7831d4fba 100644 --- a/submitqueue/extension/storage/mock/storage_mock.go +++ b/submitqueue/extension/storage/mock/storage_mock.go @@ -177,34 +177,6 @@ func (mr *MockStorageMockRecorder) GetRequestBatchStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestBatchStore", reflect.TypeOf((*MockStorage)(nil).GetRequestBatchStore)) } -// GetRequestLogStore mocks base method. -func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetRequestLogStore") - ret0, _ := ret[0].(storage.RequestLogStore) - return ret0 -} - -// GetRequestLogStore indicates an expected call of GetRequestLogStore. -func (mr *MockStorageMockRecorder) GetRequestLogStore() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestLogStore", reflect.TypeOf((*MockStorage)(nil).GetRequestLogStore)) -} - -// GetRequestQueueSummaryStore mocks base method. -func (m *MockStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetRequestQueueSummaryStore") - ret0, _ := ret[0].(storage.RequestQueueSummaryStore) - return ret0 -} - -// GetRequestQueueSummaryStore indicates an expected call of GetRequestQueueSummaryStore. -func (mr *MockStorageMockRecorder) GetRequestQueueSummaryStore() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestQueueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestQueueSummaryStore)) -} - // GetRequestStore mocks base method. func (m *MockStorage) GetRequestStore() storage.RequestStore { m.ctrl.T.Helper() @@ -219,34 +191,6 @@ func (mr *MockStorageMockRecorder) GetRequestStore() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestStore", reflect.TypeOf((*MockStorage)(nil).GetRequestStore)) } -// GetRequestSummaryStore mocks base method. -func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetRequestSummaryStore") - ret0, _ := ret[0].(storage.RequestSummaryStore) - return ret0 -} - -// GetRequestSummaryStore indicates an expected call of GetRequestSummaryStore. -func (mr *MockStorageMockRecorder) GetRequestSummaryStore() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestSummaryStore)) -} - -// GetRequestURIStore mocks base method. -func (m *MockStorage) GetRequestURIStore() storage.RequestURIStore { - m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "GetRequestURIStore") - ret0, _ := ret[0].(storage.RequestURIStore) - return ret0 -} - -// GetRequestURIStore indicates an expected call of GetRequestURIStore. -func (mr *MockStorageMockRecorder) GetRequestURIStore() *gomock.Call { - mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestURIStore", reflect.TypeOf((*MockStorage)(nil).GetRequestURIStore)) -} - // GetSpeculationPathSetStore mocks base method. func (m *MockStorage) GetSpeculationPathSetStore() storage.SpeculationPathSetStore { m.ctrl.T.Helper() diff --git a/submitqueue/extension/storage/mysql/BUILD.bazel b/submitqueue/extension/storage/mysql/BUILD.bazel index bc7d829c7..dd06b0ee6 100644 --- a/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/submitqueue/extension/storage/mysql/BUILD.bazel @@ -10,11 +10,7 @@ go_library( "path_build_store.go", "queue_batch_state_store.go", "request_batch_store.go", - "request_log_store.go", - "request_queue_summary_store.go", "request_store.go", - "request_summary_store.go", - "request_uri_store.go", "speculation_path_set_store.go", "storage.go", ], @@ -38,11 +34,7 @@ go_test( "change_store_test.go", "queue_batch_state_store_test.go", "request_batch_store_test.go", - "request_log_store_test.go", - "request_queue_summary_store_test.go", "request_store_test.go", - "request_summary_store_test.go", - "request_uri_store_test.go", "speculation_path_set_store_test.go", "storage_test.go", ], diff --git a/submitqueue/extension/storage/mysql/storage.go b/submitqueue/extension/storage/mysql/storage.go index 30c179a1f..8c59717d2 100644 --- a/submitqueue/extension/storage/mysql/storage.go +++ b/submitqueue/extension/storage/mysql/storage.go @@ -59,10 +59,6 @@ func (s *Storage) For(queueName string) (storage.Storage, error) { buildStore: NewBuildStore(s.db, s.scope.SubScope("build_store"), queueName), speculationPathSetStore: NewSpeculationPathSetStore(s.db, s.scope.SubScope("speculation_path_set_store"), queueName), pathBuildStore: NewPathBuildStore(s.db, s.scope.SubScope("path_build_store"), queueName), - requestQueueStore: NewRequestQueueSummaryStore(s.db, s.scope.SubScope("request_queue_summary_store"), queueName), - requestSummaryStore: NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName), - requestLogStore: NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName), - requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName), }, nil } @@ -82,10 +78,6 @@ type boundStorage struct { buildStore storage.BuildStore speculationPathSetStore storage.SpeculationPathSetStore pathBuildStore storage.PathBuildStore - requestQueueStore storage.RequestQueueSummaryStore - requestSummaryStore storage.RequestSummaryStore - requestLogStore storage.RequestLogStore - requestURIStore storage.RequestURIStore } // Verify boundStorage implements the queue-scoped aggregate at compile time. @@ -135,23 +127,3 @@ func (f *boundStorage) GetSpeculationPathSetStore() storage.SpeculationPathSetSt func (f *boundStorage) GetPathBuildStore() storage.PathBuildStore { return f.pathBuildStore } - -// GetRequestQueueSummaryStore returns the bound MySQL-backed RequestQueueSummaryStore. -func (f *boundStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore { - return f.requestQueueStore -} - -// GetRequestSummaryStore returns the bound MySQL-backed RequestSummaryStore. -func (f *boundStorage) GetRequestSummaryStore() storage.RequestSummaryStore { - return f.requestSummaryStore -} - -// GetRequestLogStore returns the bound MySQL-backed RequestLogStore. -func (f *boundStorage) GetRequestLogStore() storage.RequestLogStore { - return f.requestLogStore -} - -// GetRequestURIStore returns the bound MySQL-backed RequestURIStore. -func (f *boundStorage) GetRequestURIStore() storage.RequestURIStore { - return f.requestURIStore -} diff --git a/submitqueue/extension/storage/mysql/storage_test.go b/submitqueue/extension/storage/mysql/storage_test.go index 75d772913..8b74b3e31 100644 --- a/submitqueue/extension/storage/mysql/storage_test.go +++ b/submitqueue/extension/storage/mysql/storage_test.go @@ -46,10 +46,6 @@ func TestNewStorage(t *testing.T) { assert.NotNil(t, bound.GetQueueBatchStateStore()) assert.NotNil(t, bound.GetBuildStore()) assert.NotNil(t, bound.GetSpeculationPathSetStore()) - assert.NotNil(t, bound.GetRequestQueueSummaryStore()) - assert.NotNil(t, bound.GetRequestSummaryStore()) - assert.NotNil(t, bound.GetRequestLogStore()) - assert.NotNil(t, bound.GetRequestURIStore()) _, err = s.For("") assert.Error(t, err, "resolving an empty queue name must fail") diff --git a/submitqueue/extension/storage/storage.go b/submitqueue/extension/storage/storage.go index ca3253f8e..5fd92be6f 100644 --- a/submitqueue/extension/storage/storage.go +++ b/submitqueue/extension/storage/storage.go @@ -94,16 +94,4 @@ type Storage interface { // GetPathBuildStore returns the PathBuildStore instance. GetPathBuildStore() PathBuildStore - - // GetRequestQueueSummaryStore returns the RequestQueueSummaryStore instance. - GetRequestQueueSummaryStore() RequestQueueSummaryStore - - // GetRequestSummaryStore returns the RequestSummaryStore instance. - GetRequestSummaryStore() RequestSummaryStore - - // GetRequestLogStore returns the RequestLogStore instance. - GetRequestLogStore() RequestLogStore - - // GetRequestURIStore returns the RequestURIStore instance. - GetRequestURIStore() RequestURIStore } diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index 0faaa4164..264249365 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -26,6 +26,7 @@ go_library( "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", "//submitqueue/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", ], @@ -61,6 +62,8 @@ go_test( "//submitqueue/extension/queueconfig/mock:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "//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", diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index 678953502..6aedad845 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/submitqueue/gateway/controller/cancel.go @@ -27,7 +27,8 @@ import ( requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" ) @@ -97,7 +98,7 @@ func (c *cancelController) Cancel(ctx context.Context, req entity.CancelRequest) return fmt.Errorf("failed to resolve storage for queue %q: %w", req.Queue, err) } if _, err := stores.GetRequestSummaryStore().Get(ctx, req.ID); err != nil { - if storage.IsNotFound(err) { + if basestorage.IsNotFound(err) { metrics.NamedCounter(c.metricsScope, opName, "not_found", 1) return errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) } diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 63d81a003..61494ecee 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -31,7 +31,7 @@ import ( "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" ) diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index a0066dead..d4c907180 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -37,8 +37,9 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap" ) @@ -342,7 +343,7 @@ func TestLand_PublishesToQueue(t *testing.T) { cnt := countermock.NewMockCounter(ctrl) cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(123), nil) - store := storagemock.NewMockStorage(ctrl) + store := gwstoragemock.NewMockStorage(ctrl) summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) uriStore := storagemock.NewMockRequestURIStore(ctrl) queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) @@ -351,7 +352,7 @@ func TestLand_PublishesToQueue(t *testing.T) { store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() - factory := storagemock.NewMockFactory(ctrl) + factory := gwstoragemock.NewMockFactory(ctrl) factory.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes() materializer := requestcore.NewMaterializer(factory) @@ -390,7 +391,7 @@ func TestLand_PublishesToQueue(t *testing.T) { ), queueStore.EXPECT().Get(gomock.Any(), gomock.Any(), "test-queue/123").DoAndReturn( func(context.Context, int64, string) (entity.RequestQueueSummary, error) { - return entity.RequestQueueSummary{}, storage.ErrNotFound + return entity.RequestQueueSummary{}, basestorage.ErrNotFound }, ), uriStore.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn( diff --git a/submitqueue/gateway/controller/list.go b/submitqueue/gateway/controller/list.go index ebad32dcd..8985d0f24 100644 --- a/submitqueue/gateway/controller/list.go +++ b/submitqueue/gateway/controller/list.go @@ -27,7 +27,8 @@ import ( "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" ) @@ -98,7 +99,7 @@ func (c *listController) List(ctx context.Context, req entity.ListRequest) (resu return entity.ListResult{}, fmt.Errorf("failed to resolve storage for queue %q: %w", req.Queue, err) } - query := storage.RequestQueueSummaryQuery{ + query := basestorage.RequestQueueSummaryQuery{ ReceivedAtOrAfterMs: req.ReceivedAtOrAfterMs, ReceivedBeforeMs: req.ReceivedBeforeMs, Limit: pageSize + 1, @@ -112,7 +113,7 @@ func (c *listController) List(ctx context.Context, req entity.ListRequest) (resu return entity.ListResult{}, fmt.Errorf("page token does not match query: %w", ErrInvalidRequest) } query.HasCursor = true - query.Cursor = storage.RequestQueueSummaryCursor{ReceivedAtMs: token.LastReceivedAtMs, RequestID: token.LastRequestID} + query.Cursor = basestorage.RequestQueueSummaryCursor{ReceivedAtMs: token.LastReceivedAtMs, RequestID: token.LastRequestID} } summaries, err := store.GetRequestQueueSummaryStore().List(ctx, query) diff --git a/submitqueue/gateway/controller/list_test.go b/submitqueue/gateway/controller/list_test.go index 9f83042d8..5fd31d949 100644 --- a/submitqueue/gateway/controller/list_test.go +++ b/submitqueue/gateway/controller/list_test.go @@ -26,18 +26,20 @@ import ( "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap" ) // listFactoryFor wraps a queue-summary store in a storage.Factory that // resolves every queue to an aggregate exposing it. -func listFactoryFor(ctrl *gomock.Controller, store storage.RequestQueueSummaryStore) storage.Factory { - agg := storagemock.NewMockStorage(ctrl) +func listFactoryFor(ctrl *gomock.Controller, store basestorage.RequestQueueSummaryStore) storage.Factory { + agg := gwstoragemock.NewMockStorage(ctrl) agg.EXPECT().GetRequestQueueSummaryStore().Return(store).AnyTimes() - f := storagemock.NewMockFactory(ctrl) + f := gwstoragemock.NewMockFactory(ctrl) f.EXPECT().For(gomock.Any()).Return(agg, nil).AnyTimes() return f } @@ -45,7 +47,7 @@ func listFactoryFor(ctrl *gomock.Controller, store storage.RequestQueueSummarySt func TestList_ReturnsPageAndCursor(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockRequestQueueSummaryStore(ctrl) - store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{ + store.EXPECT().List(gomock.Any(), basestorage.RequestQueueSummaryQuery{ ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, Limit: 3, }).Return([]entity.RequestQueueSummary{ {RequestID: "q/3", Queue: "q", ChangeURIs: []string{}, ReceivedAtMs: 190, Status: entity.RequestStatusAccepted, Metadata: map[string]string{}}, @@ -71,9 +73,9 @@ func TestList_UsesCursor(t *testing.T) { ctrl := gomock.NewController(t) store := storagemock.NewMockRequestQueueSummaryStore(ctrl) token := encodeListPageToken(listPageToken{Queue: "q", ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, LastReceivedAtMs: 180, LastRequestID: "q/2"}) - store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{ + store.EXPECT().List(gomock.Any(), basestorage.RequestQueueSummaryQuery{ ReceivedAtOrAfterMs: 100, ReceivedBeforeMs: 200, Limit: 51, - HasCursor: true, Cursor: storage.RequestQueueSummaryCursor{ReceivedAtMs: 180, RequestID: "q/2"}, + HasCursor: true, Cursor: basestorage.RequestQueueSummaryCursor{ReceivedAtMs: 180, RequestID: "q/2"}, }).Return([]entity.RequestQueueSummary{}, nil) controller := newConfiguredListController(ctrl, store) @@ -109,7 +111,7 @@ func TestList_Errors(t *testing.T) { name: "store failure", request: entity.ListRequest{Queue: "q", ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2}, setup: func(store *storagemock.MockRequestQueueSummaryStore) { - store.EXPECT().List(gomock.Any(), storage.RequestQueueSummaryQuery{ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, Limit: 51}).Return(nil, backendErr) + store.EXPECT().List(gomock.Any(), basestorage.RequestQueueSummaryQuery{ReceivedAtOrAfterMs: 1, ReceivedBeforeMs: 2, Limit: 51}).Return(nil, backendErr) }, }, } @@ -140,7 +142,7 @@ func TestList_Errors(t *testing.T) { } } -func newConfiguredListController(ctrl *gomock.Controller, store storage.RequestQueueSummaryStore) ListController { +func newConfiguredListController(ctrl *gomock.Controller, store basestorage.RequestQueueSummaryStore) ListController { queueConfigs := qcmock.NewMockStore(ctrl) queueConfigs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) return NewListController(zap.NewNop().Sugar(), tally.NoopScope, listFactoryFor(ctrl, store), queueConfigs) diff --git a/submitqueue/gateway/controller/log/BUILD.bazel b/submitqueue/gateway/controller/log/BUILD.bazel index cec2124a0..b3e8d6384 100644 --- a/submitqueue/gateway/controller/log/BUILD.bazel +++ b/submitqueue/gateway/controller/log/BUILD.bazel @@ -28,6 +28,7 @@ go_test( "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", + "//submitqueue/gateway/extension/storage/mock: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", diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go index ec00912c1..38e353b0b 100644 --- a/submitqueue/gateway/controller/log/log_test.go +++ b/submitqueue/gateway/controller/log/log_test.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap/zaptest" ) @@ -35,7 +36,7 @@ import ( // newUnusedMaterializer returns a materializer whose stores expect no calls, // for cases that fail before any persistence. func newUnusedMaterializer(ctrl *gomock.Controller) *requestcore.Materializer { - return requestcore.NewMaterializer(storagemock.NewMockFactory(ctrl)) + return requestcore.NewMaterializer(gwstoragemock.NewMockFactory(ctrl)) } func TestController_Process(t *testing.T) { @@ -136,7 +137,7 @@ func TestController_Process_RejectsTenantPayloadQueueMismatch(t *testing.T) { } func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr, queueErr error) *requestcore.Materializer { - store := storagemock.NewMockStorage(ctrl) + store := gwstoragemock.NewMockStorage(ctrl) logStore := storagemock.NewMockRequestLogStore(ctrl) summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) queueStore := storagemock.NewMockRequestQueueSummaryStore(ctrl) @@ -145,7 +146,7 @@ func newLogControllerStore(ctrl *gomock.Controller, insertErr, getErr, updateErr store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() store.EXPECT().GetRequestURIStore().Return(uriStore).AnyTimes() - factory := storagemock.NewMockFactory(ctrl) + factory := gwstoragemock.NewMockFactory(ctrl) factory.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes() materializer := requestcore.NewMaterializer(factory) logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(insertErr) diff --git a/submitqueue/gateway/controller/request_history.go b/submitqueue/gateway/controller/request_history.go index 9a0151d15..26bb6d47b 100644 --- a/submitqueue/gateway/controller/request_history.go +++ b/submitqueue/gateway/controller/request_history.go @@ -25,7 +25,8 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" ) @@ -71,7 +72,7 @@ func (c *requestHistoryController) GetRequestHistoryByID(ctx context.Context, re logs, err = stores.GetRequestLogStore().List(ctx, req.ID) if err != nil { - if storage.IsNotFound(err) { + if basestorage.IsNotFound(err) { return nil, errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) } return nil, fmt.Errorf("GetRequestHistoryByID failed to list request logs sqid=%s: %w", req.ID, err) @@ -117,7 +118,7 @@ func (c *requestHistoryController) GetRequestHistoryByChangeURI(ctx context.Cont for _, mapping := range mappings { logs, err := logStore.List(ctx, mapping.RequestID) if err != nil { - if storage.IsNotFound(err) { + if basestorage.IsNotFound(err) { continue } return nil, fmt.Errorf("GetRequestHistoryByChangeURI failed to list request logs change_uri=%s sqid=%s: %w", req.ChangeURI, mapping.RequestID, err) diff --git a/submitqueue/gateway/controller/request_history_test.go b/submitqueue/gateway/controller/request_history_test.go index f6b6e38f7..04f4c95d9 100644 --- a/submitqueue/gateway/controller/request_history_test.go +++ b/submitqueue/gateway/controller/request_history_test.go @@ -24,7 +24,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap" @@ -64,7 +64,7 @@ func TestGetRequestHistoryByChangeURI(t *testing.T) { }, nil) logStore.EXPECT().List(gomock.Any(), "queue/10").Return([]entity.RequestLog{{RequestID: "queue/10", TimestampMs: 10, Status: entity.RequestStatusLanded}}, nil) logStore.EXPECT().List(gomock.Any(), "b/2").Return([]entity.RequestLog{{RequestID: "b/2", TimestampMs: 2, Status: entity.RequestStatusStarted}}, nil) - logStore.EXPECT().List(gomock.Any(), "missing/3").Return(nil, storage.ErrNotFound) + logStore.EXPECT().List(gomock.Any(), "missing/3").Return(nil, basestorage.ErrNotFound) logStore.EXPECT().List(gomock.Any(), "queue/1").Return([]entity.RequestLog{{RequestID: "queue/1", TimestampMs: 1, Status: entity.RequestStatusAccepted}}, nil) logStore.EXPECT().List(gomock.Any(), "a/2").Return([]entity.RequestLog{{RequestID: "a/2", TimestampMs: 2, Status: entity.RequestStatusError}}, nil) @@ -114,7 +114,7 @@ func TestHistoryErrors(t *testing.T) { { name: "sqid not found", setup: func(logStore *storagemock.MockRequestLogStore, _ *storagemock.MockRequestURIStore) { - logStore.EXPECT().List(gomock.Any(), "missing/1").Return(nil, storage.ErrNotFound) + logStore.EXPECT().List(gomock.Any(), "missing/1").Return(nil, basestorage.ErrNotFound) }, call: func(c RequestHistoryController) error { _, err := c.GetRequestHistoryByID(context.Background(), entity.GetRequestHistoryByIDRequest{ID: "missing/1", Queue: "missing"}) @@ -161,7 +161,7 @@ func TestHistoryErrors(t *testing.T) { name: "all mapped logs absent", setup: func(logStore *storagemock.MockRequestLogStore, uriStore *storagemock.MockRequestURIStore) { uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{{RequestID: "queue/1"}}, nil) - logStore.EXPECT().List(gomock.Any(), "queue/1").Return(nil, storage.ErrNotFound) + logStore.EXPECT().List(gomock.Any(), "queue/1").Return(nil, basestorage.ErrNotFound) }, call: func(c RequestHistoryController) error { _, err := c.GetRequestHistoryByChangeURI(context.Background(), entity.GetRequestHistoryByChangeURIRequest{ChangeURI: "uri", Queue: "queue"}) diff --git a/submitqueue/gateway/controller/request_summary.go b/submitqueue/gateway/controller/request_summary.go index aa17d8359..2d63f41d9 100644 --- a/submitqueue/gateway/controller/request_summary.go +++ b/submitqueue/gateway/controller/request_summary.go @@ -22,7 +22,8 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" ) @@ -68,7 +69,7 @@ func (c *requestSummaryController) GetRequestSummaryByID(ctx context.Context, re summary, err = stores.GetRequestSummaryStore().Get(ctx, req.ID) if err != nil { - if storage.IsNotFound(err) { + if basestorage.IsNotFound(err) { return entity.RequestSummary{}, errs.NewUserError(&RequestNotFoundError{Sqid: req.ID}) } return entity.RequestSummary{}, fmt.Errorf("GetRequestSummaryByID failed to get request summary sqid=%s: %w", req.ID, err) @@ -117,7 +118,7 @@ func (c *requestSummaryController) GetRequestSummaryByChangeURI(ctx context.Cont for _, mapping := range mappings { summary, err := summaryStore.Get(ctx, mapping.RequestID) if err != nil { - if storage.IsNotFound(err) { + if basestorage.IsNotFound(err) { return nil, &InternalConsistencyError{Message: fmt.Sprintf("request summary missing for mapped change URI %q and sqid %q", req.ChangeURI, mapping.RequestID)} } return nil, fmt.Errorf("GetRequestSummaryByChangeURI failed to get request summary change_uri=%s sqid=%s: %w", req.ChangeURI, mapping.RequestID, err) diff --git a/submitqueue/gateway/controller/request_summary_test.go b/submitqueue/gateway/controller/request_summary_test.go index 7ee5f555c..d96355968 100644 --- a/submitqueue/gateway/controller/request_summary_test.go +++ b/submitqueue/gateway/controller/request_summary_test.go @@ -24,7 +24,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" "go.uber.org/mock/gomock" "go.uber.org/zap" @@ -109,7 +109,7 @@ func TestStatusErrors(t *testing.T) { { name: "sqid not found", setup: func(summaryStore *storagemock.MockRequestSummaryStore, _ *storagemock.MockRequestURIStore) { - summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, storage.ErrNotFound) + summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, basestorage.ErrNotFound) }, call: func(c RequestSummaryController) error { _, err := c.GetRequestSummaryByID(context.Background(), entity.GetRequestSummaryByIDRequest{ID: "missing/1", Queue: "missing"}) @@ -171,7 +171,7 @@ func TestStatusErrors(t *testing.T) { name: "mapped summary missing", setup: func(summaryStore *storagemock.MockRequestSummaryStore, uriStore *storagemock.MockRequestURIStore) { uriStore.EXPECT().ListByURI(gomock.Any(), "uri", 101).Return([]entity.RequestURI{{RequestID: "missing/1"}}, nil) - summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, storage.ErrNotFound) + summaryStore.EXPECT().Get(gomock.Any(), "missing/1").Return(entity.RequestSummary{}, basestorage.ErrNotFound) }, call: func(c RequestSummaryController) error { _, err := c.GetRequestSummaryByChangeURI(context.Background(), entity.GetRequestSummaryByChangeURIRequest{ChangeURI: "uri", Queue: "queue"}) diff --git a/submitqueue/gateway/controller/storage_fixture_test.go b/submitqueue/gateway/controller/storage_fixture_test.go index ccfa32213..ca71971e3 100644 --- a/submitqueue/gateway/controller/storage_fixture_test.go +++ b/submitqueue/gateway/controller/storage_fixture_test.go @@ -21,14 +21,16 @@ import ( requestcore "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" "go.uber.org/mock/gomock" ) // controllerStorageFixture provides shared stateful storage behavior for gateway controller tests. type controllerStorageFixture struct { - storage *storagemock.MockStorage + storage *gwstoragemock.MockStorage summaryStore *storagemock.MockRequestSummaryStore queueStore *storagemock.MockRequestQueueSummaryStore uriStore *storagemock.MockRequestURIStore @@ -42,7 +44,7 @@ type controllerStorageFixture struct { func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixture { fixture := &controllerStorageFixture{ - storage: storagemock.NewMockStorage(ctrl), + storage: gwstoragemock.NewMockStorage(ctrl), summaryStore: storagemock.NewMockRequestSummaryStore(ctrl), queueStore: storagemock.NewMockRequestQueueSummaryStore(ctrl), uriStore: storagemock.NewMockRequestURIStore(ctrl), @@ -59,7 +61,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt fixture.mu.Lock() defer fixture.mu.Unlock() if _, ok := fixture.summaries[summary.RequestID]; ok { - return storage.ErrAlreadyExists + return basestorage.ErrAlreadyExists } fixture.summaries[summary.RequestID] = summary return nil @@ -69,7 +71,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt defer fixture.mu.Unlock() summary, ok := fixture.summaries[requestID] if !ok { - return entity.RequestSummary{}, storage.ErrNotFound + return entity.RequestSummary{}, basestorage.ErrNotFound } return summary, nil }).AnyTimes() @@ -78,10 +80,10 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt defer fixture.mu.Unlock() current, ok := fixture.summaries[summary.RequestID] if !ok { - return storage.ErrNotFound + return basestorage.ErrNotFound } if current.Version != oldVersion { - return storage.ErrVersionMismatch + return basestorage.ErrVersionMismatch } summary.Version = newVersion fixture.summaries[summary.RequestID] = summary @@ -93,7 +95,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt defer fixture.mu.Unlock() key := queueSummaryTestKey(summary.Queue, summary.ReceivedAtMs, summary.RequestID) if _, ok := fixture.queueSummaries[key]; ok { - return storage.ErrAlreadyExists + return basestorage.ErrAlreadyExists } fixture.queueSummaries[key] = summary return nil @@ -106,7 +108,7 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt return summary, nil } } - return entity.RequestQueueSummary{}, storage.ErrNotFound + return entity.RequestQueueSummary{}, basestorage.ErrNotFound }).AnyTimes() fixture.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, summary entity.RequestQueueSummary, oldVersion, newVersion int32) error { fixture.mu.Lock() @@ -114,10 +116,10 @@ func newControllerStorageFixture(ctrl *gomock.Controller) *controllerStorageFixt key := queueSummaryTestKey(summary.Queue, summary.ReceivedAtMs, summary.RequestID) current, ok := fixture.queueSummaries[key] if !ok { - return storage.ErrNotFound + return basestorage.ErrNotFound } if current.Version != oldVersion { - return storage.ErrVersionMismatch + return basestorage.ErrVersionMismatch } summary.Version = newVersion fixture.queueSummaries[key] = summary @@ -164,16 +166,16 @@ func (f *controllerStorageFixture) newFactory(ctrl *gomock.Controller) storage.F } // factoryForStorage returns a storage.Factory resolving every queue to store. -func factoryForStorage(ctrl *gomock.Controller, store *storagemock.MockStorage) storage.Factory { - factory := storagemock.NewMockFactory(ctrl) +func factoryForStorage(ctrl *gomock.Controller, store *gwstoragemock.MockStorage) storage.Factory { + factory := gwstoragemock.NewMockFactory(ctrl) factory.EXPECT().For(gomock.Any()).Return(store, nil).AnyTimes() return factory } // storageWithSummaryStore returns a queue-scoped aggregate whose summary store is // the given mock; the other read-model stores accept any call and do nothing. -func storageWithSummaryStore(ctrl *gomock.Controller, summaries *storagemock.MockRequestSummaryStore) *storagemock.MockStorage { - store := storagemock.NewMockStorage(ctrl) +func storageWithSummaryStore(ctrl *gomock.Controller, summaries *storagemock.MockRequestSummaryStore) *gwstoragemock.MockStorage { + store := gwstoragemock.NewMockStorage(ctrl) store.EXPECT().GetRequestSummaryStore().Return(summaries).AnyTimes() return store } @@ -187,7 +189,7 @@ func readModelFactory( logs *storagemock.MockRequestLogStore, uris *storagemock.MockRequestURIStore, ) storage.Factory { - store := storagemock.NewMockStorage(ctrl) + store := gwstoragemock.NewMockStorage(ctrl) if summaries != nil { store.EXPECT().GetRequestSummaryStore().Return(summaries).AnyTimes() } diff --git a/submitqueue/gateway/extension/storage/BUILD.bazel b/submitqueue/gateway/extension/storage/BUILD.bazel new file mode 100644 index 000000000..c3c7f9c9e --- /dev/null +++ b/submitqueue/gateway/extension/storage/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["storage.go"], + importpath = "github.com/uber/submitqueue/submitqueue/gateway/extension/storage", + visibility = ["//visibility:public"], + deps = ["//submitqueue/extension/storage:go_default_library"], +) diff --git a/submitqueue/gateway/extension/storage/mock/BUILD.bazel b/submitqueue/gateway/extension/storage/mock/BUILD.bazel new file mode 100644 index 000000000..fca5271fb --- /dev/null +++ b/submitqueue/gateway/extension/storage/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["storage_mock.go"], + importpath = "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/submitqueue/gateway/extension/storage/mock/storage_mock.go b/submitqueue/gateway/extension/storage/mock/storage_mock.go new file mode 100644 index 000000000..0f15384ab --- /dev/null +++ b/submitqueue/gateway/extension/storage/mock/storage_mock.go @@ -0,0 +1,137 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: storage.go +// +// Generated by this command: +// +// mockgen -source=storage.go -destination=mock/storage_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + reflect "reflect" + + storage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage0 "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + gomock "go.uber.org/mock/gomock" +) + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(config storage0.Config) (storage0.Storage, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", config) + ret0, _ := ret[0].(storage0.Storage) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(config any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), config) +} + +// MockStorage is a mock of Storage interface. +type MockStorage struct { + ctrl *gomock.Controller + recorder *MockStorageMockRecorder + isgomock struct{} +} + +// MockStorageMockRecorder is the mock recorder for MockStorage. +type MockStorageMockRecorder struct { + mock *MockStorage +} + +// NewMockStorage creates a new mock instance. +func NewMockStorage(ctrl *gomock.Controller) *MockStorage { + mock := &MockStorage{ctrl: ctrl} + mock.recorder = &MockStorageMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockStorage) EXPECT() *MockStorageMockRecorder { + return m.recorder +} + +// GetRequestLogStore mocks base method. +func (m *MockStorage) GetRequestLogStore() storage.RequestLogStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestLogStore") + ret0, _ := ret[0].(storage.RequestLogStore) + return ret0 +} + +// GetRequestLogStore indicates an expected call of GetRequestLogStore. +func (mr *MockStorageMockRecorder) GetRequestLogStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestLogStore", reflect.TypeOf((*MockStorage)(nil).GetRequestLogStore)) +} + +// GetRequestQueueSummaryStore mocks base method. +func (m *MockStorage) GetRequestQueueSummaryStore() storage.RequestQueueSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestQueueSummaryStore") + ret0, _ := ret[0].(storage.RequestQueueSummaryStore) + return ret0 +} + +// GetRequestQueueSummaryStore indicates an expected call of GetRequestQueueSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestQueueSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestQueueSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestQueueSummaryStore)) +} + +// GetRequestSummaryStore mocks base method. +func (m *MockStorage) GetRequestSummaryStore() storage.RequestSummaryStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestSummaryStore") + ret0, _ := ret[0].(storage.RequestSummaryStore) + return ret0 +} + +// GetRequestSummaryStore indicates an expected call of GetRequestSummaryStore. +func (mr *MockStorageMockRecorder) GetRequestSummaryStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestSummaryStore", reflect.TypeOf((*MockStorage)(nil).GetRequestSummaryStore)) +} + +// GetRequestURIStore mocks base method. +func (m *MockStorage) GetRequestURIStore() storage.RequestURIStore { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRequestURIStore") + ret0, _ := ret[0].(storage.RequestURIStore) + return ret0 +} + +// GetRequestURIStore indicates an expected call of GetRequestURIStore. +func (mr *MockStorageMockRecorder) GetRequestURIStore() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRequestURIStore", reflect.TypeOf((*MockStorage)(nil).GetRequestURIStore)) +} diff --git a/submitqueue/gateway/extension/storage/mysql/BUILD.bazel b/submitqueue/gateway/extension/storage/mysql/BUILD.bazel new file mode 100644 index 000000000..b4fba78e6 --- /dev/null +++ b/submitqueue/gateway/extension/storage/mysql/BUILD.bazel @@ -0,0 +1,43 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = [ + "request_log_store.go", + "request_queue_summary_store.go", + "request_summary_store.go", + "request_uri_store.go", + "storage.go", + ], + importpath = "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql", + 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_go_sql_driver_mysql//:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = [ + "request_log_store_test.go", + "request_queue_summary_store_test.go", + "request_summary_store_test.go", + "request_uri_store_test.go", + "storage_test.go", + ], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "@com_github_data_dog_go_sqlmock//:go_default_library", + "@com_github_go_sql_driver_mysql//: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", + ], +) diff --git a/submitqueue/extension/storage/mysql/request_log_store.go b/submitqueue/gateway/extension/storage/mysql/request_log_store.go similarity index 98% rename from submitqueue/extension/storage/mysql/request_log_store.go rename to submitqueue/gateway/extension/storage/mysql/request_log_store.go index 17739a169..d7114127a 100644 --- a/submitqueue/extension/storage/mysql/request_log_store.go +++ b/submitqueue/gateway/extension/storage/mysql/request_log_store.go @@ -25,7 +25,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) type requestLogStore struct { diff --git a/submitqueue/extension/storage/mysql/request_log_store_test.go b/submitqueue/gateway/extension/storage/mysql/request_log_store_test.go similarity index 98% rename from submitqueue/extension/storage/mysql/request_log_store_test.go rename to submitqueue/gateway/extension/storage/mysql/request_log_store_test.go index 9217e076b..8e4a04a4c 100644 --- a/submitqueue/extension/storage/mysql/request_log_store_test.go +++ b/submitqueue/gateway/extension/storage/mysql/request_log_store_test.go @@ -25,7 +25,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) // testLogQueue is the queue every request-log store in this file is bound to. diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store.go b/submitqueue/gateway/extension/storage/mysql/request_queue_summary_store.go similarity index 99% rename from submitqueue/extension/storage/mysql/request_queue_summary_store.go rename to submitqueue/gateway/extension/storage/mysql/request_queue_summary_store.go index f00c49d6a..c381bd8fe 100644 --- a/submitqueue/extension/storage/mysql/request_queue_summary_store.go +++ b/submitqueue/gateway/extension/storage/mysql/request_queue_summary_store.go @@ -25,7 +25,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) type requestQueueSummaryStore struct { diff --git a/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go b/submitqueue/gateway/extension/storage/mysql/request_queue_summary_store_test.go similarity index 99% rename from submitqueue/extension/storage/mysql/request_queue_summary_store_test.go rename to submitqueue/gateway/extension/storage/mysql/request_queue_summary_store_test.go index 8627748fc..56e5e522c 100644 --- a/submitqueue/extension/storage/mysql/request_queue_summary_store_test.go +++ b/submitqueue/gateway/extension/storage/mysql/request_queue_summary_store_test.go @@ -26,7 +26,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) func setupRequestQueueSummaryStoreTest(t *testing.T) (*sql.DB, sqlmock.Sqlmock, storage.RequestQueueSummaryStore) { diff --git a/submitqueue/extension/storage/mysql/request_summary_store.go b/submitqueue/gateway/extension/storage/mysql/request_summary_store.go similarity index 98% rename from submitqueue/extension/storage/mysql/request_summary_store.go rename to submitqueue/gateway/extension/storage/mysql/request_summary_store.go index a84b84546..53f57073d 100644 --- a/submitqueue/extension/storage/mysql/request_summary_store.go +++ b/submitqueue/gateway/extension/storage/mysql/request_summary_store.go @@ -26,7 +26,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) type requestSummaryStore struct { diff --git a/submitqueue/extension/storage/mysql/request_summary_store_test.go b/submitqueue/gateway/extension/storage/mysql/request_summary_store_test.go similarity index 99% rename from submitqueue/extension/storage/mysql/request_summary_store_test.go rename to submitqueue/gateway/extension/storage/mysql/request_summary_store_test.go index fea7dad1d..689d6baaf 100644 --- a/submitqueue/extension/storage/mysql/request_summary_store_test.go +++ b/submitqueue/gateway/extension/storage/mysql/request_summary_store_test.go @@ -26,7 +26,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) // testSummaryQueue is the queue every request-summary store in this file is bound to. diff --git a/submitqueue/extension/storage/mysql/request_uri_store.go b/submitqueue/gateway/extension/storage/mysql/request_uri_store.go similarity index 98% rename from submitqueue/extension/storage/mysql/request_uri_store.go rename to submitqueue/gateway/extension/storage/mysql/request_uri_store.go index 220d6c736..704f8b876 100644 --- a/submitqueue/extension/storage/mysql/request_uri_store.go +++ b/submitqueue/gateway/extension/storage/mysql/request_uri_store.go @@ -25,7 +25,7 @@ import ( "github.com/uber/submitqueue/platform/metrics" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) type requestURIStore struct { diff --git a/submitqueue/extension/storage/mysql/request_uri_store_test.go b/submitqueue/gateway/extension/storage/mysql/request_uri_store_test.go similarity index 98% rename from submitqueue/extension/storage/mysql/request_uri_store_test.go rename to submitqueue/gateway/extension/storage/mysql/request_uri_store_test.go index cc4734901..5a4ad0c37 100644 --- a/submitqueue/extension/storage/mysql/request_uri_store_test.go +++ b/submitqueue/gateway/extension/storage/mysql/request_uri_store_test.go @@ -26,7 +26,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/extension/storage" ) // testURIQueue is the queue every request-URI store in this file is bound to. diff --git a/submitqueue/gateway/extension/storage/mysql/schema/BUILD.bazel b/submitqueue/gateway/extension/storage/mysql/schema/BUILD.bazel new file mode 100644 index 000000000..3412d7733 --- /dev/null +++ b/submitqueue/gateway/extension/storage/mysql/schema/BUILD.bazel @@ -0,0 +1,5 @@ +filegroup( + name = "schema", + srcs = glob(["*.sql"]), + visibility = ["//visibility:public"], +) diff --git a/submitqueue/extension/storage/mysql/schema/change_uri_request_mapping.sql b/submitqueue/gateway/extension/storage/mysql/schema/change_uri_request_mapping.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/change_uri_request_mapping.sql rename to submitqueue/gateway/extension/storage/mysql/schema/change_uri_request_mapping.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_log.sql b/submitqueue/gateway/extension/storage/mysql/schema/request_log.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_log.sql rename to submitqueue/gateway/extension/storage/mysql/schema/request_log.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_summary.sql b/submitqueue/gateway/extension/storage/mysql/schema/request_summary.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_summary.sql rename to submitqueue/gateway/extension/storage/mysql/schema/request_summary.sql diff --git a/submitqueue/extension/storage/mysql/schema/request_summary_by_queue.sql b/submitqueue/gateway/extension/storage/mysql/schema/request_summary_by_queue.sql similarity index 100% rename from submitqueue/extension/storage/mysql/schema/request_summary_by_queue.sql rename to submitqueue/gateway/extension/storage/mysql/schema/request_summary_by_queue.sql diff --git a/submitqueue/gateway/extension/storage/mysql/storage.go b/submitqueue/gateway/extension/storage/mysql/storage.go new file mode 100644 index 000000000..897713299 --- /dev/null +++ b/submitqueue/gateway/extension/storage/mysql/storage.go @@ -0,0 +1,95 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "database/sql" + "fmt" + + _ "github.com/go-sql-driver/mysql" + "github.com/uber-go/tally" + + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" + storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" +) + +// mysqlErrDuplicateEntry is MySQL error code 1062 ("Duplicate entry"), returned on a unique or primary key violation. +// It requires a unique index on the table to be raised. +const mysqlErrDuplicateEntry = 1062 + +// Storage is the MySQL backend for the gateway's stores. It owns the shared +// connection pool and binds queue-scoped store aggregates over the shared +// tables on demand via For. The wiring layer adapts For into the +// storage.Factory seam; per-queue backend routing stays a host decision. +type Storage struct { + db *sql.DB + scope tally.Scope +} + +// NewStorage creates a new MySQL storage backend over the given connection pool. +func NewStorage(db *sql.DB, scope tally.Scope) (*Storage, error) { + return &Storage{db: db, scope: scope}, nil +} + +// For returns the queue-scoped store aggregate bound to queueName over the +// shared pool. Every store the aggregate hands back reads and writes only that +// queue's records. +func (s *Storage) For(queueName string) (storage.Storage, error) { + if queueName == "" { + return nil, fmt.Errorf("queue name must not be empty") + } + return &boundStorage{ + requestQueueStore: NewRequestQueueSummaryStore(s.db, s.scope.SubScope("request_queue_summary_store"), queueName), + requestSummaryStore: NewRequestSummaryStore(s.db, s.scope.SubScope("request_summary_store"), queueName), + requestLogStore: NewRequestLogStore(s.db, s.scope.SubScope("request_log_store"), queueName), + requestURIStore: NewRequestURIStore(s.db, s.scope.SubScope("request_uri_store"), queueName), + }, nil +} + +// Close closes the underlying database connection. +func (s *Storage) Close() error { + return s.db.Close() +} + +// boundStorage is the queue-scoped store aggregate returned by For. +type boundStorage struct { + requestQueueStore basestorage.RequestQueueSummaryStore + requestSummaryStore basestorage.RequestSummaryStore + requestLogStore basestorage.RequestLogStore + requestURIStore basestorage.RequestURIStore +} + +// Verify boundStorage implements the queue-scoped aggregate at compile time. +var _ storage.Storage = (*boundStorage)(nil) + +// GetRequestQueueSummaryStore returns the bound MySQL-backed RequestQueueSummaryStore. +func (f *boundStorage) GetRequestQueueSummaryStore() basestorage.RequestQueueSummaryStore { + return f.requestQueueStore +} + +// GetRequestSummaryStore returns the bound MySQL-backed RequestSummaryStore. +func (f *boundStorage) GetRequestSummaryStore() basestorage.RequestSummaryStore { + return f.requestSummaryStore +} + +// GetRequestLogStore returns the bound MySQL-backed RequestLogStore. +func (f *boundStorage) GetRequestLogStore() basestorage.RequestLogStore { + return f.requestLogStore +} + +// GetRequestURIStore returns the bound MySQL-backed RequestURIStore. +func (f *boundStorage) GetRequestURIStore() basestorage.RequestURIStore { + return f.requestURIStore +} diff --git a/submitqueue/gateway/extension/storage/mysql/storage_test.go b/submitqueue/gateway/extension/storage/mysql/storage_test.go new file mode 100644 index 000000000..08b9c197d --- /dev/null +++ b/submitqueue/gateway/extension/storage/mysql/storage_test.go @@ -0,0 +1,60 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" +) + +// testMetrics returns a test metrics scope for use in tests. +func testMetrics() tally.Scope { + return tally.NoopScope +} + +func TestNewStorage(t *testing.T) { + db, _, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + + s, err := NewStorage(db, testMetrics()) + require.NoError(t, err) + + bound, err := s.For("monorepo") + require.NoError(t, err) + assert.NotNil(t, bound.GetRequestLogStore()) + assert.NotNil(t, bound.GetRequestSummaryStore()) + assert.NotNil(t, bound.GetRequestQueueSummaryStore()) + assert.NotNil(t, bound.GetRequestURIStore()) + + _, err = s.For("") + assert.Error(t, err, "resolving an empty queue name must fail") +} + +func TestMysqlStorage_Close(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + + s, err := NewStorage(db, testMetrics()) + require.NoError(t, err) + + mock.ExpectClose() + require.NoError(t, s.Close()) + require.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/submitqueue/gateway/extension/storage/storage.go b/submitqueue/gateway/extension/storage/storage.go new file mode 100644 index 000000000..760492157 --- /dev/null +++ b/submitqueue/gateway/extension/storage/storage.go @@ -0,0 +1,58 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package storage resolves the gateway's queue-scoped stores: the append-only +// request log and the three read models behind request-summary retrieval and +// List. The store contracts themselves stay in +// submitqueue/extension/storage — only the aggregate is service-scoped, so +// what the gateway can reach is narrower than what the domain defines. +package storage + +//go:generate mockgen -source=storage.go -destination=mock/storage_mock.go -package=mock + +import ( + basestorage "github.com/uber/submitqueue/submitqueue/extension/storage" +) + +// Config identifies the queue a Storage instance is resolved for. It is the +// shared storage config: an alias rather than a new type, so a caller holding +// one can resolve either service's storage with it. +type Config = basestorage.Config + +// Factory resolves the queue-scoped Storage aggregate for a queue. Mirrors the +// extension contract: the host wiring decides which backend serves which +// queue; implementations bind the queue over their backend so a resolved +// instance can only read and write that queue's data. +type Factory interface { + // For returns the Storage aggregate bound to the queue named in config. + For(config Config) (Storage, error) +} + +// Storage aggregates the gateway's queue-scoped stores into a single +// injectable dependency. An instance is resolved per queue through Factory and +// is bound to that queue: entity arguments whose Queue field disagrees with +// the binding are rejected, and reads never surface another queue's records. +type Storage interface { + // GetRequestLogStore returns the RequestLogStore instance. + GetRequestLogStore() basestorage.RequestLogStore + + // GetRequestSummaryStore returns the RequestSummaryStore instance. + GetRequestSummaryStore() basestorage.RequestSummaryStore + + // GetRequestQueueSummaryStore returns the RequestQueueSummaryStore instance. + GetRequestQueueSummaryStore() basestorage.RequestQueueSummaryStore + + // GetRequestURIStore returns the RequestURIStore instance. + GetRequestURIStore() basestorage.RequestURIStore +} diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel index 502f9f46f..b7d436f60 100644 --- a/test/e2e/submitqueue/BUILD.bazel +++ b/test/e2e/submitqueue/BUILD.bazel @@ -16,7 +16,7 @@ go_test( "//service/submitqueue/demo/provider/git:config", "//service/submitqueue/gateway/server:docker_test_context", "//service/submitqueue/orchestrator/server:docker_test_context", - "//submitqueue/extension/storage/mysql/schema", + "//submitqueue/extension/storage:schema", "@git", "@git//:git_receive_pack", "@git//:git_upload_archive", diff --git a/test/e2e/submitqueue/git_suite_test.go b/test/e2e/submitqueue/git_suite_test.go index 2cc0be56e..9d7fe5480 100644 --- a/test/e2e/submitqueue/git_suite_test.go +++ b/test/e2e/submitqueue/git_suite_test.go @@ -133,7 +133,7 @@ func (s *GitMergeSuite) SetupSuite() { s.queueDB, err = s.stack.ConnectMySQLService("mysql-queue") require.NoError(t, err) - testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/extension/storage/mysql/schema")) + testutil.ApplySubmitQueueStorageSchema(t, s.log, s.db) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) testutil.ApplySchema(t, s.log, s.queueDB, testutil.SchemaDir("platform/extension/messagequeue/mysql/schema")) diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go index 7bfe04198..18d855d97 100644 --- a/test/e2e/submitqueue/suite_test.go +++ b/test/e2e/submitqueue/suite_test.go @@ -125,7 +125,7 @@ func (s *E2EIntegrationSuite) SetupSuite() { require.NoError(t, err, "failed to connect to queue MySQL") // Apply schemas programmatically to application database - testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/extension/storage/mysql/schema")) + testutil.ApplySubmitQueueStorageSchema(t, s.log, s.db) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) // Apply schemas programmatically to queue database diff --git a/test/integration/submitqueue/extension/storage/BUILD.bazel b/test/integration/submitqueue/extension/storage/BUILD.bazel index 7585d20d3..3c6736468 100644 --- a/test/integration/submitqueue/extension/storage/BUILD.bazel +++ b/test/integration/submitqueue/extension/storage/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/base/mergestrategy:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/test/integration/submitqueue/extension/storage/mysql/BUILD.bazel b/test/integration/submitqueue/extension/storage/mysql/BUILD.bazel index 5e75fb893..b1e7942ad 100644 --- a/test/integration/submitqueue/extension/storage/mysql/BUILD.bazel +++ b/test/integration/submitqueue/extension/storage/mysql/BUILD.bazel @@ -6,6 +6,7 @@ go_test( data = [ "docker-compose.yml", "//submitqueue/extension/storage/mysql/schema", + "//submitqueue/gateway/extension/storage/mysql/schema", ], tags = [ "integration", @@ -14,6 +15,8 @@ go_test( deps = [ "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mysql:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage/mysql:go_default_library", "//test/integration/submitqueue/extension/storage:go_default_library", "//test/testutil:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", diff --git a/test/integration/submitqueue/extension/storage/mysql/storage_test.go b/test/integration/submitqueue/extension/storage/mysql/storage_test.go index dda5a6871..3c8da0d9e 100644 --- a/test/integration/submitqueue/extension/storage/mysql/storage_test.go +++ b/test/integration/submitqueue/extension/storage/mysql/storage_test.go @@ -25,6 +25,8 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/submitqueue/extension/storage" mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" + gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + gwmysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql" storagesuite "github.com/uber/submitqueue/test/integration/submitqueue/extension/storage" "github.com/uber/submitqueue/test/testutil" ) @@ -69,8 +71,8 @@ func (s *MySQLStorageIntegrationSuite) SetupSuite() { require.NoError(t, err, "failed to connect to MySQL") // Apply schemas programmatically from directory - schemaDir := testutil.SchemaDir("submitqueue/extension/storage/mysql/schema") - testutil.ApplySchema(t, s.log, s.db, schemaDir) + testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/extension/storage/mysql/schema")) + testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/gateway/extension/storage/mysql/schema")) s.log.Logf("Schemas applied successfully") @@ -78,10 +80,14 @@ func (s *MySQLStorageIntegrationSuite) SetupSuite() { store, err := mysqlstorage.NewStorage(s.db, tally.NoopScope) require.NoError(t, err, "failed to create storage") + gwStore, err := gwmysqlstorage.NewStorage(s.db, tally.NoopScope) + require.NoError(t, err, "failed to create gateway storage") + // Provide the storage backend to the contract suite through the // queue-scoped factory adapter. s.SetContext(ctx) s.SetFactory(mysqlFactory{backend: store}) + s.SetGatewayFactory(gatewayMySQLFactory{backend: gwStore}) s.SetLogger(s.log) t.Cleanup(func() { @@ -109,3 +115,14 @@ type mysqlFactory struct { func (f mysqlFactory) For(config storage.Config) (storage.Storage, error) { return f.backend.For(config.QueueName) } + +// gatewayMySQLFactory adapts the gateway's MySQL storage backend to the +// gateway storage.Factory seam for the contract suite. +type gatewayMySQLFactory struct { + backend *gwmysqlstorage.Storage +} + +// For returns the queue-scoped store aggregate bound to the queue named in config. +func (f gatewayMySQLFactory) For(config gwstorage.Config) (gwstorage.Storage, error) { + return f.backend.For(config.QueueName) +} diff --git a/test/integration/submitqueue/extension/storage/suite.go b/test/integration/submitqueue/extension/storage/suite.go index 9a3b104f7..79597c280 100644 --- a/test/integration/submitqueue/extension/storage/suite.go +++ b/test/integration/submitqueue/extension/storage/suite.go @@ -27,6 +27,7 @@ import ( "github.com/uber/submitqueue/platform/base/mergestrategy" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" + gwstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "github.com/uber/submitqueue/test/testutil" ) @@ -36,9 +37,10 @@ import ( // embed this suite and call SetFactory(). type StorageContractSuite struct { suite.Suite - ctx context.Context - factory storage.Factory - log *testutil.TestLogger + ctx context.Context + factory storage.Factory + gatewayFactory gwstorage.Factory + log *testutil.TestLogger } // SetContext sets the context for tests @@ -52,6 +54,20 @@ func (s *StorageContractSuite) SetFactory(factory storage.Factory) { s.factory = factory } +// SetGatewayFactory is called by implementation tests to provide the +// queue-scoped gateway storage factory under test. +func (s *StorageContractSuite) SetGatewayFactory(factory gwstorage.Factory) { + s.gatewayFactory = factory +} + +// forGatewayQueue resolves the gateway's queue-scoped store aggregate, failing +// the test on resolution errors. +func (s *StorageContractSuite) forGatewayQueue(queue string) gwstorage.Storage { + store, err := s.gatewayFactory.For(gwstorage.Config{QueueName: queue}) + s.Require().NoError(err) + return store +} + // forQueue resolves the queue-scoped store aggregate for a queue, failing the // test on resolution errors. func (s *StorageContractSuite) forQueue(queue string) storage.Storage { @@ -672,7 +688,7 @@ func (s *StorageContractSuite) TestStorage_RequestSummaryCreateGetAndCAS() { Status: entity.RequestStatusAccepted, RequestVersion: 1, StatusTimestampMs: 100, Version: 1, LastError: "", Metadata: nil, } - store := s.forQueue(queue).GetRequestSummaryStore() + store := s.forGatewayQueue(queue).GetRequestSummaryStore() require.NoError(t, store.Create(ctx, summary)) require.ErrorIs(t, store.Create(ctx, summary), storage.ErrAlreadyExists) @@ -753,7 +769,7 @@ func (s *StorageContractSuite) TestStorage_RequestSummaryCreateGetAndCAS() { func (s *StorageContractSuite) TestStorage_RequestQueueSummaryListAndCursor() { t := s.T() ctx := s.ctx - store := s.forQueue("queue-summary").GetRequestQueueSummaryStore() + store := s.forGatewayQueue("queue-summary").GetRequestQueueSummaryStore() rows := []entity.RequestQueueSummary{ {RequestID: "queue-summary/1", Queue: "queue-summary", ChangeURIs: nil, ReceivedAtMs: 100, Status: entity.RequestStatusAccepted, Version: 1, Metadata: nil}, {RequestID: "queue-summary/2", Queue: "queue-summary", ChangeURIs: []string{"uri/2"}, ReceivedAtMs: 200, Status: entity.RequestStatusLanded, Version: 1, Metadata: map[string]string{}}, @@ -851,7 +867,7 @@ func (s *StorageContractSuite) TestStorage_RequestURIListIsBoundedAndOrdered() { t := s.T() ctx := s.ctx const queue = "uri-q" - store := s.forQueue(queue).GetRequestURIStore() + store := s.forGatewayQueue(queue).GetRequestURIStore() rows := []entity.RequestURI{ {ChangeURI: "uri/shared", Queue: queue, ReceivedAtMs: 100, RequestID: "uri/1"}, {ChangeURI: "uri/shared", Queue: queue, ReceivedAtMs: 200, RequestID: "uri/2"}, @@ -872,7 +888,7 @@ func (s *StorageContractSuite) TestStorage_RequestURIListIsBoundedAndOrdered() { assert.Empty(t, empty) // The same change URI in another queue is a distinct mapping set. - otherStore := s.forQueue("uri-q-other").GetRequestURIStore() + otherStore := s.forGatewayQueue("uri-q-other").GetRequestURIStore() require.NoError(t, otherStore.Create(ctx, entity.RequestURI{ ChangeURI: "uri/shared", Queue: "uri-q-other", ReceivedAtMs: 100, RequestID: "other/1", }), "the same change URI in another queue is a distinct row") @@ -889,7 +905,7 @@ func (s *StorageContractSuite) TestStorage_RequestLogAppendAndList() { t := s.T() ctx := s.ctx const queue = "log-q" - store := s.forQueue(queue).GetRequestLogStore() + store := s.forGatewayQueue(queue).GetRequestLogStore() _, err := store.List(ctx, "log/missing") require.ErrorIs(t, err, storage.ErrNotFound) @@ -914,7 +930,7 @@ func (s *StorageContractSuite) TestStorage_RequestLogAppendAndList() { })) // The same request ID in another queue is an independent history. - otherStore := s.forQueue("log-q-other").GetRequestLogStore() + otherStore := s.forGatewayQueue("log-q-other").GetRequestLogStore() require.NoError(t, otherStore.Insert(ctx, entity.RequestLog{ RequestID: "log/1", Queue: "log-q-other", TimestampMs: 150, Status: entity.RequestStatusLanded, Metadata: map[string]string{}, })) diff --git a/test/integration/submitqueue/gateway/BUILD.bazel b/test/integration/submitqueue/gateway/BUILD.bazel index fe437299e..534c739e4 100644 --- a/test/integration/submitqueue/gateway/BUILD.bazel +++ b/test/integration/submitqueue/gateway/BUILD.bazel @@ -8,7 +8,7 @@ go_test( "//platform/extension/messagequeue/mysql/schema", "//service/submitqueue/gateway/server:docker-compose.yml", "//service/submitqueue/gateway/server:docker_test_context", - "//submitqueue/extension/storage/mysql/schema", + "//submitqueue/gateway/extension/storage/mysql/schema", ], tags = [ "integration", @@ -24,8 +24,8 @@ go_test( "//submitqueue/core/request:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", - "//submitqueue/extension/storage:go_default_library", - "//submitqueue/extension/storage/mysql:go_default_library", + "//submitqueue/gateway/extension/storage:go_default_library", + "//submitqueue/gateway/extension/storage/mysql:go_default_library", "//test/testutil:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/test/integration/submitqueue/gateway/suite_test.go b/test/integration/submitqueue/gateway/suite_test.go index c178d3c87..279958b65 100644 --- a/test/integration/submitqueue/gateway/suite_test.go +++ b/test/integration/submitqueue/gateway/suite_test.go @@ -44,8 +44,8 @@ import ( corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/storage" - mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" + "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" + mysqlstorage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mysql" "github.com/uber/submitqueue/test/testutil" "go.uber.org/zap" "google.golang.org/grpc" @@ -104,7 +104,7 @@ func (s *GatewayIntegrationSuite) SetupSuite() { require.NoError(t, err, "failed to connect to queue MySQL") // Apply schemas programmatically to application database - testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/extension/storage/mysql/schema")) + testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("submitqueue/gateway/extension/storage/mysql/schema")) testutil.ApplySchema(t, s.log, s.db, testutil.SchemaDir("platform/extension/counter/mysql/schema")) // Apply schemas programmatically to queue database diff --git a/test/testutil/schema.go b/test/testutil/schema.go index c736c88ad..fff9bc9bc 100644 --- a/test/testutil/schema.go +++ b/test/testutil/schema.go @@ -46,6 +46,29 @@ func SchemaDir(relativePath string) string { return Runfile(relativePath) } +// SubmitQueueStorageSchemaDirs returns every directory making up the +// SubmitQueue storage schema, in the order they must be applied. +// +// The schema is split per service: each owns the tables it reads. A deployment +// that colocates the services on one database applies all of them, and callers +// that do — the end-to-end tests, the local compose stack — take this rather +// than naming each service's directory and going stale when one is added. +func SubmitQueueStorageSchemaDirs() []string { + return []string{ + "submitqueue/gateway/extension/storage/mysql/schema", + "submitqueue/extension/storage/mysql/schema", + } +} + +// ApplySubmitQueueStorageSchema applies every directory named by +// SubmitQueueStorageSchemaDirs. +func ApplySubmitQueueStorageSchema(t *testing.T, log *TestLogger, db *sql.DB) { + t.Helper() + for _, dir := range SubmitQueueStorageSchemaDirs() { + ApplySchema(t, log, db, SchemaDir(dir)) + } +} + // ApplySchema reads all .sql files from the schema directory and executes them on the database. func ApplySchema(t *testing.T, log *TestLogger, db *sql.DB, schemaDirectory string) { t.Helper() diff --git a/tool/linter/queueshard/main.go b/tool/linter/queueshard/main.go index d900bc11d..0b880e3f7 100644 --- a/tool/linter/queueshard/main.go +++ b/tool/linter/queueshard/main.go @@ -34,15 +34,17 @@ import ( // schemaShardColumns identifies the allowed shard columns for each schema root. var schemaShardColumns = map[string]map[string]bool{ - "submitqueue/extension/storage/mysql/schema": {"queue": true, "name": true}, - "stovepipe/extension/storage/mysql/schema": {"queue": true, "name": true}, - "platform/extension/counter/mysql/schema": {"queue": true, "name": true}, - "platform/extension/messagequeue/mysql/schema": {"tenant": true}, + "submitqueue/extension/storage/mysql/schema": {"queue": true, "name": true}, + "submitqueue/gateway/extension/storage/mysql/schema": {"queue": true, "name": true}, + "stovepipe/extension/storage/mysql/schema": {"queue": true, "name": true}, + "platform/extension/counter/mysql/schema": {"queue": true, "name": true}, + "platform/extension/messagequeue/mysql/schema": {"tenant": true}, } // schemaRoots are the directories scanned for table definitions. var schemaRoots = []string{ "submitqueue/extension/storage/mysql/schema", + "submitqueue/gateway/extension/storage/mysql/schema", "stovepipe/extension/storage/mysql/schema", "platform/extension/counter/mysql/schema", "platform/extension/messagequeue/mysql/schema",