Skip to content

Commit e091f18

Browse files
JamyDevclaude
JamyDev
and
claude
committed
feat(orchestrator): add pipeline self-declaration
Introduce `submitqueue/orchestrator/pipeline.go` with three exports: - `Deps` struct: the full set of dependencies (logger, scope, storage, counter, and four extension factories) the orchestrator pipeline needs. This struct IS the service's public API toward deployers. - `Stages` slice: the complete 12-stage pipeline topology as a typed table of `pipeline.Stage[Deps]`. Each row declares a primary controller constructor and its DLQ reconciler. Adding a stage = adding one row. - `PublishOnlyTopics`: topics the orchestrator publishes to but does not consume (log, merge-conflict-check, runway-merge). - `Controllers` struct + `NewControllers`: RPC-facing controllers (currently Ping), NOT bound to any wire contract. Pure addition alongside the existing main.go wiring — no behavioral changes. Ref: doc/rfc/submitqueue/modular-queue-wiring.md (Step 3) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 6159f66 commit e091f18

2 files changed

Lines changed: 291 additions & 0 deletions

File tree

‎submitqueue/orchestrator/BUILD.bazel‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
load("@rules_go//go:def.bzl", "go_library")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["pipeline.go"],
6+
importpath = "github.com/uber/submitqueue/submitqueue/orchestrator",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//api/runway/messagequeue:go_default_library",
10+
"//platform/consumer:go_default_library",
11+
"//platform/extension/counter:go_default_library",
12+
"//platform/pipeline:go_default_library",
13+
"//submitqueue/core/topickey:go_default_library",
14+
"//submitqueue/extension/buildrunner:go_default_library",
15+
"//submitqueue/extension/changeprovider:go_default_library",
16+
"//submitqueue/extension/conflict:go_default_library",
17+
"//submitqueue/extension/scorer:go_default_library",
18+
"//submitqueue/extension/storage:go_default_library",
19+
"//submitqueue/extension/validator:go_default_library",
20+
"//submitqueue/orchestrator/controller:go_default_library",
21+
"//submitqueue/orchestrator/controller/batch:go_default_library",
22+
"//submitqueue/orchestrator/controller/build:go_default_library",
23+
"//submitqueue/orchestrator/controller/buildsignal:go_default_library",
24+
"//submitqueue/orchestrator/controller/cancel:go_default_library",
25+
"//submitqueue/orchestrator/controller/conclude:go_default_library",
26+
"//submitqueue/orchestrator/controller/dlq:go_default_library",
27+
"//submitqueue/orchestrator/controller/merge:go_default_library",
28+
"//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library",
29+
"//submitqueue/orchestrator/controller/mergesignal:go_default_library",
30+
"//submitqueue/orchestrator/controller/score:go_default_library",
31+
"//submitqueue/orchestrator/controller/speculate:go_default_library",
32+
"//submitqueue/orchestrator/controller/start:go_default_library",
33+
"//submitqueue/orchestrator/controller/validate:go_default_library",
34+
"@com_github_uber_go_tally//:go_default_library",
35+
"@org_uber_go_zap//:go_default_library",
36+
],
37+
)

‎submitqueue/orchestrator/pipeline.go‎

