Skip to content

Commit d39ccab

Browse files
manjari25manjari25
committed
feat(submitqueue): Add noop and file queuepause store
1 parent d1a293e commit d39ccab

7 files changed

Lines changed: 374 additions & 0 deletions

File tree

‎submitqueue/extension/queuepause/README.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,13 @@ Returns the pause state of a queue by name. The state is two independent switche
1010

1111
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.
1212

13+
## Implementations
14+
15+
- `noop` never pauses a queue. It is the default when no pause state is configured.
16+
- `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.
17+
18+
A deployer can supply its own implementation backed by a config service.
19+
1320
## Entities
1421

1522
`QueuePauseState` lives in `submitqueue/entity/queue_pause_state.go`.

‎submitqueue/extension/queuepause/file/BUILD.bazel‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["file.go"],
6+
importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/file",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//submitqueue/entity:go_default_library",
10+
"//submitqueue/extension/queuepause:go_default_library",
11+
"@in_gopkg_yaml_v3//:go_default_library",
12+
],
13+
)
14+
15+
go_test(
16+
name = "go_default_test",
17+
srcs = ["file_test.go"],
18+
embed = [":go_default_library"],
19+
deps = [
20+
"//submitqueue/entity:go_default_library",
21+
"@com_github_stretchr_testify//assert:go_default_library",
22+
"@com_github_stretchr_testify//require:go_default_library",
23+
],
24+
)

‎submitqueue/extension/queuepause/file/file.go‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package file provides a YAML-file-backed implementation of queuepause.Store.
16+
// The file is re-read on every Get, so editing it pauses or unpauses a queue
17+
// without a restart. The file lists the queues that have pause switches set:
18+
//
19+
// queues:
20+
// - name: my-queue
21+
// reject_lands: true
22+
// freeze_processing: false
23+
//
24+
// A queue with no entry is not paused. The file must exist and parse when the
25+
// Store is constructed, so a mistyped path fails at startup. A later read or
26+
// parse failure is returned from Get as an error, never treated as unpaused.
27+
package file
28+
29+
import (
30+
"bytes"
31+
"context"
32+
"errors"
33+
"fmt"
34+
"io"
35+
"os"
36+
37+
"github.com/uber/submitqueue/submitqueue/entity"
38+
"github.com/uber/submitqueue/submitqueue/extension/queuepause"
39+
yamlv3 "gopkg.in/yaml.v3"
40+
)
41+
42+
var _ queuepause.Store = Store{}
43+
44+
type fileContents struct {
45+
Queues []queueEntry `yaml:"queues"`
46+
}
47+
48+
type queueEntry struct {
49+
Name string `yaml:"name"`
50+
RejectLands bool `yaml:"reject_lands"`
51+
FreezeProcessing bool `yaml:"freeze_processing"`
52+
}
53+
54+
// Store is a queuepause.Store backed by a YAML file that is re-read on every
55+
// Get. Safe for concurrent use.
56+
type Store struct {
57+
path string
58+
}
59+
60+
// NewStore returns a Store reading the YAML file at path.
61+
// Returns an error if the file is unreadable or invalid, so that a
62+
// misconfigured path is caught at startup.
63+
func NewStore(path string) (Store, error) {
64+
if _, err := load(path); err != nil {
65+
return Store{}, err
66+
}
67+
return Store{path: path}, nil
68+
}
69+
70+
// Get re-reads the file and returns the pause state of the named queue, or the
71+
// zero state if the queue has no entry.
72+
func (s Store) Get(_ context.Context, queue string) (entity.QueuePauseState, error) {
73+
states, err := load(s.path)
74+
if err != nil {
75+
return entity.QueuePauseState{}, err
76+
}
77+
return states[queue], nil
78+
}
79+
80+
func load(path string) (map[string]entity.QueuePauseState, error) {
81+
data, err := os.ReadFile(path)
82+
if err != nil {
83+
return nil, fmt.Errorf("failed to read queue pause file %q: %w", path, err)
84+
}
85+
86+
// Unknown fields are rejected: a misspelled switch would otherwise be
87+
// ignored and leave the queue unpaused.
88+
decoder := yamlv3.NewDecoder(bytes.NewReader(data))
89+
decoder.KnownFields(true)
90+
var contents fileContents
91+
if err := decoder.Decode(&contents); err != nil && !errors.Is(err, io.EOF) {
92+
return nil, fmt.Errorf("failed to parse queue pause file %q: %w", path, err)
93+
}
94+
95+
states := make(map[string]entity.QueuePauseState, len(contents.Queues))
96+
for _, q := range contents.Queues {
97+
if q.Name == "" {
98+
return nil, fmt.Errorf("queue pause entry in %q has empty name", path)
99+
}
100+
if _, dup := states[q.Name]; dup {
101+
return nil, fmt.Errorf("queue pause file %q has duplicate name %q", path, q.Name)
102+
}
103+
states[q.Name] = entity.QueuePauseState{
104+
RejectLands: q.RejectLands,
105+
FreezeProcessing: q.FreezeProcessing,
106+
}
107+
}
108+
return states, nil
109+
}

‎submitqueue/extension/queuepause/file/file_test.go‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package file
16+
17+
import (
18+
"context"
19+
"os"
20+
"path/filepath"
21+
"testing"
22+
23+
"github.com/stretchr/testify/assert"
24+
"github.com/stretchr/testify/require"
25+
"github.com/uber/submitqueue/submitqueue/entity"
26+
)
27+
28+
func writeFile(t *testing.T, contents string) string {
29+
t.Helper()
30+
path := filepath.Join(t.TempDir(), "pause.yaml")
31+
require.NoError(t, os.WriteFile(path, []byte(contents), 0o600))
32+
return path
33+
}
34+
35+
func TestGet(t *testing.T) {
36+
const contents = `
37+
queues:
38+
- name: rejecting
39+
reject_lands: true
40+
- name: freezing
41+
freeze_processing: true
42+
- name: both
43+
reject_lands: true
44+
freeze_processing: true
45+
- name: listed-but-running
46+
`
47+
tests := []struct {
48+
name string
49+
queue string
50+
want entity.QueuePauseState
51+
}{
52+
{name: "reject only", queue: "rejecting", want: entity.QueuePauseState{RejectLands: true}},
53+
{name: "freeze only", queue: "freezing", want: entity.QueuePauseState{FreezeProcessing: true}},
54+
{name: "both switches", queue: "both", want: entity.QueuePauseState{RejectLands: true, FreezeProcessing: true}},
55+
{name: "listed with no switches set", queue: "listed-but-running", want: entity.QueuePauseState{}},
56+
{name: "unlisted queue is not paused", queue: "other", want: entity.QueuePauseState{}},
57+
}
58+
store, err := NewStore(writeFile(t, contents))
59+
require.NoError(t, err)
60+
61+
for _, tt := range tests {
62+
t.Run(tt.name, func(t *testing.T) {
63+
got, err := store.Get(context.Background(), tt.queue)
64+
require.NoError(t, err)
65+
assert.Equal(t, tt.want, got)
66+
})
67+
}
68+
}
69+
70+
func TestNewStoreEmptyFile(t *testing.T) {
71+
store, err := NewStore(writeFile(t, ""))
72+
require.NoError(t, err)
73+
74+
got, err := store.Get(context.Background(), "any")
75+
require.NoError(t, err)
76+
assert.Equal(t, entity.QueuePauseState{}, got)
77+
}
78+
79+
func TestNewStoreRejectsInvalidFile(t *testing.T) {
80+
tests := []struct {
81+
name string
82+
contents string
83+
}{
84+
{name: "malformed yaml", contents: "queues: [unterminated"},
85+
{name: "empty queue name", contents: "queues:\n - reject_lands: true\n"},
86+
{name: "duplicate queue name", contents: "queues:\n - name: a\n - name: a\n"},
87+
{name: "misspelled switch", contents: "queues:\n - name: a\n reject_land: true\n"},
88+
}
89+
for _, tt := range tests {
90+
t.Run(tt.name, func(t *testing.T) {
91+
_, err := NewStore(writeFile(t, tt.contents))
92+
require.Error(t, err)
93+
})
94+
}
95+
96+
t.Run("missing file", func(t *testing.T) {
97+
_, err := NewStore(filepath.Join(t.TempDir(), "absent.yaml"))
98+
require.Error(t, err)
99+
})
100+
}
101+
102+
func TestGetReflectsFileChanges(t *testing.T) {
103+
path := writeFile(t, "queues:\n - name: q\n")
104+
store, err := NewStore(path)
105+
require.NoError(t, err)
106+
107+
got, err := store.Get(context.Background(), "q")
108+
require.NoError(t, err)
109+
assert.Equal(t, entity.QueuePauseState{}, got)
110+
111+
require.NoError(t, os.WriteFile(path, []byte("queues:\n - name: q\n reject_lands: true\n"), 0o600))
112+
got, err = store.Get(context.Background(), "q")
113+
require.NoError(t, err)
114+
assert.Equal(t, entity.QueuePauseState{RejectLands: true}, got)
115+
}
116+
117+
func TestGetFailsWhenFileBecomesUnreadable(t *testing.T) {
118+
tests := []struct {
119+
name string
120+
mutate func(t *testing.T, path string)
121+
}{
122+
{name: "file removed", mutate: func(t *testing.T, path string) { require.NoError(t, os.Remove(path)) }},
123+
{name: "file corrupted", mutate: func(t *testing.T, path string) {
124+
require.NoError(t, os.WriteFile(path, []byte("queues: [unterminated"), 0o600))
125+
}},
126+
}
127+
for _, tt := range tests {
128+
t.Run(tt.name, func(t *testing.T) {
129+
path := writeFile(t, "queues:\n - name: q\n reject_lands: true\n")
130+
store, err := NewStore(path)
131+
require.NoError(t, err)
132+
133+
tt.mutate(t, path)
134+
135+
_, err = store.Get(context.Background(), "q")
136+
require.Error(t, err)
137+
})
138+
}
139+
}

‎submitqueue/extension/queuepause/noop/BUILD.bazel‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["noop.go"],
6+
importpath = "github.com/uber/submitqueue/submitqueue/extension/queuepause/noop",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//submitqueue/entity:go_default_library",
10+
"//submitqueue/extension/queuepause:go_default_library",
11+
],
12+
)
13+
14+
go_test(
15+
name = "go_default_test",
16+
srcs = ["noop_test.go"],
17+
embed = [":go_default_library"],
18+
deps = [
19+
"//submitqueue/entity:go_default_library",
20+
"@com_github_stretchr_testify//assert:go_default_library",
21+
"@com_github_stretchr_testify//require:go_default_library",
22+
],
23+
)

‎submitqueue/extension/queuepause/noop/noop.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package noop provides a queuepause.Store under which no queue is ever paused.
16+
package noop
17+
18+
import (
19+
"context"
20+
21+
"github.com/uber/submitqueue/submitqueue/entity"
22+
"github.com/uber/submitqueue/submitqueue/extension/queuepause"
23+
)
24+
25+
var _ queuepause.Store = Store{}
26+
27+
// Store is a queuepause.Store that reports every queue as not paused.
28+
type Store struct{}
29+
30+
// New returns a Store that never pauses a queue.
31+
func New() Store {
32+
return Store{}
33+
}
34+
35+
// Get always returns the zero state.
36+
func (Store) Get(context.Context, string) (entity.QueuePauseState, error) {
37+
return entity.QueuePauseState{}, nil
38+
}

‎submitqueue/extension/queuepause/noop/noop_test.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package noop
16+
17+
import (
18+
"context"
19+
"testing"
20+
21+
"github.com/stretchr/testify/assert"
22+
"github.com/stretchr/testify/require"
23+
"github.com/uber/submitqueue/submitqueue/entity"
24+
)
25+
26+
func TestGetNeverPauses(t *testing.T) {
27+
for _, queue := range []string{"", "any", "another"} {
28+
t.Run(queue, func(t *testing.T) {
29+
got, err := New().Get(context.Background(), queue)
30+
require.NoError(t, err)
31+
assert.Equal(t, entity.QueuePauseState{}, got)
32+
})
33+
}
34+
}

0 commit comments

Comments
 (0)