diff --git a/submitqueue/extension/queuepause/README.md b/submitqueue/extension/queuepause/README.md index 61b714d78..7e3b7f8a9 100644 --- a/submitqueue/extension/queuepause/README.md +++ b/submitqueue/extension/queuepause/README.md @@ -10,6 +10,15 @@ Returns the pause state of a queue by name. The state is two independent switche The store only reports state. The gateway and orchestrator enforce it, so every deployment runs the same rejection and freezing logic and only the source of the state differs. +## Implementations + +- `noop` never pauses a queue. It is the default when no pause state is configured. +- `file` reads a YAML file listing the queues that have switches set, and re-reads it on every `Get`, so editing the file pauses or unpauses a queue without a restart. The file must exist and parse when the store is constructed, so a mistyped path fails at startup. A later read or parse failure is returned as an error and never treated as not paused. Unknown fields are rejected so a misspelled switch cannot silently leave a queue unpaused. + +- `fake` is an in-memory store for tests. A test sets a queue's state, changes it while consumers run, and can make every read fail. It is never used in production. + +A deployer can supply its own implementation backed by a config service. + ## Entities `QueuePause` lives in `submitqueue/entity/queue_pause.go`. diff --git a/submitqueue/extension/queuepause/fake/BUILD.bazel b/submitqueue/extension/queuepause/fake/BUILD.bazel new file mode 100644 index 000000000..5e2d56b30 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["fake.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/fake", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["fake_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/submitqueue/extension/queuepause/fake/fake.go b/submitqueue/extension/queuepause/fake/fake.go new file mode 100644 index 000000000..2cb0d1e30 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/fake.go @@ -0,0 +1,69 @@ +// Copyright (c) 2026 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 fake provides an in-memory queuepause.Store for tests and examples. +// Seed or change a queue's state with Set; a queue never set is not paused. +// FailWith injects a read error on every Get to exercise the error path without +// a real backend. Safe for concurrent use, so a test can change state while a +// consumer is reading. It is intended for tests and examples only, never +// production. +package fake + +import ( + "context" + "sync" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" +) + +var _ queuepause.Store = (*Store)(nil) + +// Store is a programmable in-memory queuepause.Store. +type Store struct { + mu sync.RWMutex + states map[string]entity.QueuePause + err error +} + +// New returns a Store under which no queue is paused. +func New() *Store { + return &Store{states: map[string]entity.QueuePause{}} +} + +// Set replaces the pause state of a queue. Setting the zero state unpauses it. +func (s *Store) Set(queue string, state entity.QueuePause) *Store { + s.mu.Lock() + defer s.mu.Unlock() + s.states[queue] = state + return s +} + +// FailWith makes every Get return err. Passing nil restores normal reads. +func (s *Store) FailWith(err error) *Store { + s.mu.Lock() + defer s.mu.Unlock() + s.err = err + return s +} + +// Get returns the state last set for the queue, or the zero state if none was. +func (s *Store) Get(_ context.Context, queue string) (entity.QueuePause, error) { + s.mu.RLock() + defer s.mu.RUnlock() + if s.err != nil { + return entity.QueuePause{}, s.err + } + return s.states[queue], nil +} diff --git a/submitqueue/extension/queuepause/fake/fake_test.go b/submitqueue/extension/queuepause/fake/fake_test.go new file mode 100644 index 000000000..8198dc389 --- /dev/null +++ b/submitqueue/extension/queuepause/fake/fake_test.go @@ -0,0 +1,64 @@ +// Copyright (c) 2026 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 fake + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestGet(t *testing.T) { + frozen := entity.QueuePause{FreezeProcessing: true, Reason: "incident"} + tests := []struct { + name string + setup func(*Store) + queue string + want entity.QueuePause + }{ + {name: "queue never set is not paused", setup: func(*Store) {}, queue: "q", want: entity.QueuePause{}}, + {name: "queue set is returned", setup: func(s *Store) { s.Set("q", frozen) }, queue: "q", want: frozen}, + {name: "other queue is unaffected", setup: func(s *Store) { s.Set("q", frozen) }, queue: "other", want: entity.QueuePause{}}, + {name: "setting the zero state unpauses", setup: func(s *Store) { s.Set("q", frozen).Set("q", entity.QueuePause{}) }, queue: "q", want: entity.QueuePause{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := New() + tt.setup(s) + + got, err := s.Get(context.Background(), tt.queue) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestFailWith(t *testing.T) { + boom := errors.New("backend down") + s := New().Set("q", entity.QueuePause{RejectLands: true}).FailWith(boom) + + got, err := s.Get(context.Background(), "q") + require.ErrorIs(t, err, boom) + assert.Equal(t, entity.QueuePause{}, got) + + got, err = s.FailWith(nil).Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{RejectLands: true}, got) +} diff --git a/submitqueue/extension/queuepause/file/BUILD.bazel b/submitqueue/extension/queuepause/file/BUILD.bazel new file mode 100644 index 000000000..304c70370 --- /dev/null +++ b/submitqueue/extension/queuepause/file/BUILD.bazel @@ -0,0 +1,24 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["file.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/file", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + "@in_gopkg_yaml_v3//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["file_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/submitqueue/extension/queuepause/file/file.go b/submitqueue/extension/queuepause/file/file.go new file mode 100644 index 000000000..c49bdfe9e --- /dev/null +++ b/submitqueue/extension/queuepause/file/file.go @@ -0,0 +1,112 @@ +// Copyright (c) 2026 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 file provides a YAML-file-backed implementation of queuepause.Store. +// The file is re-read on every Get, so editing it pauses or unpauses a queue +// without a restart. The file lists the queues that have pause switches set: +// +// queues: +// - name: my-queue +// reject_lands: true +// freeze_processing: false +// reason: repo freeze until 17:00 +// +// A queue with no entry is not paused. The file must exist and parse when the +// Store is constructed, so a mistyped path fails at startup. A later read or +// parse failure is returned from Get as an error, never treated as unpaused. +package file + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" + yamlv3 "gopkg.in/yaml.v3" +) + +var _ queuepause.Store = Store{} + +type fileContents struct { + Queues []queueEntry `yaml:"queues"` +} + +type queueEntry struct { + Name string `yaml:"name"` + RejectLands bool `yaml:"reject_lands"` + FreezeProcessing bool `yaml:"freeze_processing"` + Reason string `yaml:"reason"` +} + +// Store is a queuepause.Store backed by a YAML file that is re-read on every +// Get. Safe for concurrent use. +type Store struct { + path string +} + +// NewStore returns a Store reading the YAML file at path. +// Returns an error if the file is unreadable or invalid, so that a +// misconfigured path is caught at startup. +func NewStore(path string) (Store, error) { + if _, err := load(path); err != nil { + return Store{}, err + } + return Store{path: path}, nil +} + +// Get re-reads the file and returns the pause state of the named queue, or the +// zero state if the queue has no entry. +func (s Store) Get(_ context.Context, queue string) (entity.QueuePause, error) { + states, err := load(s.path) + if err != nil { + return entity.QueuePause{}, err + } + return states[queue], nil +} + +func load(path string) (map[string]entity.QueuePause, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("failed to read queue pause file %q: %w", path, err) + } + + // Unknown fields are rejected: a misspelled switch would otherwise be + // ignored and leave the queue unpaused. + decoder := yamlv3.NewDecoder(bytes.NewReader(data)) + decoder.KnownFields(true) + var contents fileContents + if err := decoder.Decode(&contents); err != nil && !errors.Is(err, io.EOF) { + return nil, fmt.Errorf("failed to parse queue pause file %q: %w", path, err) + } + + states := make(map[string]entity.QueuePause, len(contents.Queues)) + for _, q := range contents.Queues { + if q.Name == "" { + return nil, fmt.Errorf("queue pause entry in %q has empty name", path) + } + if _, dup := states[q.Name]; dup { + return nil, fmt.Errorf("queue pause file %q has duplicate name %q", path, q.Name) + } + states[q.Name] = entity.QueuePause{ + RejectLands: q.RejectLands, + FreezeProcessing: q.FreezeProcessing, + Reason: q.Reason, + } + } + return states, nil +} diff --git a/submitqueue/extension/queuepause/file/file_test.go b/submitqueue/extension/queuepause/file/file_test.go new file mode 100644 index 000000000..0981b7e1a --- /dev/null +++ b/submitqueue/extension/queuepause/file/file_test.go @@ -0,0 +1,143 @@ +// Copyright (c) 2026 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 file + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func writeFile(t *testing.T, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "pause.yaml") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + return path +} + +func TestGet(t *testing.T) { + const contents = ` +queues: + - name: rejecting + reject_lands: true + - name: freezing + freeze_processing: true + - name: both + reject_lands: true + freeze_processing: true + - name: with-reason + reject_lands: true + reason: repo freeze until 17:00 + - name: listed-but-running +` + tests := []struct { + name string + queue string + want entity.QueuePause + }{ + {name: "reject only", queue: "rejecting", want: entity.QueuePause{RejectLands: true}}, + {name: "freeze only", queue: "freezing", want: entity.QueuePause{FreezeProcessing: true}}, + {name: "both switches", queue: "both", want: entity.QueuePause{RejectLands: true, FreezeProcessing: true}}, + {name: "reason is carried", queue: "with-reason", want: entity.QueuePause{RejectLands: true, Reason: "repo freeze until 17:00"}}, + {name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePause{}}, + {name: "unlisted queue is not paused", queue: "other", want: entity.QueuePause{}}, + } + store, err := NewStore(writeFile(t, contents)) + require.NoError(t, err) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := store.Get(context.Background(), tt.queue) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestNewStoreEmptyFile(t *testing.T) { + store, err := NewStore(writeFile(t, "")) + require.NoError(t, err) + + got, err := store.Get(context.Background(), "any") + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{}, got) +} + +func TestNewStoreRejectsInvalidFile(t *testing.T) { + tests := []struct { + name string + contents string + }{ + {name: "malformed yaml", contents: "queues: [unterminated"}, + {name: "empty queue name", contents: "queues:\n - reject_lands: true\n"}, + {name: "duplicate queue name", contents: "queues:\n - name: a\n - name: a\n"}, + {name: "misspelled switch", contents: "queues:\n - name: a\n reject_land: true\n"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := NewStore(writeFile(t, tt.contents)) + require.Error(t, err) + }) + } + + t.Run("missing file", func(t *testing.T) { + _, err := NewStore(filepath.Join(t.TempDir(), "absent.yaml")) + require.Error(t, err) + }) +} + +func TestGetReflectsFileChanges(t *testing.T) { + path := writeFile(t, "queues:\n - name: q\n") + store, err := NewStore(path) + require.NoError(t, err) + + got, err := store.Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{}, got) + + require.NoError(t, os.WriteFile(path, []byte("queues:\n - name: q\n reject_lands: true\n"), 0o600)) + got, err = store.Get(context.Background(), "q") + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{RejectLands: true}, got) +} + +func TestGetFailsWhenFileBecomesUnreadable(t *testing.T) { + tests := []struct { + name string + mutate func(t *testing.T, path string) + }{ + {name: "file removed", mutate: func(t *testing.T, path string) { require.NoError(t, os.Remove(path)) }}, + {name: "file corrupted", mutate: func(t *testing.T, path string) { + require.NoError(t, os.WriteFile(path, []byte("queues: [unterminated"), 0o600)) + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + path := writeFile(t, "queues:\n - name: q\n reject_lands: true\n") + store, err := NewStore(path) + require.NoError(t, err) + + tt.mutate(t, path) + + _, err = store.Get(context.Background(), "q") + require.Error(t, err) + }) + } +} diff --git a/submitqueue/extension/queuepause/noop/BUILD.bazel b/submitqueue/extension/queuepause/noop/BUILD.bazel new file mode 100644 index 000000000..e65643a38 --- /dev/null +++ b/submitqueue/extension/queuepause/noop/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["noop.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/queuepause:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["noop_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/submitqueue/extension/queuepause/noop/noop.go b/submitqueue/extension/queuepause/noop/noop.go new file mode 100644 index 000000000..99054da71 --- /dev/null +++ b/submitqueue/extension/queuepause/noop/noop.go @@ -0,0 +1,38 @@ +// Copyright (c) 2026 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 noop provides a queuepause.Store under which no queue is ever paused. +package noop + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queuepause" +) + +var _ queuepause.Store = Store{} + +// Store is a queuepause.Store that reports every queue as not paused. +type Store struct{} + +// New returns a Store that never pauses a queue. +func New() Store { + return Store{} +} + +// Get always returns the zero state. +func (Store) Get(context.Context, string) (entity.QueuePause, error) { + return entity.QueuePause{}, nil +} diff --git a/submitqueue/extension/queuepause/noop/noop_test.go b/submitqueue/extension/queuepause/noop/noop_test.go new file mode 100644 index 000000000..c8924a650 --- /dev/null +++ b/submitqueue/extension/queuepause/noop/noop_test.go @@ -0,0 +1,34 @@ +// Copyright (c) 2026 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 noop + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber/submitqueue/submitqueue/entity" +) + +func TestGetNeverPauses(t *testing.T) { + for _, queue := range []string{"", "any", "another"} { + t.Run(queue, func(t *testing.T) { + got, err := New().Get(context.Background(), queue) + require.NoError(t, err) + assert.Equal(t, entity.QueuePause{}, got) + }) + } +}