Skip to content

Commit 4826405

Browse files
committed
feat(orchestrator): Consume hook events
1 parent 07e57cc commit 4826405

5 files changed

Lines changed: 144 additions & 1 deletion

File tree

service/submitqueue/orchestrator/server/BUILD.bazel

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ go_library(
1616
importpath = "github.com/uber/submitqueue/service/submitqueue/orchestrator/server",
1717
visibility = ["//visibility:private"],
1818
deps = [
19+
"//api/base/hook:go_default_library",
1920
"//api/submitqueue/orchestrator/protopb:go_default_library",
2021
"//platform/buildkite:go_default_library",
2122
"//platform/errs/generic:go_default_library",
@@ -25,6 +26,8 @@ go_library(
2526
"//platform/extension/consumergate/noop:go_default_library",
2627
"//platform/extension/counter:go_default_library",
2728
"//platform/extension/counter/mysql:go_default_library",
29+
"//platform/extension/hook:go_default_library",
30+
"//platform/extension/hook/noop:go_default_library",
2831
"//platform/extension/messagequeue/mysql:go_default_library",
2932
"//platform/git/repo:go_default_library",
3033
"//platform/githubactions:go_default_library",

service/submitqueue/orchestrator/server/main.go

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import (
2929
_ "github.com/go-sql-driver/mysql"
3030

3131
"github.com/uber-go/tally"
32+
basehook "github.com/uber/submitqueue/api/base/hook"
3233
pb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb"
3334
genericerrs "github.com/uber/submitqueue/platform/errs/generic"
3435
mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
@@ -37,6 +38,8 @@ import (
3738
consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
3839
"github.com/uber/submitqueue/platform/extension/counter"
3940
mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql"
41+
hookext "github.com/uber/submitqueue/platform/extension/hook"
42+
hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
4043
queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
4144
"github.com/uber/submitqueue/platform/pipeline"
4245
"github.com/uber/submitqueue/submitqueue/core/changeset"
@@ -199,6 +202,7 @@ func run() error {
199202
Analyzer: profiles.AnalyzerFactory(),
200203
Speculator: profiles.SpeculatorFactory(),
201204
Validator: validatorFactory{},
205+
Hooks: hookResolver{},
202206
}
203207

204208
// Assemble the pipeline: one call builds the topic registry, creates
@@ -439,6 +443,17 @@ func (f counterFactory) For(config counter.Config) (counter.Counter, error) {
439443
return mysqlcounter.NewCounter(f.db, f.scope, config.QueueName), nil
440444
}
441445

446+
// hookResolver sends every event to the no-op hook. Which hooks an event goes
447+
// to is host policy, so the resolver lives here rather than in the extension
448+
// package. A deployment with real integrations swaps this for one that selects
449+
// on the event's source and type.
450+
type hookResolver struct{}
451+
452+
// For returns the hooks that run for event.
453+
func (hookResolver) For(*basehook.HookEvent) []hookext.Hook {
454+
return []hookext.Hook{hooknoop.New()}
455+
}
456+
442457
// validatorFactory routes every queue to the always-passing fake validator.
443458
// Choosing an impl per queue is host policy, so the adapter lives here rather
444459
// than in the extension package. A deployment with real validation swaps this

submitqueue/orchestrator/BUILD.bazel

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
1-
load("@rules_go//go:def.bzl", "go_library")
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
22

33
go_library(
44
name = "go_default_library",
55
srcs = ["pipeline.go"],
66
importpath = "github.com/uber/submitqueue/submitqueue/orchestrator",
77
visibility = ["//visibility:public"],
88
deps = [
9+
"//api/base/hook:go_default_library",
910
"//api/runway/messagequeue:go_default_library",
1011
"//platform/consumer:go_default_library",
1112
"//platform/extension/counter:go_default_library",
13+
"//platform/extension/hook:go_default_library",
14+
"//platform/hook:go_default_library",
1215
"//platform/pipeline:go_default_library",
1316
"//submitqueue/core/topickey:go_default_library",
1417
"//submitqueue/extension/buildrunner:go_default_library",
@@ -35,3 +38,19 @@ go_library(
3538
"@org_uber_go_zap//:go_default_library",
3639
],
3740
)
41+
42+
go_test(
43+
name = "go_default_test",
44+
srcs = ["pipeline_test.go"],
45+
embed = [":go_default_library"],
46+
deps = [
47+
"//api/base/hook:go_default_library",
48+
"//platform/extension/hook:go_default_library",
49+
"//platform/extension/hook/noop:go_default_library",
50+
"//platform/pipeline:go_default_library",
51+
"@com_github_stretchr_testify//assert:go_default_library",
52+
"@com_github_stretchr_testify//require:go_default_library",
53+
"@com_github_uber_go_tally//:go_default_library",
54+
"@org_uber_go_zap//zaptest:go_default_library",
55+
],
56+
)

submitqueue/orchestrator/pipeline.go

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,15 @@
1818
package orchestrator
1919

2020
import (
21+
"fmt"
22+
2123
"github.com/uber-go/tally"
24+
basehook "github.com/uber/submitqueue/api/base/hook"
2225
runwaymq "github.com/uber/submitqueue/api/runway/messagequeue"
2326
"github.com/uber/submitqueue/platform/consumer"
2427
"github.com/uber/submitqueue/platform/extension/counter"
28+
hookext "github.com/uber/submitqueue/platform/extension/hook"
29+
platformhook "github.com/uber/submitqueue/platform/hook"
2530
"github.com/uber/submitqueue/platform/pipeline"
2631
"github.com/uber/submitqueue/submitqueue/core/topickey"
2732
"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
@@ -78,6 +83,11 @@ type Deps struct {
7883

7984
// Validator resolves the validator for each queue.
8085
Validator validator.Factory
86+
87+
// Hooks resolves the hooks that receive each lifecycle event for
88+
// fire-and-forget side effects. Wire a resolver returning noop when the
89+
// deployment has no integrations.
90+
Hooks hookext.Hooks
8191
}
8292

8393
// Stages is the orchestrator's pipeline topology as a typed table.
@@ -223,6 +233,22 @@ var Stages = []pipeline.Stage[Deps]{
223233
return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
224234
},
225235
},
236+
// Any stage can publish here and this one publishes nothing onward, so the
237+
// row sits outside the flow the rest of the table is ordered by.
238+
{
239+
Key: basehook.TopicKeyHook,
240+
Name: "submitqueue-hook",
241+
ConsumerGroup: "orchestrator",
242+
New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
243+
if d.Hooks == nil {
244+
return nil, fmt.Errorf("hooks resolver is required; wire one returning noop when the deployment has no integrations")
245+
}
246+
return platformhook.NewController(d.Logger, d.Scope, d.Hooks, sc.TopicKey, sc.ConsumerGroup), nil
247+
},
248+
DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
249+
return platformhook.NewDLQController(d.Logger, d.Scope, sc.TopicKey, sc.ConsumerGroup), nil
250+
},
251+
},
226252
}
227253

228254
// PublishOnlyTopics declares topics the orchestrator publishes to but does
Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
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 orchestrator
16+
17+
import (
18+
"testing"
19+
20+
"github.com/stretchr/testify/assert"
21+
"github.com/stretchr/testify/require"
22+
"github.com/uber-go/tally"
23+
basehook "github.com/uber/submitqueue/api/base/hook"
24+
hookext "github.com/uber/submitqueue/platform/extension/hook"
25+
hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
26+
"github.com/uber/submitqueue/platform/pipeline"
27+
"go.uber.org/zap/zaptest"
28+
)
29+
30+
type noopHooks struct{}
31+
32+
func (noopHooks) For(*basehook.HookEvent) []hookext.Hook {
33+
return []hookext.Hook{hooknoop.New()}
34+
}
35+
36+
func hookStage(t *testing.T) pipeline.Stage[Deps] {
37+
t.Helper()
38+
for _, s := range Stages {
39+
if s.Key == basehook.TopicKeyHook {
40+
return s
41+
}
42+
}
43+
require.FailNow(t, "no stage is registered for the hook topic key")
44+
return pipeline.Stage[Deps]{}
45+
}
46+
47+
func TestHookStage(t *testing.T) {
48+
deps := func(t *testing.T) Deps {
49+
return Deps{
50+
Logger: zaptest.NewLogger(t).Sugar(),
51+
Scope: tally.NoopScope,
52+
Hooks: noopHooks{},
53+
}
54+
}
55+
stageContext := func(key string) pipeline.StageContext {
56+
return pipeline.StageContext{
57+
TopicKey: basehook.TopicKey(key),
58+
ConsumerGroup: "orchestrator",
59+
}
60+
}
61+
62+
t.Run("the hook controller subscribes to the key the engine assigns", func(t *testing.T) {
63+
controller, err := hookStage(t).New(deps(t), stageContext("hook"))
64+
require.NoError(t, err)
65+
assert.Equal(t, basehook.TopicKeyHook, controller.TopicKey())
66+
})
67+
68+
t.Run("the dead-letter controller subscribes to the key the engine derives", func(t *testing.T) {
69+
controller, err := hookStage(t).DLQ(deps(t), stageContext("hook_dlq"))
70+
require.NoError(t, err)
71+
assert.Equal(t, basehook.TopicKey("hook_dlq"), controller.TopicKey())
72+
})
73+
74+
t.Run("a host that wires no hooks resolver fails to construct", func(t *testing.T) {
75+
d := deps(t)
76+
d.Hooks = nil
77+
_, err := hookStage(t).New(d, stageContext("hook"))
78+
require.Error(t, err)
79+
})
80+
}

0 commit comments

Comments
 (0)