Lines changed: 254 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,254 @@
1+
// Copyright (c) 2025 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 orchestrator declares the SubmitQueue orchestrator's pipeline
16+
// topology, extension seams, and controller set. The host (main.go) fills
17+
// Deps and passes Stages to pipeline.Construct; no assembly logic lives here.
18+
package orchestrator
19+
20+
import (
21+
"github.com/uber-go/tally"
22+
runwaymq "github.com/uber/submitqueue/api/runway/messagequeue"
23+
"github.com/uber/submitqueue/platform/consumer"
24+
"github.com/uber/submitqueue/platform/extension/counter"
25+
"github.com/uber/submitqueue/platform/pipeline"
26+
"github.com/uber/submitqueue/submitqueue/core/topickey"
27+
"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
28+
"github.com/uber/submitqueue/submitqueue/extension/changeprovider"
29+
"github.com/uber/submitqueue/submitqueue/extension/conflict"
30+
"github.com/uber/submitqueue/submitqueue/extension/scorer"
31+
"github.com/uber/submitqueue/submitqueue/extension/storage"
32+
"github.com/uber/submitqueue/submitqueue/extension/validator"
33+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller"
34+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/batch"
35+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/build"
36+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal"
37+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel"
38+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/conclude"
39+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq"
40+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge"
41+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal"
42+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal"
43+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/score"
44+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate"
45+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/start"
46+
"github.com/uber/submitqueue/submitqueue/orchestrator/controller/validate"
47+
"go.uber.org/zap"
48+
)
49+
50+
// Deps is the full set of dependencies the orchestrator pipeline needs.
51+
// This struct IS the service's public API toward deployers: fill every
52+
// field, pass it and Stages to pipeline.Construct, and you get a running
53+
// orchestrator pipeline.
54+
type Deps struct {
55+
// Logger is the structured logger for all controllers.
56+
Logger *zap.SugaredLogger
57+
58+
// Scope is the metrics scope for all controllers.
59+
Scope tally.Scope
60+
61+
// Storage provides request, batch, and change stores.
62+
Storage storage.Storage
63+
64+
// Counter provides distributed batch counters.
65+
Counter counter.Counter
66+
67+
// BuildRunner resolves the build runner for each queue.
68+
BuildRunner buildrunner.Factory
69+
70+
// ChangeProvider resolves the change provider for each queue.
71+
ChangeProvider changeprovider.Factory
72+
73+
// Scorer resolves the scorer for each queue.
74+
Scorer scorer.Factory
75+
76+
// Analyzer resolves the conflict analyzer for each queue.
77+
Analyzer conflict.Factory
78+
79+
// Validator resolves the validator for each queue.
80+
Validator validator.Factory
81+
}
82+
83+
// Stages is the orchestrator's pipeline topology as a typed table.
84+
// Adding a stage = adding one row. Nothing else, anywhere.
85+
//
86+
// Pipeline:
87+
//
88+
// start → cancel → validate ⇢ (runway) ⇢ mergeconflictsignal → batch → score → speculate → build → buildsignal ─┐
89+
// ↑ ↘ ↻ poll │
90+
// │ merge → conclude │
91+
// │ │ │
92+
// └────────┴───────────────────────┘
93+
var Stages = []pipeline.Stage[Deps]{
94+
{
95+
Key: topickey.TopicKeyStart,
96+
Name: "start",
97+
ConsumerGroup: "orchestrator-start",
98+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
99+
return start.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
100+
},
101+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
102+
return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeLandRequestID, sc.TopicKey, sc.ConsumerGroup), nil
103+
},
104+
},
105+
{
106+
Key: topickey.TopicKeyCancel,
107+
Name: "cancel",
108+
ConsumerGroup: "orchestrator-cancel",
109+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
110+
return cancel.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
111+
},
112+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
113+
return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeCancelRequestID, sc.TopicKey, sc.ConsumerGroup), nil
114+
},
115+
},
116+
{
117+
Key: topickey.TopicKeyValidate,
118+
Name: "validate",
119+
ConsumerGroup: "orchestrator-validate",
120+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
121+
return validate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, d.ChangeProvider, d.Validator, runwaymq.TopicKeyMergeConflictCheck, sc.TopicKey, sc.ConsumerGroup), nil
122+
},
123+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
124+
return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil
125+
},
126+
},
127+
{
128+
Key: runwaymq.TopicKeyMergeConflictCheckSignal,
129+
Name: "merge-conflict-check-signal",
130+
ConsumerGroup: "orchestrator-mergeconflictsignal",
131+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
132+
return mergeconflictsignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
133+
},
134+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
135+
return dlq.NewDLQMergeConflictSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
136+
},
137+
},
138+
{
139+
Key: topickey.TopicKeyBatch,
140+
Name: "batch",
141+
ConsumerGroup: "orchestrator-batch",
142+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
143+
return batch.NewController(d.Logger, d.Scope, sc.Registry, d.Counter, d.Storage, d.Analyzer, sc.TopicKey, sc.ConsumerGroup), nil
144+
},
145+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
146+
return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil
147+
},
148+
},
149+
{
150+
Key: topickey.TopicKeyScore,
151+
Name: "score",
152+
ConsumerGroup: "orchestrator-score",
153+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
154+
return score.NewController(d.Logger, d.Scope, d.Storage, d.Scorer, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
155+
},
156+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
157+
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
158+
},
159+
},
160+
{
161+
Key: topickey.TopicKeySpeculate,
162+
Name: "speculate",
163+
ConsumerGroup: "orchestrator-speculate",
164+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
165+
return speculate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
166+
},
167+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
168+
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
169+
},
170+
},
171+
{
172+
Key: topickey.TopicKeyBuild,
173+
Name: "build",
174+
ConsumerGroup: "orchestrator-build",
175+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
176+
return build.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
177+
},
178+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
179+
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
180+
},
181+
},
182+
{
183+
Key: topickey.TopicKeyBuildSignal,
184+
Name: "buildsignal",
185+
ConsumerGroup: "orchestrator-buildsignal",
186+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
187+
return buildsignal.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
188+
},
189+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
190+
return dlq.NewDLQBuildSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
191+
},
192+
},
193+
{
194+
Key: topickey.TopicKeyMerge,
195+
Name: "submitqueue-merge",
196+
ConsumerGroup: "orchestrator-merge",
197+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
198+
return merge.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, runwaymq.TopicKeyMerge, sc.TopicKey, sc.ConsumerGroup), nil
199+
},
200+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
201+
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
202+
},
203+
},
204+
{
205+
Key: runwaymq.TopicKeyMergeSignal,
206+
Name: "merge-signal",
207+
ConsumerGroup: "orchestrator-mergesignal",
208+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
209+
return mergesignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
210+
},
211+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
212+
return dlq.NewDLQMergeSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
213+
},
214+
},
215+
{
216+
Key: topickey.TopicKeyConclude,
217+
Name: "conclude",
218+
ConsumerGroup: "orchestrator-conclude",
219+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
220+
return conclude.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
221+
},
222+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
223+
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
224+
},
225+
},
226+
}
227+
228+
// PublishOnlyTopics declares topics the orchestrator publishes to but does
229+
// not consume. These are registered in the TopicRegistry so controllers
230+
// can look up topic names for publishing.
231+
var PublishOnlyTopics = []pipeline.PublishOnlyTopic{
232+
// Log: the orchestrator emits request-log entries; the gateway consumes them.
233+
{Key: topickey.TopicKeyLog, Name: "log"},
234+
// Merge-conflict check: the orchestrator publishes check requests to runway.
235+
{Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check"},
236+
// Merge: the orchestrator publishes merge requests to runway.
237+
{Key: runwaymq.TopicKeyMerge, Name: "runway-merge"},
238+
}
239+
240+
// Controllers holds the orchestrator's RPC-facing controllers, constructed
241+
// but NOT bound to any wire contract. Binding to a proto service + transport
242+
// is host glue, because deployers may use different protos or transports.
243+
type Controllers struct {
244+
// Ping is the health-check controller.
245+
Ping *controller.PingController
246+
}
247+
248+
// NewControllers creates the orchestrator's RPC controllers from the given Deps.
249+
// The PingController takes a base *zap.Logger, so we desugar the SugaredLogger.
250+
func NewControllers(d Deps) Controllers {
251+
return Controllers{
252+
Ping: controller.NewPingController(d.Logger.Desugar(), d.Scope),
253+
}
254+
}

0 commit comments

Comments
 (0)