diff --git a/service/submitqueue/README.md b/service/submitqueue/README.md index 365a77947..e35933c5a 100644 --- a/service/submitqueue/README.md +++ b/service/submitqueue/README.md @@ -21,6 +21,7 @@ Each service requires the application and queue MySQL databases. Service-specifi | `MYSQL_DSN` | both | yes | Application database DSN | — | | `QUEUE_MYSQL_DSN` | both | yes | Queue database DSN | — | | `QUEUE_CONFIG_PATH` | Gateway | yes | YAML registry of accepted queue names | — | +| `QUEUE_PAUSE_CONFIG_PATH` | Gateway | no | YAML file of per-queue pause switches, re-read on every land | pausing disabled when unset | | `PROFILES_CONFIG_PATH` | Orchestrator | no | YAML extension profiles and per-queue routing | built-in example profiles | | `HOSTNAME` | both | no | Base subscriber name; Gateway appends its PID | time-based service name | | `QUEUE_LOG_LEVEL` | both | no | Message-queue logger level | `info` | diff --git a/service/submitqueue/gateway/server/BUILD.bazel b/service/submitqueue/gateway/server/BUILD.bazel index 28b4b4364..ba0697afc 100644 --- a/service/submitqueue/gateway/server/BUILD.bazel +++ b/service/submitqueue/gateway/server/BUILD.bazel @@ -32,6 +32,9 @@ go_library( "//service/submitqueue/gateway/server/mapper:go_default_library", "//submitqueue/core/topickey:go_default_library", "//submitqueue/extension/queueconfig/yaml:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "//submitqueue/extension/queuepause/file:go_default_library", + "//submitqueue/extension/queuepause/noop:go_default_library", "//submitqueue/gateway/controller:go_default_library", "//submitqueue/gateway/controller/log:go_default_library", "//submitqueue/gateway/core/request:go_default_library", @@ -86,6 +89,7 @@ go_test( embed = [":gateway_lib"], # keep deps = [ "//api/submitqueue/gateway/protopb:go_default_library", + "//platform/errs:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", "//submitqueue/gateway/controller:go_default_library", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index 88611a65b..26c4a9e6a 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -44,6 +44,9 @@ import ( "github.com/uber/submitqueue/service/submitqueue/gateway/server/mapper" "github.com/uber/submitqueue/submitqueue/core/topickey" yamlqueueconfig "github.com/uber/submitqueue/submitqueue/extension/queueconfig/yaml" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + queuepausefile "github.com/uber/submitqueue/submitqueue/extension/queuepause/file" + queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop" "github.com/uber/submitqueue/submitqueue/gateway/controller" logctrl "github.com/uber/submitqueue/submitqueue/gateway/controller/log" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" @@ -149,6 +152,8 @@ func (s *GatewayServer) GetRequestHistoryByChangeURI(ctx context.Context, req *p return &pb.GetRequestHistoryByChangeURIResponse{Histories: mapper.RequestHistoriesToProto(histories)}, nil } +// gatewayStatusError maps controller errors to status codes. IsInvalidRequest +// matches every user error, so each specific user error is checked before it. func gatewayStatusError(err error) error { switch { case controller.IsRequestNotFound(err): @@ -157,6 +162,8 @@ func gatewayStatusError(err error) error { return status.Error(codes.ResourceExhausted, err.Error()) case controller.IsInternalConsistency(err): return status.Error(codes.Internal, err.Error()) + case controller.IsQueuePaused(err): + return status.Error(codes.FailedPrecondition, err.Error()) case controller.IsInvalidRequest(err), controller.IsUnrecognizedQueue(err): return status.Error(codes.InvalidArgument, err.Error()) default: @@ -164,6 +171,22 @@ func gatewayStatusError(err error) error { } } +// newQueuePauseStore enables the file-backed pause store only when +// QUEUE_PAUSE_CONFIG_PATH is set; otherwise no queue is ever paused. +func newQueuePauseStore(logger *zap.Logger) (queuepause.Store, error) { + path := os.Getenv("QUEUE_PAUSE_CONFIG_PATH") + if path == "" { + logger.Info("queue pause disabled") + return queuepausenoop.New(), nil + } + store, err := queuepausefile.NewStore(path) + if err != nil { + return nil, fmt.Errorf("failed to load queue pause config: %w", err) + } + logger.Info("queue pause configured", zap.String("path", path)) + return store, nil +} + func main() { code := 0 if err := run(); err != nil { @@ -264,6 +287,10 @@ func run() error { if err != nil { return fmt.Errorf("failed to load queue configs: %w", err) } + queuePauses, err := newQueuePauseStore(logger) + if err != nil { + return err + } configuredQueues, err := queueConfigs.List(ctx) if err != nil { return fmt.Errorf("failed to list queue configs: %w", err) @@ -359,7 +386,7 @@ func run() error { storageFty := storageFactory{backend: store} materializer := requestcore.NewMaterializer(storageFty) pingController := controller.NewPingController(logger, scope) - landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, registry) + landController := controller.NewLandController(logger.Sugar(), scope, cnt, storageFty, materializer, queueConfigs, queuePauses, registry) cancelController := controller.NewCancelController(logger.Sugar(), scope, storageFty, materializer, registry) requestSummaryController := controller.NewRequestSummaryController(logger.Sugar(), scope, storageFty) listController := controller.NewListController(logger.Sugar(), scope, storageFty, queueConfigs) diff --git a/service/submitqueue/gateway/server/main_test.go b/service/submitqueue/gateway/server/main_test.go index 2fc6cce17..612fd42d5 100644 --- a/service/submitqueue/gateway/server/main_test.go +++ b/service/submitqueue/gateway/server/main_test.go @@ -23,6 +23,7 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/submitqueue/entity" qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" "github.com/uber/submitqueue/submitqueue/gateway/controller" @@ -102,9 +103,14 @@ func TestGatewayStatusError(t *testing.T) { }, { name: "unrecognized queue", - err: &controller.UnrecognizedQueueError{Queue: "missing"}, + err: errs.NewUserError(&controller.UnrecognizedQueueError{Queue: "missing"}), code: codes.InvalidArgument, }, + { + name: "queue paused", + err: errs.NewUserError(&controller.QueuePausedError{Queue: "paused"}), + code: codes.FailedPrecondition, + }, { name: "request not found", err: &controller.RequestNotFoundError{Sqid: "1"}, diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index a56cdec6b..53f91cc4d 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -26,6 +26,7 @@ go_library( "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/gateway/core/request:go_default_library", "//submitqueue/gateway/extension/storage:go_default_library", @@ -63,6 +64,9 @@ go_test( "//submitqueue/entity:go_default_library", "//submitqueue/extension/queueconfig:go_default_library", "//submitqueue/extension/queueconfig/mock:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "//submitqueue/extension/queuepause/fake:go_default_library", + "//submitqueue/extension/queuepause/noop:go_default_library", "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mock:go_default_library", "//submitqueue/gateway/core/request:go_default_library", diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 4819b0893..11b4eaa58 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -31,6 +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/queuepause" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" storage "github.com/uber/submitqueue/submitqueue/gateway/extension/storage" "go.uber.org/zap" @@ -63,6 +64,28 @@ func IsUnrecognizedQueue(err error) bool { return errors.As(err, &target) } +// QueuePausedError indicates the queue is refusing new land requests. +type QueuePausedError struct { + Queue string + // Reason is the operator's explanation for the pause, empty when none was given. + Reason string +} + +// Error implements the error interface. +func (e *QueuePausedError) Error() string { + msg := fmt.Sprintf("queue %q is paused and not accepting land requests", e.Queue) + if e.Reason != "" { + msg += ": " + e.Reason + } + return msg +} + +// IsQueuePaused returns true if any error in the chain is a *QueuePausedError. +func IsQueuePaused(err error) bool { + var target *QueuePausedError + return errors.As(err, &target) +} + // counterDomainRequest names the per-queue sequence that mints request IDs. const counterDomainRequest = "request" @@ -80,13 +103,14 @@ type landController struct { stores storage.Factory materializer *requestcore.Materializer queueConfigs queueconfig.Store + queuePauses queuepause.Store registry consumer.TopicRegistry } // NewLandController creates a new instance of the gateway land controller. // The controller publishes land requests to the topic registered under // topickey.TopicKeyStart in the registry. -func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, registry consumer.TopicRegistry) LandController { +func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters counter.Factory, stores storage.Factory, materializer *requestcore.Materializer, queueConfigs queueconfig.Store, queuePauses queuepause.Store, registry consumer.TopicRegistry) LandController { return &landController{ logger: logger, metricsScope: scope.SubScope("land_controller"), @@ -94,6 +118,7 @@ func NewLandController(logger *zap.SugaredLogger, scope tally.Scope, counters co stores: stores, materializer: materializer, queueConfigs: queueConfigs, + queuePauses: queuePauses, registry: registry, } } @@ -121,6 +146,15 @@ func (c *landController) Land(ctx context.Context, req entity.LandRequest) (resu return entity.LandResult{}, fmt.Errorf("failed to look up queue %q: %w", queue, err) } + // A failed read fails the land: an unreadable pause state must not let a paused queue accept work. + pauseState, err := c.queuePauses.Get(ctx, queue) + if err != nil { + return entity.LandResult{}, fmt.Errorf("failed to read pause state for queue=%s: %w", queue, err) + } + if pauseState.RejectLands { + return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue, Reason: pauseState.Reason}) + } + // Generate a queue-scoped request ID for the land request. // The inbound entity arrives with an empty ID; the controller owns minting it. stores, err := c.stores.For(storage.Config{QueueName: queue}) diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index d1157ba1a..6f31bf913 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -36,6 +36,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/queuepause" + queuepausefake "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake" + queuepausenoop "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop" storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" requestcore "github.com/uber/submitqueue/submitqueue/gateway/core/request" gwstoragemock "github.com/uber/submitqueue/submitqueue/gateway/extension/storage/mock" @@ -87,7 +90,7 @@ func (f staticCounterFactory) For(config counter.Config) (counter.Counter, error // noopStorage returns stateful request storage whose writes succeed. func newNoopLandController(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter) LandController { fixture := newControllerStorageFixture(ctrl) - return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), newTestRegistryWithNoopPublisher(t, ctrl)) + return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) } // noopQueueConfigStore returns a mock queueconfig.Store that always reports @@ -178,6 +181,7 @@ func TestLand_ResolvesCounterForRequestQueue(t *testing.T) { fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), + queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl), ) ctx := context.Background() @@ -307,7 +311,7 @@ func TestLand_ReturnsUnrecognizedQueueWhenStoreReportsNotFound(t *testing.T) { qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) fixture := newControllerStorageFixture(ctrl) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("missing-queue")) @@ -330,7 +334,7 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { qcs.EXPECT().Get(gomock.Any(), "test-queue").Return(entity.QueueConfig{}, fmt.Errorf("config backend down")) fixture := newControllerStorageFixture(ctrl) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, newTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("test-queue")) @@ -340,6 +344,81 @@ func TestLand_PropagatesQueueConfigStoreError(t *testing.T) { assert.False(t, IsInvalidRequest(err)) } +func newLandControllerWithPauseStore(t *testing.T, ctrl *gomock.Controller, cnt counter.Counter, qcs *qcmock.MockStore, pauses queuepause.Store) LandController { + fixture := newControllerStorageFixture(ctrl) + return NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), qcs, pauses, newTestRegistryWithNoopPublisher(t, ctrl)) +} + +func TestLand_QueuePause(t *testing.T) { + tests := []struct { + name string + state entity.QueuePause + wantLanded bool + wantReason string + }{ + {name: "not paused", state: entity.QueuePause{}, wantLanded: true}, + {name: "freeze only still accepts lands", state: entity.QueuePause{FreezeProcessing: true}, wantLanded: true}, + {name: "reject only", state: entity.QueuePause{RejectLands: true}, wantLanded: false}, + {name: "reject and freeze", state: entity.QueuePause{RejectLands: true, FreezeProcessing: true}, wantLanded: false}, + {name: "reject carries the reason", state: entity.QueuePause{RejectLands: true, Reason: "repo freeze until 17:00"}, wantLanded: false, wantReason: "repo freeze until 17:00"}, + {name: "reason alone does not reject", state: entity.QueuePause{Reason: "stale note"}, wantLanded: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + if tt.wantLanded { + cnt.EXPECT().Next(gomock.Any(), gomock.Any()).Return(int64(1), nil) + } + pauses := queuepausefake.New().Set("test-queue", tt.state) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) + + result, err := controller.Land(context.Background(), testLandRequest("test-queue")) + + if tt.wantLanded { + require.NoError(t, err) + assert.Equal(t, "1", result.ID) + return + } + require.Error(t, err) + assert.True(t, IsQueuePaused(err)) + assert.True(t, errs.IsUserError(err)) + assert.False(t, errs.IsRetryable(err)) + var typed *QueuePausedError + require.ErrorAs(t, err, &typed) + assert.Equal(t, "test-queue", typed.Queue) + assert.Equal(t, tt.wantReason, typed.Reason) + }) + } +} + +func TestLand_FailsClosedWhenPauseStoreErrors(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down")) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, noopQueueConfigStore(ctrl), pauses) + + _, err := controller.Land(context.Background(), testLandRequest("test-queue")) + + require.Error(t, err) + assert.False(t, IsQueuePaused(err)) + assert.False(t, errs.IsUserError(err)) +} + +func TestLand_ReportsUnrecognizedQueueBeforeReadingPauseState(t *testing.T) { + ctrl := gomock.NewController(t) + cnt := countermock.NewMockCounter(ctrl) + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "missing-queue").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) + pauses := queuepausefake.New().FailWith(fmt.Errorf("pause backend down")) + controller := newLandControllerWithPauseStore(t, ctrl, cnt, qcs, pauses) + + _, err := controller.Land(context.Background(), testLandRequest("missing-queue")) + + require.Error(t, err) + assert.True(t, IsUnrecognizedQueue(err)) +} + func TestLand_PublishesToQueue(t *testing.T) { var publishedTopic string var publishedMessage entityqueue.Message @@ -414,7 +493,7 @@ func TestLand_PublishesToQueue(t *testing.T) { ), ) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), registry) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, store), materializer, noopQueueConfigStore(ctrl), queuepausenoop.New(), registry) ctx := context.Background() req := entity.LandRequest{ @@ -486,7 +565,7 @@ func TestLand_ReturnsErrorWhenPublishFails(t *testing.T) { registry, publisher := newTestRegistry(t, ctrl) publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable")) - controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), registry) + controller := NewLandController(zap.NewNop().Sugar(), tally.NoopScope, staticCounterFactory{counter: cnt}, factoryForStorage(ctrl, storageWithSummaryStore(ctrl, summaryStore)), newControllerStorageFixture(ctrl).newMaterializer(ctrl), noopQueueConfigStore(ctrl), queuepausenoop.New(), registry) ctx := context.Background() _, err := controller.Land(ctx, testLandRequest("test-queue")) @@ -508,6 +587,7 @@ func TestLand_ReturnsSqidWhenAcceptedLogFailsAfterPublish(t *testing.T) { fixture.newFactory(ctrl), fixture.newMaterializer(ctrl), noopQueueConfigStore(ctrl), + queuepausenoop.New(), newTestRegistryWithNoopPublisher(t, ctrl), ) result, err := controller.Land(context.Background(), testLandRequest("test-queue"))