diff --git a/platform/lifecycle/BUILD.bazel b/platform/lifecycle/BUILD.bazel new file mode 100644 index 00000000..2033e6dc --- /dev/null +++ b/platform/lifecycle/BUILD.bazel @@ -0,0 +1,18 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["lifecycle.go"], + importpath = "github.com/uber/submitqueue/platform/lifecycle", + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["lifecycle_test.go"], + embed = [":go_default_library"], + deps = [ + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + ], +) diff --git a/platform/lifecycle/lifecycle.go b/platform/lifecycle/lifecycle.go new file mode 100644 index 00000000..8c76eb24 --- /dev/null +++ b/platform/lifecycle/lifecycle.go @@ -0,0 +1,94 @@ +// Copyright (c) 2025 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 lifecycle provides the Component interface and Group type for +// managing ordered start/stop lifecycles. Every runnable subsystem (consumer, +// publisher, server) implements Component; Group composes them into a single +// Component with deterministic ordering and rollback on partial failure. +package lifecycle + +import ( + "context" + "errors" + "fmt" +) + +// Component is anything with a lifecycle. Construct returns one; hosts drive it. +type Component interface { + // Start initializes and starts the component. The context governs the + // start-up phase (e.g. connecting, subscribing); long-running work may + // outlive the context and must be terminated by calling Stop. + Start(ctx context.Context) error + + // Stop gracefully shuts down the component. The context provides a + // deadline for the shutdown; implementations should respect it and + // return promptly when the context is cancelled. + Stop(ctx context.Context) error +} + +// Group runs an ordered list of Components as one Component. +// +// - Start: members in order; if member i fails to start, members i-1…0 are +// stopped in reverse and the error is returned — no half-started state. +// - Stop: members in REVERSE order (work-acceptors drain before the +// connections under them close); errors joined, none swallowed. +type Group struct { + members []Component +} + +// NewGroup creates a Group from the given components. Nil members are silently +// skipped so callers can pass optional components without nil-checking. +func NewGroup(members ...Component) *Group { + filtered := make([]Component, 0, len(members)) + for _, m := range members { + if m != nil { + filtered = append(filtered, m) + } + } + return &Group{members: filtered} +} + +// Start starts all members in order. If any member fails to start, all +// previously started members are stopped in reverse order and the original +// start error is returned. The stop errors from rollback, if any, are joined +// with the start error. +func (g *Group) Start(ctx context.Context) error { + for i, m := range g.members { + if err := m.Start(ctx); err != nil { + // Rollback: stop members i-1…0 in reverse order. + rollbackErr := g.stopRange(ctx, i-1) + return errors.Join(fmt.Errorf("component %d failed to start: %w", i, err), rollbackErr) + } + } + return nil +} + +// Stop stops all members in reverse order. All stop errors are joined so +// none is swallowed; a single member's failure does not prevent the others +// from being stopped. +func (g *Group) Stop(ctx context.Context) error { + return g.stopRange(ctx, len(g.members)-1) +} + +// stopRange stops members from index hi down to 0 (inclusive), collecting +// all errors. A negative hi is a no-op. +func (g *Group) stopRange(ctx context.Context, hi int) error { + var errs []error + for i := hi; i >= 0; i-- { + if err := g.members[i].Stop(ctx); err != nil { + errs = append(errs, fmt.Errorf("component %d failed to stop: %w", i, err)) + } + } + return errors.Join(errs...) +} diff --git a/platform/lifecycle/lifecycle_test.go b/platform/lifecycle/lifecycle_test.go new file mode 100644 index 00000000..d58c29b7 --- /dev/null +++ b/platform/lifecycle/lifecycle_test.go @@ -0,0 +1,182 @@ +// Copyright (c) 2025 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 lifecycle + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// spy records the order of Start/Stop calls and can be configured to fail. +type spy struct { + name string + startErr error + stopErr error + log *[]string +} + +func (s *spy) Start(_ context.Context) error { + *s.log = append(*s.log, "start:"+s.name) + return s.startErr +} + +func (s *spy) Stop(_ context.Context) error { + *s.log = append(*s.log, "stop:"+s.name) + return s.stopErr +} + +func TestGroup_StartStop_HappyPath(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + + require.NoError(t, g.Start(context.Background())) + assert.Equal(t, []string{"start:a", "start:b", "start:c"}, log) + + log = nil + require.NoError(t, g.Stop(context.Background())) + assert.Equal(t, []string{"stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_StartRollback_OnFailure(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", startErr: fmt.Errorf("b broke"), log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "b broke") + + // a was started and then rolled back; b failed; c was never started + assert.Equal(t, []string{"start:a", "start:b", "stop:a"}, log) +} + +func TestGroup_StartRollback_FirstMemberFails(t *testing.T) { + var log []string + a := &spy{name: "a", startErr: fmt.Errorf("a broke"), log: &log} + b := &spy{name: "b", log: &log} + + g := NewGroup(a, b) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "a broke") + + // Nothing to roll back — a failed on start, b never started + assert.Equal(t, []string{"start:a"}, log) +} + +func TestGroup_StartRollback_JoinsStopErrors(t *testing.T) { + var log []string + a := &spy{name: "a", stopErr: fmt.Errorf("a stop failed"), log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", startErr: fmt.Errorf("c broke"), log: &log} + + g := NewGroup(a, b, c) + + err := g.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "c broke") + assert.Contains(t, err.Error(), "a stop failed") + + // a and b started, c failed, then b and a rolled back in reverse + assert.Equal(t, []string{"start:a", "start:b", "start:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_Stop_CollectsAllErrors(t *testing.T) { + var log []string + a := &spy{name: "a", stopErr: fmt.Errorf("a stop failed"), log: &log} + b := &spy{name: "b", stopErr: fmt.Errorf("b stop failed"), log: &log} + c := &spy{name: "c", log: &log} + + g := NewGroup(a, b, c) + require.NoError(t, g.Start(context.Background())) + + log = nil + err := g.Stop(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "a stop failed") + assert.Contains(t, err.Error(), "b stop failed") + + // All three stopped in reverse despite errors + assert.Equal(t, []string{"stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_NilMembers_Skipped(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + + g := NewGroup(nil, a, nil) + + require.NoError(t, g.Start(context.Background())) + assert.Equal(t, []string{"start:a"}, log) + + log = nil + require.NoError(t, g.Stop(context.Background())) + assert.Equal(t, []string{"stop:a"}, log) +} + +func TestGroup_Empty(t *testing.T) { + g := NewGroup() + require.NoError(t, g.Start(context.Background())) + require.NoError(t, g.Stop(context.Background())) +} + +func TestGroup_Nested(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", log: &log} + d := &spy{name: "d", log: &log} + + inner := NewGroup(b, c) + outer := NewGroup(a, inner, d) + + require.NoError(t, outer.Start(context.Background())) + assert.Equal(t, []string{"start:a", "start:b", "start:c", "start:d"}, log) + + log = nil + require.NoError(t, outer.Stop(context.Background())) + assert.Equal(t, []string{"stop:d", "stop:c", "stop:b", "stop:a"}, log) +} + +func TestGroup_Nested_RollbackOnInnerFailure(t *testing.T) { + var log []string + a := &spy{name: "a", log: &log} + b := &spy{name: "b", log: &log} + c := &spy{name: "c", startErr: fmt.Errorf("c broke"), log: &log} + d := &spy{name: "d", log: &log} + + inner := NewGroup(b, c) + outer := NewGroup(a, inner, d) + + err := outer.Start(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "c broke") + + // a started, inner started b then c failed, inner rolled back b, + // then outer rolled back a. d never started. + assert.Equal(t, []string{"start:a", "start:b", "start:c", "stop:b", "stop:a"}, log) +} diff --git a/platform/pipeline/BUILD.bazel b/platform/pipeline/BUILD.bazel new file mode 100644 index 00000000..001b3468 --- /dev/null +++ b/platform/pipeline/BUILD.bazel @@ -0,0 +1,32 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["pipeline.go"], + importpath = "github.com/uber/submitqueue/platform/pipeline", + visibility = ["//visibility:public"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/errs:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/lifecycle:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["pipeline_test.go"], + embed = [":go_default_library"], + deps = [ + "//platform/consumer:go_default_library", + "//platform/extension/messagequeue:go_default_library", + "//platform/extension/messagequeue/mock:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/platform/pipeline/pipeline.go b/platform/pipeline/pipeline.go new file mode 100644 index 00000000..e3a73538 --- /dev/null +++ b/platform/pipeline/pipeline.go @@ -0,0 +1,309 @@ +// Copyright (c) 2025 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 pipeline provides a typed engine for assembling queue-driven +// service pipelines from declarative data. A service declares its topology +// as a []Stage[D] table and its dependencies as a Deps struct; Construct +// builds all consumers, registers controllers, pairs DLQ stages, and +// returns a single lifecycle.Component the host drives with Start/Stop. +package pipeline + +import ( + "context" + "fmt" + "time" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/errs" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + "github.com/uber/submitqueue/platform/lifecycle" + "go.uber.org/zap" +) + +// timeNow is a hook for tests to control time. Production uses time.Now. +var timeNow = time.Now + +// Stage is one row of a service's topology table. D is the service's Deps type. +type Stage[D any] struct { + // Key is the stage's logical topic key (e.g. topickey.TopicKeyStart). + // The engine maps it to a physical topic name via the TopicNames option. + Key consumer.TopicKey + + // Name is the physical topic name for this stage (e.g. "start"). + // Used as the default when no TopicNames override is provided. + Name string + + // ConsumerGroup is the consumer group suffix for this stage's subscription + // (e.g. "orchestrator-start"). + ConsumerGroup string + + // New builds the stage's controller from the service's Deps and engine- + // provided StageContext. The engine calls it once, eagerly, inside + // Construct — so a nil/missing dependency fails at boot with the stage's + // name on it, never mid-delivery. + New func(D, StageContext) (consumer.Controller, error) + + // DLQ, when non-nil, declares "this stage dead-letters". The engine then + // derives the paired DLQ topic (_dlq, retry budget, DLQ-of-DLQ + // disabled) AND registers this reconciler on the DLQ consumer. Declaring + // one without getting the other is impossible — that's the invariant. + DLQ func(D, StageContext) (consumer.Controller, error) +} + +// StageContext carries engine-produced values that controllers need at +// construction time but the host does not own: the assembled topic +// registry (for publishing to downstream stages), the stage's own topic +// key, and its consumer group. +type StageContext struct { + // Registry is the fully assembled TopicRegistry. Controllers use it to + // look up topic names and queue backends for publishing downstream. + Registry consumer.TopicRegistry + + // TopicKey is this stage's logical topic key. + TopicKey consumer.TopicKey + + // ConsumerGroup is this stage's consumer group name. + ConsumerGroup string +} + +// PublishOnlyTopic declares a topic the service publishes to but does not +// consume. The engine registers it in the TopicRegistry so controllers +// can publish to it, but creates no subscription or controller. +type PublishOnlyTopic struct { + // Key is the logical topic key. + Key consumer.TopicKey + + // Name is the physical topic name. + Name string +} + +// options holds the resolved configuration for a Construct call. +type options struct { + topicNames map[consumer.TopicKey]string + classifiers []errs.Classifier + publishOnly []PublishOnlyTopic + extraComponents []lifecycle.Component +} + +// Option configures a Construct call. +type Option func(*options) + +// TopicNames provides a mapping from logical topic keys to physical topic +// names. Keys not present in the map fall back to the Stage.Name default. +func TopicNames(m map[consumer.TopicKey]string) Option { + return func(o *options) { o.topicNames = m } +} + +// Classifiers sets the error classifiers for the primary consumer's +// ErrorProcessor. DLQ consumers always use AlwaysRetryableProcessor. +func Classifiers(c ...errs.Classifier) Option { + return func(o *options) { o.classifiers = c } +} + +// PublishOnly adds topics the service publishes to but does not consume. +func PublishOnly(topics ...PublishOnlyTopic) Option { + return func(o *options) { o.publishOnly = append(o.publishOnly, topics...) } +} + +// ExtraComponents adds lifecycle components that are started before +// consumers and stopped after them. +func ExtraComponents(c ...lifecycle.Component) Option { + return func(o *options) { o.extraComponents = append(o.extraComponents, c...) } +} + +// dlqTopicKey returns the DLQ topic key for a primary stage key. +// Matches the convention in submitqueue/orchestrator/controller/dlq.TopicKey. +const dlqTopicSuffix = "_dlq" + +func dlqTopicKey(primary consumer.TopicKey) consumer.TopicKey { + return consumer.TopicKey(string(primary) + dlqTopicSuffix) +} + +// Construct is the single assembly function for a queue-driven service. +// It builds the topic registry, creates primary and DLQ consumers, +// eagerly constructs all controllers, and returns a lifecycle.Component +// that starts and stops everything in the correct order. +// +// The returned Component starts in this order: +// 1. Extra components (infrastructure) +// 2. Primary consumer (work-accepting) +// 3. DLQ consumer (reconciliation) +// +// Stop reverses the order: DLQ consumer drains first, then primary, then +// infrastructure. +func Construct[D any]( + logger *zap.SugaredLogger, + scope tally.Scope, + queue extqueue.Queue, + subscriberName string, + deps D, + stages []Stage[D], + opts ...Option, +) (lifecycle.Component, error) { + if len(stages) == 0 { + return nil, fmt.Errorf("pipeline: at least one stage is required") + } + + o := &options{} + for _, opt := range opts { + opt(o) + } + + // Build topic configs for the registry. + configs, err := buildTopicConfigs(queue, subscriberName, stages, o) + if err != nil { + return nil, err + } + + registry, err := consumer.NewTopicRegistry(configs) + if err != nil { + return nil, fmt.Errorf("pipeline: failed to create topic registry: %w", err) + } + + // Create the primary consumer with user-provided classifiers. + primaryProcessor := errs.NewClassifierProcessor(o.classifiers...) + primary := consumer.New(logger, scope, registry, primaryProcessor) + + // Create the DLQ consumer with always-retryable processor. + dlq := consumer.New(logger, scope, registry, errs.AlwaysRetryableProcessor) + + hasDLQ := false + + // Eagerly construct and register all controllers. + for _, s := range stages { + sc := StageContext{ + Registry: registry, + TopicKey: s.Key, + ConsumerGroup: s.ConsumerGroup, + } + + ctl, err := s.New(deps, sc) + if err != nil { + return nil, fmt.Errorf("pipeline: stage %s: failed to create controller: %w", s.Key, err) + } + if err := primary.Register(ctl); err != nil { + return nil, fmt.Errorf("pipeline: stage %s: failed to register controller: %w", s.Key, err) + } + + if s.DLQ != nil { + dlqSC := StageContext{ + Registry: registry, + TopicKey: dlqTopicKey(s.Key), + ConsumerGroup: s.ConsumerGroup + "-dlq", + } + rec, err := s.DLQ(deps, dlqSC) + if err != nil { + return nil, fmt.Errorf("pipeline: stage %s dlq: failed to create controller: %w", s.Key, err) + } + if err := dlq.Register(rec); err != nil { + return nil, fmt.Errorf("pipeline: stage %s dlq: failed to register controller: %w", s.Key, err) + } + hasDLQ = true + } + } + + // Compose the lifecycle group. + members := make([]lifecycle.Component, 0, len(o.extraComponents)+2) + members = append(members, o.extraComponents...) + members = append(members, &consumerComponent{name: "primary", c: primary}) + if hasDLQ { + members = append(members, &consumerComponent{name: "dlq", c: dlq}) + } + + return lifecycle.NewGroup(members...), nil +} + +// buildTopicConfigs constructs the []consumer.TopicConfig from stages and options. +func buildTopicConfigs[D any]( + queue extqueue.Queue, + subscriberName string, + stages []Stage[D], + o *options, +) ([]consumer.TopicConfig, error) { + // Pre-size: each stage gets a primary config + optional DLQ config, + // plus publish-only topics. + configs := make([]consumer.TopicConfig, 0, 2*len(stages)+len(o.publishOnly)) + + for _, s := range stages { + topicName := resolveTopicName(s.Key, s.Name, o.topicNames) + + configs = append(configs, consumer.TopicConfig{ + Key: s.Key, + Name: topicName, + Queue: queue, + Subscription: extqueue.DefaultSubscriptionConfig( + subscriberName, s.ConsumerGroup, + ), + }) + + if s.DLQ != nil { + configs = append(configs, consumer.TopicConfig{ + Key: dlqTopicKey(s.Key), + Name: topicName + dlqTopicSuffix, + Queue: queue, + Subscription: extqueue.DLQSubscriptionConfig( + subscriberName, s.ConsumerGroup+"-dlq", + ), + }) + } + } + + for _, p := range o.publishOnly { + topicName := resolveTopicName(p.Key, p.Name, o.topicNames) + configs = append(configs, consumer.TopicConfig{ + Key: p.Key, + Name: topicName, + Queue: queue, + }) + } + + return configs, nil +} + +// resolveTopicName returns the override name if present, otherwise the default. +func resolveTopicName(key consumer.TopicKey, defaultName string, overrides map[consumer.TopicKey]string) string { + if overrides != nil { + if name, ok := overrides[key]; ok { + return name + } + } + return defaultName +} + +// consumerComponent adapts consumer.Consumer to lifecycle.Component. +// Consumer.Stop takes a timeoutMs int64; Component.Stop takes a context. +// We derive the timeout from the context's deadline if set, defaulting to 30s. +type consumerComponent struct { + name string + c consumer.Consumer +} + +func (a *consumerComponent) Start(ctx context.Context) error { + return a.c.Start(ctx) +} + +func (a *consumerComponent) Stop(ctx context.Context) error { + const defaultStopTimeoutMs = 30000 + timeoutMs := int64(defaultStopTimeoutMs) + if deadline, ok := ctx.Deadline(); ok { + remaining := deadline.Sub(timeNow()) + if remaining > 0 { + timeoutMs = remaining.Milliseconds() + } else { + timeoutMs = 0 + } + } + return a.c.Stop(timeoutMs) +} diff --git a/platform/pipeline/pipeline_test.go b/platform/pipeline/pipeline_test.go new file mode 100644 index 00000000..a5412ce7 --- /dev/null +++ b/platform/pipeline/pipeline_test.go @@ -0,0 +1,397 @@ +// Copyright (c) 2025 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 pipeline + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/consumer" + extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" + mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +// testDeps is a minimal Deps type for testing. +type testDeps struct { + logger *zap.SugaredLogger +} + +// fakeController satisfies consumer.Controller for testing. +type fakeController struct { + key consumer.TopicKey + group string +} + +func (f *fakeController) Process(_ context.Context, _ consumer.Delivery) error { return nil } +func (f *fakeController) Name() string { return string(f.key) } +func (f *fakeController) TopicKey() consumer.TopicKey { return f.key } +func (f *fakeController) ConsumerGroup() string { return f.group } + +func newTestLogger() *zap.SugaredLogger { + l, _ := zap.NewDevelopment() + return l.Sugar() +} + +func TestConstruct_SingleStage_NoDLQ(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_WithDLQ(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_MultipleStages(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + { + Key: "validate", + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_EmptyStages_Error(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + + deps := testDeps{logger: newTestLogger()} + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, nil) + require.Error(t, err) + assert.Contains(t, err.Error(), "at least one stage is required") +} + +func TestConstruct_ControllerCreationFailure(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return nil, fmt.Errorf("missing dependency") + }, + }, + } + + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.Error(t, err) + assert.Contains(t, err.Error(), "stage start") + assert.Contains(t, err.Error(), "missing dependency") +} + +func TestConstruct_DLQControllerCreationFailure(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return nil, fmt.Errorf("dlq dependency missing") + }, + }, + } + + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.Error(t, err) + assert.Contains(t, err.Error(), "stage start dlq") + assert.Contains(t, err.Error(), "dlq dependency missing") +} + +func TestConstruct_WithPublishOnly(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages, + PublishOnly( + PublishOnlyTopic{Key: "log", Name: "log"}, + PublishOnlyTopic{Key: "merge-request", Name: "merge-request"}, + ), + ) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_WithTopicNameOverrides(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + comp, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages, + TopicNames(map[consumer.TopicKey]string{ + "start": "custom-start-topic", + }), + ) + require.NoError(t, err) + assert.NotNil(t, comp) +} + +func TestConstruct_StageContext_Populated(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + q.EXPECT().Subscriber().Return(mqmock.NewMockSubscriber(ctrl)).AnyTimes() + q.EXPECT().Publisher().Return(mqmock.NewMockPublisher(ctrl)).AnyTimes() + + deps := testDeps{logger: newTestLogger()} + + var primarySC, dlqSC StageContext + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { + primarySC = sc + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { + dlqSC = sc + return &fakeController{key: sc.TopicKey, group: sc.ConsumerGroup}, nil + }, + }, + } + + _, err := Construct(deps.logger, tally.NoopScope, q, "test-sub", deps, stages) + require.NoError(t, err) + + // Primary StageContext should have the stage's own key and group. + assert.Equal(t, consumer.TopicKey("start"), primarySC.TopicKey) + assert.Equal(t, "orchestrator-start", primarySC.ConsumerGroup) + + // DLQ StageContext should have the derived DLQ key and group. + assert.Equal(t, consumer.TopicKey("start_dlq"), dlqSC.TopicKey) + assert.Equal(t, "orchestrator-start-dlq", dlqSC.ConsumerGroup) + + // Both should share the same registry. + assert.Equal(t, primarySC.Registry, dlqSC.Registry) +} + +func TestResolveTopicName(t *testing.T) { + tests := []struct { + name string + key consumer.TopicKey + defaultN string + overrides map[consumer.TopicKey]string + want string + }{ + { + name: "no overrides", + key: "start", + defaultN: "start", + want: "start", + }, + { + name: "nil overrides", + key: "start", + defaultN: "start", + overrides: nil, + want: "start", + }, + { + name: "key not in overrides", + key: "start", + defaultN: "start", + overrides: map[consumer.TopicKey]string{"other": "other-name"}, + want: "start", + }, + { + name: "key in overrides", + key: "start", + defaultN: "start", + overrides: map[consumer.TopicKey]string{"start": "custom-start"}, + want: "custom-start", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveTopicName(tt.key, tt.defaultN, tt.overrides) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestDLQTopicKey(t *testing.T) { + assert.Equal(t, consumer.TopicKey("start_dlq"), dlqTopicKey("start")) + assert.Equal(t, consumer.TopicKey("validate_dlq"), dlqTopicKey("validate")) +} + +func TestBuildTopicConfigs(t *testing.T) { + ctrl := gomock.NewController(t) + q := mqmock.NewMockQueue(ctrl) + + stages := []Stage[testDeps]{ + { + Key: "start", + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil }, + DLQ: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil }, + }, + { + Key: "validate", + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d testDeps, sc StageContext) (consumer.Controller, error) { return nil, nil }, + // No DLQ for this stage. + }, + } + + o := &options{ + publishOnly: []PublishOnlyTopic{ + {Key: "log", Name: "log"}, + }, + } + + configs, err := buildTopicConfigs(q, "test-sub", stages, o) + require.NoError(t, err) + + // Expected: start (primary + DLQ) + validate (primary only) + log (publish-only) = 4 + assert.Len(t, configs, 4) + + // Verify primary stage config. + assert.Equal(t, consumer.TopicKey("start"), configs[0].Key) + assert.Equal(t, "start", configs[0].Name) + assert.Equal(t, "orchestrator-start", configs[0].Subscription.ConsumerGroup) + + // Verify DLQ config derived from primary. + assert.Equal(t, consumer.TopicKey("start_dlq"), configs[1].Key) + assert.Equal(t, "start_dlq", configs[1].Name) + assert.Equal(t, "orchestrator-start-dlq", configs[1].Subscription.ConsumerGroup) + + // Verify DLQ subscription has DLQ disabled (no cascade). + expected := extqueue.DLQSubscriptionConfig("test-sub", "orchestrator-start-dlq") + assert.Equal(t, expected.DLQ.Enabled, configs[1].Subscription.DLQ.Enabled) + assert.Equal(t, expected.Retry.MaxAttempts, configs[1].Subscription.Retry.MaxAttempts) + + // Verify validate stage (no DLQ). + assert.Equal(t, consumer.TopicKey("validate"), configs[2].Key) + + // Verify publish-only topic. + assert.Equal(t, consumer.TopicKey("log"), configs[3].Key) + assert.Equal(t, "log", configs[3].Name) + assert.Equal(t, "", configs[3].Subscription.ConsumerGroup) +} diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index 3bd852c8..1c835f83 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -7,23 +7,21 @@ exports_files( go_library( name = "orchestrator_lib", - srcs = ["main.go"], + srcs = [ + "main.go", + "profiles.go", + ], importpath = "github.com/uber/submitqueue/service/submitqueue/orchestrator/server", visibility = ["//visibility:private"], deps = [ - "//api/runway/messagequeue:go_default_library", "//api/submitqueue/orchestrator/protopb:go_default_library", - "//platform/consumer:go_default_library", - "//platform/errs:go_default_library", "//platform/errs/generic:go_default_library", "//platform/errs/mysql:go_default_library", - "//platform/extension/counter:go_default_library", "//platform/extension/counter/mysql:go_default_library", - "//platform/extension/messagequeue:go_default_library", "//platform/extension/messagequeue/mysql:go_default_library", "//platform/http:go_default_library", + "//platform/pipeline:go_default_library", "//submitqueue/core/changeset:go_default_library", - "//submitqueue/core/topickey:go_default_library", "//submitqueue/entity:go_default_library", "//submitqueue/extension/buildrunner:go_default_library", "//submitqueue/extension/buildrunner/fake:go_default_library", @@ -41,23 +39,9 @@ go_library( "//submitqueue/extension/scorer/composite:go_default_library", "//submitqueue/extension/scorer/fake:go_default_library", "//submitqueue/extension/scorer/heuristic:go_default_library", - "//submitqueue/extension/storage:go_default_library", "//submitqueue/extension/storage/mysql:go_default_library", "//submitqueue/extension/validator/fake:go_default_library", - "//submitqueue/orchestrator/controller:go_default_library", - "//submitqueue/orchestrator/controller/batch:go_default_library", - "//submitqueue/orchestrator/controller/build:go_default_library", - "//submitqueue/orchestrator/controller/buildsignal:go_default_library", - "//submitqueue/orchestrator/controller/cancel:go_default_library", - "//submitqueue/orchestrator/controller/conclude:go_default_library", - "//submitqueue/orchestrator/controller/dlq:go_default_library", - "//submitqueue/orchestrator/controller/merge:go_default_library", - "//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library", - "//submitqueue/orchestrator/controller/mergesignal:go_default_library", - "//submitqueue/orchestrator/controller/score:go_default_library", - "//submitqueue/orchestrator/controller/speculate:go_default_library", - "//submitqueue/orchestrator/controller/start:go_default_library", - "//submitqueue/orchestrator/controller/validate:go_default_library", + "//submitqueue/orchestrator:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_golang_google_grpc//:go_default_library", diff --git a/service/submitqueue/orchestrator/server/main.go b/service/submitqueue/orchestrator/server/main.go index 9c3f367b..0b2b21a0 100644 --- a/service/submitqueue/orchestrator/server/main.go +++ b/service/submitqueue/orchestrator/server/main.go @@ -31,53 +31,22 @@ import ( "golang.org/x/oauth2" "github.com/uber-go/tally" - runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" pb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb" - "github.com/uber/submitqueue/platform/consumer" - "github.com/uber/submitqueue/platform/errs" genericerrs "github.com/uber/submitqueue/platform/errs/generic" mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql" - "github.com/uber/submitqueue/platform/extension/counter" mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql" - extqueue "github.com/uber/submitqueue/platform/extension/messagequeue" queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/platform/http" + "github.com/uber/submitqueue/platform/pipeline" "github.com/uber/submitqueue/submitqueue/core/changeset" - "github.com/uber/submitqueue/submitqueue/core/topickey" - "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/buildrunner" - buildfake "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" cpfake "github.com/uber/submitqueue/submitqueue/extension/changeprovider/fake" githubprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/github" phabprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/phabricator" routingprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/routing" - "github.com/uber/submitqueue/submitqueue/extension/conflict" - "github.com/uber/submitqueue/submitqueue/extension/conflict/all" - conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake" - "github.com/uber/submitqueue/submitqueue/extension/conflict/fileoverlap" - "github.com/uber/submitqueue/submitqueue/extension/conflict/none" - "github.com/uber/submitqueue/submitqueue/extension/scorer" - "github.com/uber/submitqueue/submitqueue/extension/scorer/composite" - scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake" - "github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic" - "github.com/uber/submitqueue/submitqueue/extension/storage" mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/batch" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/build" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/conclude" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/score" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/start" - "github.com/uber/submitqueue/submitqueue/orchestrator/controller/validate" + "github.com/uber/submitqueue/submitqueue/orchestrator" "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/reflection" @@ -86,12 +55,12 @@ import ( // OrchestratorServer wraps the controller and implements the gRPC service interface type OrchestratorServer struct { pb.UnimplementedSubmitQueueOrchestratorServer - controller *controller.PingController + controllers orchestrator.Controllers } // Ping delegates to the controller func (s *OrchestratorServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error) { - return s.controller.Ping(ctx, req) + return s.controllers.Ping.Ping(ctx, req) } func main() { @@ -199,94 +168,70 @@ func run() error { logger.Info("initialized queue", zap.String("dsn", queueDSN)) - // Create topic registry + // Subscriber name for consumer group identity subscriberName := os.Getenv("HOSTNAME") if subscriberName == "" { subscriberName = fmt.Sprintf("orchestrator-%d", time.Now().Unix()) } - registry, err := newTopicRegistry(mysqlQueue, subscriberName) + // Build per-queue extension profiles (host-private). Each queue resolves + // to its own set of extension implementations (scorer, conflict analyzer, …), + // falling back to a baseline profile for queues without an explicit entry. + profiles, err := newProfiles(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore())) if err != nil { - return fmt.Errorf("failed to create topic registry: %w", err) - } - - // Two consumers share the topic registry but apply different error - // classification policies. The primary consumer runs the standard - // per-node classifier walk. The DLQ consumer uses the AlwaysRetryableProcessor - // so every non-nil error from a DLQ controller is forced retryable — - // reconciliation must redeliver on any failure because the DLQ - // subscriptions are final destinations (there is no further DLQ). - primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry, - errs.NewClassifierProcessor( + return fmt.Errorf("failed to build profiles: %w", err) + } + + // Populate the orchestrator's Deps — the library's public API. Factory + // fields are thin adapters that cross the host/library boundary via the + // existing Factory interfaces. + deps := orchestrator.Deps{ + Logger: logger.Sugar(), + Scope: scope, + Storage: store, + Counter: cnt, + BuildRunner: profiles.BuildRunnerFactory(), + ChangeProvider: profiles.ChangeProviderFactory(), + Scorer: profiles.ScorerFactory(), + Analyzer: profiles.AnalyzerFactory(), + Validator: validatorfake.NewFactory(), + } + + // Assemble the pipeline: one call builds the topic registry, creates + // primary and DLQ consumers, eagerly constructs all controllers, and + // returns a single lifecycle.Component the host drives with Start/Stop. + pl, err := pipeline.Construct( + logger.Sugar(), + scope, + mysqlQueue, + subscriberName, + deps, + orchestrator.Stages, + pipeline.PublishOnly(orchestrator.PublishOnlyTopics...), + pipeline.Classifiers( genericerrs.Classifier, - // Storage (submitqueue/extension/storage/mysql) and queue (platform/extension/messagequeue/mysql) - // both run on the same MySQL driver, so a single classifier covers - // errors surfaced from either backend. + // Storage (submitqueue/extension/storage/mysql) and queue + // (platform/extension/messagequeue/mysql) both run on the same + // MySQL driver, so a single classifier covers errors surfaced + // from either backend. mysqlerrs.Classifier, ), ) - dlqConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer-dlq"), registry, - errs.AlwaysRetryableProcessor, - ) - - // Build the per-queue extension registry: each queue resolves to its own - // set of extension implementations (scorer, conflict analyzer, …), falling - // back to a baseline profile for queues without an explicit entry. This is - // the single place queue topology is known; the extension packages stay - // queue-agnostic. - queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore())) if err != nil { - return fmt.Errorf("failed to build queue registry: %w", err) + return fmt.Errorf("failed to construct pipeline: %w", err) } - // Per-extension factories all resolve against the registry by queue name. - cpf := changeProviderFactory{queues} - brf := buildRunnerFactory{queues} - scf := scorerFactory{queues} - cof := analyzerFactory{queues} - - // Register controllers - primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, cnt, store) - if err != nil { - return err + // Start the pipeline (extra components → primary consumer → DLQ consumer). + if err := pl.Start(ctx); err != nil { + return fmt.Errorf("failed to start pipeline: %w", err) } - dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, registry, store) - if err != nil { - return err - } - - logger.Info("controllers registered", zap.Int("primary", primaryCount), zap.Int("dlq", dlqCount)) - - // Start consumers. DLQ first because Start begins processing - // messages immediately; if the second (primary) consumer fails to - // start, the half we already started is the DLQ side, whose work - // is idempotent reconciliation and is safe to interrupt mid-flight - // for rollback. - if err := dlqConsumer.Start(ctx); err != nil { - // The error can also be a result of a context cancellation due to SIGINT or SIGTERM. - // This is expected, just propagate it. - return fmt.Errorf("failed to start dlq consumer: %w", err) - } - if err := primaryConsumer.Start(ctx); err != nil { - // Best-effort: stop the dlq consumer we just started so the - // caller does not need to know which half failed. Aggregate both - // errors with errors.Join so the operator sees the original cause. - stopErr := dlqConsumer.Stop(30000) - return errors.Join( - fmt.Errorf("failed to start primary consumer: %w", err), - stopErr, - ) - } - logger.Info("consumers started") - - // Create gRPC server + logger.Info("pipeline started") + + // Create gRPC server and wire RPC controllers grpcServer := grpc.NewServer() - // Create ping controller and wrap it for gRPC - pingController := controller.NewPingController(logger, scope) - orchestratorServer := &OrchestratorServer{ - controller: pingController, - } + ctls := orchestrator.NewControllers(deps) + orchestratorServer := &OrchestratorServer{controllers: ctls} pb.RegisterSubmitQueueOrchestratorServer(grpcServer, orchestratorServer) // Register reflection service for debugging with grpcurl @@ -312,21 +257,15 @@ func run() error { }() // Wait for interrupt signal or server critical error - // If interruption is signaled, gracefully stop the server - // If server exits with an error, cancel the context to signal cancellation to the queue consumers - // After this, stop consumers - // If an error happens during shutdown, return the actual error, not the context cancellation error var serverErr error select { case <-ctx.Done(): fmt.Println("Shutting down orchestrator server due to interruption signal...") // Set the error to the context cancellation error to be surfaced as a desired exit code by the main function - // to indicate that the server was stopped as intended - // It may be overridden by the server error if any err = ctx.Err() - // stop GRPC server and wait for it to exit + // Stop GRPC server and wait for it to exit grpcServer.GracefulStop() serverErr = <-serverErrCh case serverErr = <-serverErrCh: @@ -340,415 +279,25 @@ func run() error { serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr) } - // Stop consumers with 30s timeout in reverse start order: primary - // first, then DLQ. The primary pipeline writes the state that DLQ - // reconciliation reads, so draining primary first means in-flight - // DLQ reconciliation finishes against a settled primary rather than - // racing its shutdown. - primaryStopErr := primaryConsumer.Stop(30000) - dlqStopErr := dlqConsumer.Stop(30000) - errStop := errors.Join(primaryStopErr, dlqStopErr) - if errStop != nil { - errStop = fmt.Errorf("failed to stop consumers: %w", errStop) + // Stop the pipeline (DLQ consumer → primary consumer → extra components, + // reverse of start order). Use a fresh context with a 30s timeout so + // shutdown proceeds even after the parent context is cancelled. + stopCtx, stopCancel := context.WithTimeout(context.Background(), 30*time.Second) + defer stopCancel() + plStopErr := pl.Stop(stopCtx) + if plStopErr != nil { + plStopErr = fmt.Errorf("failed to stop pipeline: %w", plStopErr) } - if errStop != nil || serverErr != nil { + if plStopErr != nil || serverErr != nil { // Override context cancellation error with the shutdown error - err = errors.Join(errStop, serverErr) + err = errors.Join(plStopErr, serverErr) } // Return the error to be surfaced as a desired exit code by the main function return err } -// newTopicRegistry builds the TopicRegistry with all topic and subscription configs. -func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { - // primaryTopics enumerates the {key, name, group-suffix} for every primary - // pipeline topic. The DLQ topic for each is derived by appending "_dlq" to - // both the topic name and the consumer group; the topic-key suffix is - // owned by the dlq package (dlq.TopicKey). - type topicSpec struct { - key consumer.TopicKey - name string - groupSuffix string - } - primaryTopics := []topicSpec{ - {topickey.TopicKeyStart, "start", "orchestrator-start"}, - {topickey.TopicKeyCancel, "cancel", "orchestrator-cancel"}, - {topickey.TopicKeyValidate, "validate", "orchestrator-validate"}, - {runwaymq.TopicKeyMergeConflictCheckSignal, "merge-conflict-check-signal", "orchestrator-mergeconflictsignal"}, - {topickey.TopicKeyBatch, "batch", "orchestrator-batch"}, - {topickey.TopicKeyScore, "score", "orchestrator-score"}, - {topickey.TopicKeySpeculate, "speculate", "orchestrator-speculate"}, - {topickey.TopicKeyBuild, "build", "orchestrator-build"}, - {topickey.TopicKeyBuildSignal, "buildsignal", "orchestrator-buildsignal"}, - {topickey.TopicKeyMerge, "submitqueue-merge", "orchestrator-merge"}, - {runwaymq.TopicKeyMergeSignal, "merge-signal", "orchestrator-mergesignal"}, - {topickey.TopicKeyConclude, "conclude", "orchestrator-conclude"}, - } - - configs := make([]consumer.TopicConfig, 0, 2*len(primaryTopics)) - for _, t := range primaryTopics { - configs = append(configs, consumer.TopicConfig{ - Key: t.key, - Name: t.name, - Queue: q, - Subscription: extqueue.DefaultSubscriptionConfig( - subscriberName, t.groupSuffix, - ), - }) - // DLQ subscription for the same primary stage. DLQSubscriptionConfig - // disables the subscription's own DLQ (no "_dlq_dlq" cascade) and sets - // an effectively unlimited retry budget to pair with the - // AlwaysRetryableProcessor wired into the DLQ consumer. - configs = append(configs, consumer.TopicConfig{ - Key: dlq.TopicKey(t.key), - Name: t.name + "_dlq", - Queue: q, - Subscription: extqueue.DLQSubscriptionConfig(subscriberName, t.groupSuffix+"-dlq"), - }) - } - - // Publish-only: the orchestrator emits request-log entries to the log topic. - // The gateway is the sole consumer and writer of request logs and public - // projections, so the orchestrator registers no consuming subscription. - configs = append(configs, consumer.TopicConfig{ - Key: topickey.TopicKeyLog, - Name: "log", - Queue: q, - }) - - // Publish-only: the orchestrator hands merge-conflict check requests to - // runway via the runway-owned merge-conflict-check queue. Runway is the - // sole consumer, so the orchestrator registers no consuming subscription - // (and no DLQ) here — the inbound result arrives on the separate - // merge-conflict-check-signal queue, which is a consumed primary topic - // above. - configs = append(configs, consumer.TopicConfig{ - Key: runwaymq.TopicKeyMergeConflictCheck, - Name: "merge-conflict-check", - Queue: q, - }) - - // Publish-only: the orchestrator hands merge requests to runway via the - // runway-owned merge queue. Runway is the sole consumer, so the - // orchestrator registers no consuming subscription (and no DLQ) here; the - // inbound result arrives on the separate merge-signal queue, which is a - // consumed primary topic above. - configs = append(configs, consumer.TopicConfig{ - Key: runwaymq.TopicKeyMerge, - Name: "runway-merge", - Queue: q, - }) - - return consumer.NewTopicRegistry(configs) -} - -// registerPrimaryControllers creates all pipeline controllers and registers -// them with the primary consumer. Pipeline: -// -// request → validate ⇢ (runway) ⇢ mergeconflictsignal → batch → score → speculate → build → buildsignal ─┐ -// ↑ ↘ ↻ poll │ -// │ merge → conclude │ -// │ │ │ -// └────────┴───────────────────────┘ -// -// The merge-conflict check is asynchronous and crosses a service boundary: -// validate publishes the full check request to the runway-owned -// merge-conflict-check queue (⇢); runway performs the merge attempt and -// publishes the result to merge-conflict-check-signal, which mergeconflictsignal -// consumes before fanning the request out to batch. - -// TODO(wiring abstraction): queueExtensions + queueRegistry currently live here -// as example-local wiring. Evaluate promoting them into a defined abstraction in -// the submitqueue domain layer (e.g. submitqueue/core/...) — not `submitqueue/extension/*` -// and not `platform/*`, since the bundle names submitqueue-specific extensions. -// Do this only when a trigger lands: (1) a second consumer needs the same wiring -// (a real prod server, or an e2e harness building real per-queue profiles); -// (2) per-queue config becomes data-driven (build profiles from queueconfig.Store -// /queues.yaml instead of Go literals); or (3) the bundle grows lifecycle -// (Close/health/hot-reload). Until then, keep it local — extracting now adds -// indirection for one hardcoded consumer. See also queueconfig.Store, which holds -// the per-queue *data* half; a promoted Registry would build impl bundles from it. -// -// queueExtensions is the full set of extension implementations for a single -// queue. Grouping them per queue (rather than per extension) lets the wiring -// read as "for this queue, here are its scorer, analyzer, change provider, …", and lets -// a queue profile start from a baseline and override only what differs. -type queueExtensions struct { - changeProvider changeprovider.ChangeProvider - buildRunner buildrunner.BuildRunner - scorer scorer.Scorer - analyzer conflict.Analyzer -} - -// queueRegistry maps a queue name to its extensions, falling back to a default -// profile for queues without an explicit entry. It is the single place that -// knows the queue topology; the extension packages remain queue-agnostic. -type queueRegistry struct { - byQueue map[string]queueExtensions - def queueExtensions -} - -// get returns the extensions for the named queue, or the default profile. -func (r queueRegistry) get(queue string) queueExtensions { - if e, ok := r.byQueue[queue]; ok { - return e - } - return r.def -} - -// The per-extension factories below are thin adapters: each satisfies its -// extension's Factory contract by resolving the queue's profile from the -// registry. All routing logic lives here in the wiring layer. -type changeProviderFactory struct{ reg queueRegistry } - -func (f changeProviderFactory) For(cfg changeprovider.Config) (changeprovider.ChangeProvider, error) { - return f.reg.get(cfg.QueueName).changeProvider, nil -} - -type buildRunnerFactory struct{ reg queueRegistry } - -func (f buildRunnerFactory) For(cfg buildrunner.Config) (buildrunner.BuildRunner, error) { - return f.reg.get(cfg.QueueName).buildRunner, nil -} - -type scorerFactory struct{ reg queueRegistry } - -func (f scorerFactory) For(cfg scorer.Config) (scorer.Scorer, error) { - return f.reg.get(cfg.QueueName).scorer, nil -} - -type analyzerFactory struct{ reg queueRegistry } - -func (f analyzerFactory) For(cfg conflict.Config) (conflict.Analyzer, error) { - return f.reg.get(cfg.QueueName).analyzer, nil -} - -func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, cnt counter.Counter, store storage.Storage) (int, error) { - var count int - requestController := start.NewController( - logger, - scope, - store, - registry, - topickey.TopicKeyStart, - "orchestrator-start", - ) - if err := c.Register(requestController); err != nil { - return count, fmt.Errorf("failed to register request controller: %w", err) - } - count++ - - cancelController := cancel.NewController( - logger, - scope, - store, - registry, - topickey.TopicKeyCancel, - "orchestrator-cancel", - ) - if err := c.Register(cancelController); err != nil { - return count, fmt.Errorf("failed to register cancel controller: %w", err) - } - count++ - - validateController := validate.NewController( - logger, - scope, - store, - registry, - cpf, - validatorfake.NewFactory(), - runwaymq.TopicKeyMergeConflictCheck, - topickey.TopicKeyValidate, - "orchestrator-validate", - ) - if err := c.Register(validateController); err != nil { - return count, fmt.Errorf("failed to register validate controller: %w", err) - } - count++ - - mergeconflictsignalController := mergeconflictsignal.NewController( - logger, - scope, - store, - registry, - runwaymq.TopicKeyMergeConflictCheckSignal, - "orchestrator-mergeconflictsignal", - ) - if err := c.Register(mergeconflictsignalController); err != nil { - return count, fmt.Errorf("failed to register mergeconflictsignal controller: %w", err) - } - count++ - - batchController := batch.NewController( - logger, - scope, - registry, - cnt, - store, - cof, - topickey.TopicKeyBatch, - "orchestrator-batch", - ) - if err := c.Register(batchController); err != nil { - return count, fmt.Errorf("failed to register batch controller: %w", err) - } - count++ - - scoreController := score.NewController( - logger, - scope, - store, - scf, - registry, - topickey.TopicKeyScore, - "orchestrator-score", - ) - if err := c.Register(scoreController); err != nil { - return count, fmt.Errorf("failed to register score controller: %w", err) - } - count++ - - speculateController := speculate.NewController( - logger, - scope, - store, - registry, - topickey.TopicKeySpeculate, - "orchestrator-speculate", - ) - if err := c.Register(speculateController); err != nil { - return count, fmt.Errorf("failed to register speculate controller: %w", err) - } - count++ - - buildController := build.NewController( - logger, - scope, - store, - brf, - registry, - topickey.TopicKeyBuild, - "orchestrator-build", - ) - if err := c.Register(buildController); err != nil { - return count, fmt.Errorf("failed to register build controller: %w", err) - } - count++ - - buildsignalController := buildsignal.NewController( - logger, - scope, - store, - brf, - registry, - topickey.TopicKeyBuildSignal, - "orchestrator-buildsignal", - ) - if err := c.Register(buildsignalController); err != nil { - return count, fmt.Errorf("failed to register buildsignal controller: %w", err) - } - count++ - - mergeController := merge.NewController( - logger, - scope, - store, - registry, - runwaymq.TopicKeyMerge, - topickey.TopicKeyMerge, - "orchestrator-merge", - ) - if err := c.Register(mergeController); err != nil { - return count, fmt.Errorf("failed to register merge controller: %w", err) - } - count++ - - mergesignalController := mergesignal.NewController( - logger, - scope, - store, - registry, - runwaymq.TopicKeyMergeSignal, - "orchestrator-mergesignal", - ) - if err := c.Register(mergesignalController); err != nil { - return count, fmt.Errorf("failed to register mergesignal controller: %w", err) - } - count++ - - concludeController := conclude.NewController( - logger, - scope, - store, - registry, - topickey.TopicKeyConclude, - "orchestrator-conclude", - ) - if err := c.Register(concludeController); err != nil { - return count, fmt.Errorf("failed to register conclude controller: %w", err) - } - count++ - - return count, nil -} - -// registerDLQControllers creates one DLQ reconciler per primary stage and -// registers them with the DLQ consumer. Each reconciler drives the affected -// request or batch into a terminal Error/Failed state so the gateway stops -// reporting it as stuck-in-progress. -func registerDLQControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, store storage.Storage) (int, error) { - dlqScope := scope.SubScope("dlq") - dlqRegs := []struct { - name string - ctl consumer.Controller - }{ - {"start_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeLandRequestID, dlq.TopicKey(topickey.TopicKeyStart), "orchestrator-start-dlq")}, - {"cancel_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeCancelRequestID, dlq.TopicKey(topickey.TopicKeyCancel), "orchestrator-cancel-dlq")}, - {"validate_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyValidate), "orchestrator-validate-dlq")}, - {"mergeconflictsignal_dlq", dlq.NewDLQMergeConflictSignalController(logger, dlqScope, store, registry, dlq.TopicKey(runwaymq.TopicKeyMergeConflictCheckSignal), "orchestrator-mergeconflictsignal-dlq")}, - {"batch_dlq", dlq.NewDLQRequestController(logger, dlqScope, store, registry, dlq.DecodeRequestID, dlq.TopicKey(topickey.TopicKeyBatch), "orchestrator-batch-dlq")}, - {"score_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyScore), "orchestrator-score-dlq")}, - {"speculate_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeySpeculate), "orchestrator-speculate-dlq")}, - {"build_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyBuild), "orchestrator-build-dlq")}, - {"buildsignal_dlq", dlq.NewDLQBuildSignalController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyBuildSignal), "orchestrator-buildsignal-dlq")}, - {"merge_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyMerge), "orchestrator-merge-dlq")}, - {"mergesignal_dlq", dlq.NewDLQMergeSignalController(logger, dlqScope, store, registry, dlq.TopicKey(runwaymq.TopicKeyMergeSignal), "orchestrator-mergesignal-dlq")}, - {"conclude_dlq", dlq.NewDLQBatchController(logger, dlqScope, store, registry, dlq.TopicKey(topickey.TopicKeyConclude), "orchestrator-conclude-dlq")}, - } - var count int - for _, reg := range dlqRegs { - if err := c.Register(reg.ctl); err != nil { - return count, fmt.Errorf("failed to register %s controller: %w", reg.name, err) - } - count++ - } - - return count, nil -} - -// getEnv returns environment variable value or default if not set. -func getEnv(key, defaultVal string) string { - if val := os.Getenv(key); val != "" { - return val - } - return defaultVal -} - -// parseTimeout parses a duration from environment variable with fallback to default. -// Returns defaultVal if envVal is empty or cannot be parsed. -func parseTimeout(envVal string, defaultVal time.Duration) time.Duration { - if envVal == "" { - return defaultVal - } - if d, err := time.ParseDuration(envVal); err == nil { - return d - } - return defaultVal -} - // newChangeProvider creates a routing ChangeProvider containing GitHub and Phab ChangeProviders. // When neither GITHUB_TOKEN nor PHAB_API_TOKEN is set, falls back to the fake change provider. func newChangeProvider(logger *zap.Logger, scope tally.Scope) (changeprovider.ChangeProvider, error) { @@ -846,93 +395,22 @@ func newPhabChangeProvider(logger *zap.Logger, scope tally.Scope) (changeprovide }), nil } -// newQueueRegistry builds the per-queue extension profiles for the example. -// Edge integrations (change provider) and the build -// runner form a shared baseline; each per-queue profile starts from that -// baseline and overrides only the extensions that differ — here the scorer and -// conflict analyzer. Queues without an explicit profile fall back to the -// baseline. This is the one place queue topology lives; extension packages stay -// queue-agnostic. -func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (queueRegistry, error) { - cp, err := newChangeProvider(logger, scope) - if err != nil { - return queueRegistry{}, fmt.Errorf("failed to create change provider: %w", err) - } - - // batchLines buckets a batch by total lines changed across all its changes — - // larger batches are likelier to fail to land. - batchLines := func(_ context.Context, changes entity.BatchChanges) (int, error) { - return changes.TotalLinesChanged(), nil - } - - // Baseline profile: shared edge integrations + a fake build runner (every - // build succeeds unless a head URI carries a failure marker), plus permissive - // defaults for scorer and conflict. The build runner - // instance is shared by the build and buildsignal controllers (same - // profile, same instance) so a build's recorded outcome survives across - // their separate factory lookups. - // - // The scorer is wrapped by scorerfake so a change URI carrying - // "sq-fake=score-error" forces a scoring error end-to-end; it is a pure - // passthrough otherwise. The analyzer is wrapped by conflictfake with a nil - // predicate (passthrough) — swap the predicate (e.g. conflictfake.FailAlways) - // on a queue to exercise the analyzer error path, as e2e-conflict-error-queue - // below does. - base := queueExtensions{ - changeProvider: cp, - buildRunner: buildfake.New(resolver), - scorer: scorerfake.New(resolver, heuristic.New( - resolver, - []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.5}}, - batchLines, scope.SubScope("scorer.default"), - )), - // TODO: replace the delegate with a real analyzer (e.g. Tango target - // analysis). "all" serializes the queue conservatively. - analyzer: conflictfake.New(all.New(), nil), - } - - // test-queue: bucketed heuristic scorer; conservative (serialized) conflicts - // inherited from the baseline. - testQueue := base - testQueue.scorer = scorerfake.New(resolver, heuristic.New( - resolver, - []heuristic.Bucket{ - {Min: 0, Max: 1, Score: 0.95}, - {Min: 2, Max: 5, Score: 0.80}, - {Min: 6, Max: 20, Score: 0.60}, - {Min: 21, Max: 1<<31 - 1, Score: 0.40}, - }, - batchLines, scope.SubScope("scorer.test-queue"), - )) - - // e2e-test-queue: composite scorer; no conflicts (maximum parallelism). - e2eQueue := base - e2eQueue.analyzer = conflictfake.New(none.New(), nil) - e2eQueue.scorer = scorerfake.New(resolver, composite.New( - map[string]scorer.Scorer{ - "size": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.8}}, batchLines, scope), - "flat": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.6}}, batchLines, scope), - }, - composite.Avg, scope.SubScope("scorer.e2e-test-queue"), - )) - - // e2e-conflict-error-queue: every conflict analysis fails, exercising the - // analyzer error path. Scorer/edge integrations inherit the baseline. - conflictErrQueue := base - conflictErrQueue.analyzer = conflictfake.New(all.New(), conflictfake.FailAlways) - - // file-overlap-queue: a real analyzer that serializes only batches sharing - // a changed file, resolving each batch's files itself via the resolver. - fileOverlapQueue := base - fileOverlapQueue.analyzer = fileoverlap.New(resolver) - - return queueRegistry{ - def: base, - byQueue: map[string]queueExtensions{ - "test-queue": testQueue, - "e2e-test-queue": e2eQueue, - "e2e-conflict-error-queue": conflictErrQueue, - "file-overlap-queue": fileOverlapQueue, - }, - }, nil +// getEnv returns environment variable value or default if not set. +func getEnv(key, defaultVal string) string { + if val := os.Getenv(key); val != "" { + return val + } + return defaultVal +} + +// parseTimeout parses a duration from environment variable with fallback to default. +// Returns defaultVal if envVal is empty or cannot be parsed. +func parseTimeout(envVal string, defaultVal time.Duration) time.Duration { + if envVal == "" { + return defaultVal + } + if d, err := time.ParseDuration(envVal); err == nil { + return d + } + return defaultVal } diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go new file mode 100644 index 00000000..4ad46504 --- /dev/null +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -0,0 +1,214 @@ +// Copyright (c) 2025 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 main + +import ( + "context" + "fmt" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/submitqueue/core/changeset" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/buildrunner" + buildfake "github.com/uber/submitqueue/submitqueue/extension/buildrunner/fake" + "github.com/uber/submitqueue/submitqueue/extension/changeprovider" + "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/conflict/all" + conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake" + "github.com/uber/submitqueue/submitqueue/extension/conflict/fileoverlap" + "github.com/uber/submitqueue/submitqueue/extension/conflict/none" + "github.com/uber/submitqueue/submitqueue/extension/scorer" + "github.com/uber/submitqueue/submitqueue/extension/scorer/composite" + scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake" + "github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic" + "go.uber.org/zap" +) + +// Profile holds the per-queue extension implementations. Grouping them per +// queue (rather than per extension) lets the wiring read as "for this queue, +// here are its scorer, analyzer, change provider, …", and lets a queue +// profile start from a baseline and override only what differs. +type Profile struct { + // ChangeProvider resolves change metadata for requests in this queue. + ChangeProvider changeprovider.ChangeProvider + + // BuildRunner triggers and polls builds for batches in this queue. + BuildRunner buildrunner.BuildRunner + + // Scorer computes landing-probability scores for batches in this queue. + Scorer scorer.Scorer + + // Analyzer detects conflicts between concurrent batches in this queue. + Analyzer conflict.Analyzer +} + +// Profiles maps a queue name to its extension Profile, falling back to a +// default profile for queues without an explicit entry. This is the single +// place that knows the queue topology; the extension packages remain +// queue-agnostic. +type Profiles struct { + byQueue map[string]Profile + defaultProfile Profile +} + +// For returns the profile for the named queue, or the default. +func (p Profiles) For(queue string) Profile { + if prof, ok := p.byQueue[queue]; ok { + return prof + } + return p.defaultProfile +} + +// ChangeProviderFactory returns a changeprovider.Factory that resolves the +// ChangeProvider for each queue from the profile registry. +func (p Profiles) ChangeProviderFactory() changeprovider.Factory { + return changeProviderFunc(func(c changeprovider.Config) (changeprovider.ChangeProvider, error) { + return p.For(c.QueueName).ChangeProvider, nil + }) +} + +// BuildRunnerFactory returns a buildrunner.Factory that resolves the +// BuildRunner for each queue from the profile registry. +func (p Profiles) BuildRunnerFactory() buildrunner.Factory { + return buildRunnerFunc(func(c buildrunner.Config) (buildrunner.BuildRunner, error) { + return p.For(c.QueueName).BuildRunner, nil + }) +} + +// ScorerFactory returns a scorer.Factory that resolves the Scorer for each +// queue from the profile registry. +func (p Profiles) ScorerFactory() scorer.Factory { + return scorerFunc(func(c scorer.Config) (scorer.Scorer, error) { + return p.For(c.QueueName).Scorer, nil + }) +} + +// AnalyzerFactory returns a conflict.Factory that resolves the Analyzer for +// each queue from the profile registry. +func (p Profiles) AnalyzerFactory() conflict.Factory { + return analyzerFunc(func(c conflict.Config) (conflict.Analyzer, error) { + return p.For(c.QueueName).Analyzer, nil + }) +} + +// Thin func-type adapters — the http.HandlerFunc trick applied to each +// extension Factory interface. Each func type satisfies the Factory contract, +// letting Profiles cross the host/library boundary without dedicated structs. + +type changeProviderFunc func(changeprovider.Config) (changeprovider.ChangeProvider, error) + +func (f changeProviderFunc) For(c changeprovider.Config) (changeprovider.ChangeProvider, error) { + return f(c) +} + +type buildRunnerFunc func(buildrunner.Config) (buildrunner.BuildRunner, error) + +func (f buildRunnerFunc) For(c buildrunner.Config) (buildrunner.BuildRunner, error) { return f(c) } + +type scorerFunc func(scorer.Config) (scorer.Scorer, error) + +func (f scorerFunc) For(c scorer.Config) (scorer.Scorer, error) { return f(c) } + +type analyzerFunc func(conflict.Config) (conflict.Analyzer, error) + +func (f analyzerFunc) For(c conflict.Config) (conflict.Analyzer, error) { return f(c) } + +// newProfiles builds the per-queue extension profiles for the example. +// Edge integrations (change provider) and the build runner form a shared +// baseline; each per-queue profile starts from that baseline and overrides +// only the extensions that differ — here the scorer and conflict analyzer. +// Queues without an explicit profile fall back to the baseline. +func newProfiles(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (Profiles, error) { + cp, err := newChangeProvider(logger, scope) + if err != nil { + return Profiles{}, fmt.Errorf("failed to create change provider: %w", err) + } + + // batchLines buckets a batch by total lines changed across all its changes — + // larger batches are likelier to fail to land. + batchLines := func(_ context.Context, changes entity.BatchChanges) (int, error) { + return changes.TotalLinesChanged(), nil + } + + // Baseline profile: shared edge integrations + a fake build runner (every + // build succeeds unless a head URI carries a failure marker), plus permissive + // defaults for scorer and conflict. The build runner instance is shared by + // the build and buildsignal controllers (same profile, same instance) so a + // build's recorded outcome survives across their separate factory lookups. + // + // The scorer is wrapped by scorerfake so a change URI carrying + // "sq-fake=score-error" forces a scoring error end-to-end; it is a pure + // passthrough otherwise. The analyzer is wrapped by conflictfake with a nil + // predicate (passthrough) — swap the predicate (e.g. conflictfake.FailAlways) + // on a queue to exercise the analyzer error path, as e2e-conflict-error-queue + // below does. + base := Profile{ + ChangeProvider: cp, + BuildRunner: buildfake.New(resolver), + Scorer: scorerfake.New(resolver, heuristic.New( + resolver, + []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.5}}, + batchLines, scope.SubScope("scorer.default"), + )), + // TODO: replace the delegate with a real analyzer (e.g. Tango target + // analysis). "all" serializes the queue conservatively. + Analyzer: conflictfake.New(all.New(), nil), + } + + // test-queue: bucketed heuristic scorer; conservative (serialized) conflicts + // inherited from the baseline. + testQueue := base + testQueue.Scorer = scorerfake.New(resolver, heuristic.New( + resolver, + []heuristic.Bucket{ + {Min: 0, Max: 1, Score: 0.95}, + {Min: 2, Max: 5, Score: 0.80}, + {Min: 6, Max: 20, Score: 0.60}, + {Min: 21, Max: 1<<31 - 1, Score: 0.40}, + }, + batchLines, scope.SubScope("scorer.test-queue"), + )) + + // e2e-test-queue: composite scorer; no conflicts (maximum parallelism). + e2eQueue := base + e2eQueue.Analyzer = conflictfake.New(none.New(), nil) + e2eQueue.Scorer = scorerfake.New(resolver, composite.New( + map[string]scorer.Scorer{ + "size": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.8}}, batchLines, scope), + "flat": heuristic.New(resolver, []heuristic.Bucket{{Min: 0, Max: 1<<31 - 1, Score: 0.6}}, batchLines, scope), + }, + composite.Avg, scope.SubScope("scorer.e2e-test-queue"), + )) + + // e2e-conflict-error-queue: every conflict analysis fails, exercising the + // analyzer error path. Scorer/edge integrations inherit the baseline. + conflictErrQueue := base + conflictErrQueue.Analyzer = conflictfake.New(all.New(), conflictfake.FailAlways) + + // file-overlap-queue: a real analyzer that serializes only batches sharing + // a changed file, resolving each batch's files itself via the resolver. + fileOverlapQueue := base + fileOverlapQueue.Analyzer = fileoverlap.New(resolver) + + return Profiles{ + defaultProfile: base, + byQueue: map[string]Profile{ + "test-queue": testQueue, + "e2e-test-queue": e2eQueue, + "e2e-conflict-error-queue": conflictErrQueue, + "file-overlap-queue": fileOverlapQueue, + }, + }, nil +} diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel new file mode 100644 index 00000000..37baf78a --- /dev/null +++ b/submitqueue/orchestrator/BUILD.bazel @@ -0,0 +1,37 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["pipeline.go"], + importpath = "github.com/uber/submitqueue/submitqueue/orchestrator", + visibility = ["//visibility:public"], + deps = [ + "//api/runway/messagequeue:go_default_library", + "//platform/consumer:go_default_library", + "//platform/extension/counter:go_default_library", + "//platform/pipeline:go_default_library", + "//submitqueue/core/topickey:go_default_library", + "//submitqueue/extension/buildrunner:go_default_library", + "//submitqueue/extension/changeprovider:go_default_library", + "//submitqueue/extension/conflict:go_default_library", + "//submitqueue/extension/scorer:go_default_library", + "//submitqueue/extension/storage:go_default_library", + "//submitqueue/extension/validator:go_default_library", + "//submitqueue/orchestrator/controller:go_default_library", + "//submitqueue/orchestrator/controller/batch:go_default_library", + "//submitqueue/orchestrator/controller/build:go_default_library", + "//submitqueue/orchestrator/controller/buildsignal:go_default_library", + "//submitqueue/orchestrator/controller/cancel:go_default_library", + "//submitqueue/orchestrator/controller/conclude:go_default_library", + "//submitqueue/orchestrator/controller/dlq:go_default_library", + "//submitqueue/orchestrator/controller/merge:go_default_library", + "//submitqueue/orchestrator/controller/mergeconflictsignal:go_default_library", + "//submitqueue/orchestrator/controller/mergesignal:go_default_library", + "//submitqueue/orchestrator/controller/score:go_default_library", + "//submitqueue/orchestrator/controller/speculate:go_default_library", + "//submitqueue/orchestrator/controller/start:go_default_library", + "//submitqueue/orchestrator/controller/validate:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + "@org_uber_go_zap//:go_default_library", + ], +) diff --git a/submitqueue/orchestrator/pipeline.go b/submitqueue/orchestrator/pipeline.go new file mode 100644 index 00000000..1d649f8f --- /dev/null +++ b/submitqueue/orchestrator/pipeline.go @@ -0,0 +1,254 @@ +// Copyright (c) 2025 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 orchestrator declares the SubmitQueue orchestrator's pipeline +// topology, extension seams, and controller set. The host (main.go) fills +// Deps and passes Stages to pipeline.Construct; no assembly logic lives here. +package orchestrator + +import ( + "github.com/uber-go/tally" + runwaymq "github.com/uber/submitqueue/api/runway/messagequeue" + "github.com/uber/submitqueue/platform/consumer" + "github.com/uber/submitqueue/platform/extension/counter" + "github.com/uber/submitqueue/platform/pipeline" + "github.com/uber/submitqueue/submitqueue/core/topickey" + "github.com/uber/submitqueue/submitqueue/extension/buildrunner" + "github.com/uber/submitqueue/submitqueue/extension/changeprovider" + "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/scorer" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "github.com/uber/submitqueue/submitqueue/extension/validator" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/batch" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/build" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/buildsignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/cancel" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/conclude" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/dlq" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/merge" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergeconflictsignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/mergesignal" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/score" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/start" + "github.com/uber/submitqueue/submitqueue/orchestrator/controller/validate" + "go.uber.org/zap" +) + +// Deps is the full set of dependencies the orchestrator pipeline needs. +// This struct IS the service's public API toward deployers: fill every +// field, pass it and Stages to pipeline.Construct, and you get a running +// orchestrator pipeline. +type Deps struct { + // Logger is the structured logger for all controllers. + Logger *zap.SugaredLogger + + // Scope is the metrics scope for all controllers. + Scope tally.Scope + + // Storage provides request, batch, and change stores. + Storage storage.Storage + + // Counter provides distributed batch counters. + Counter counter.Counter + + // BuildRunner resolves the build runner for each queue. + BuildRunner buildrunner.Factory + + // ChangeProvider resolves the change provider for each queue. + ChangeProvider changeprovider.Factory + + // Scorer resolves the scorer for each queue. + Scorer scorer.Factory + + // Analyzer resolves the conflict analyzer for each queue. + Analyzer conflict.Factory + + // Validator resolves the validator for each queue. + Validator validator.Factory +} + +// Stages is the orchestrator's pipeline topology as a typed table. +// Adding a stage = adding one row. Nothing else, anywhere. +// +// Pipeline: +// +// start → cancel → validate ⇢ (runway) ⇢ mergeconflictsignal → batch → score → speculate → build → buildsignal ─┐ +// ↑ ↘ ↻ poll │ +// │ merge → conclude │ +// │ │ │ +// └────────┴───────────────────────┘ +var Stages = []pipeline.Stage[Deps]{ + { + Key: topickey.TopicKeyStart, + Name: "start", + ConsumerGroup: "orchestrator-start", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return start.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeLandRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyCancel, + Name: "cancel", + ConsumerGroup: "orchestrator-cancel", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return cancel.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeCancelRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyValidate, + Name: "validate", + ConsumerGroup: "orchestrator-validate", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return validate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, d.ChangeProvider, d.Validator, runwaymq.TopicKeyMergeConflictCheck, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: runwaymq.TopicKeyMergeConflictCheckSignal, + Name: "merge-conflict-check-signal", + ConsumerGroup: "orchestrator-mergeconflictsignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return mergeconflictsignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQMergeConflictSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBatch, + Name: "batch", + ConsumerGroup: "orchestrator-batch", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return batch.NewController(d.Logger, d.Scope, sc.Registry, d.Counter, d.Storage, d.Analyzer, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQRequestController(d.Logger, d.Scope, d.Storage, sc.Registry, dlq.DecodeRequestID, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyScore, + Name: "score", + ConsumerGroup: "orchestrator-score", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return score.NewController(d.Logger, d.Scope, d.Storage, d.Scorer, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeySpeculate, + Name: "speculate", + ConsumerGroup: "orchestrator-speculate", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return speculate.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBuild, + Name: "build", + ConsumerGroup: "orchestrator-build", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return build.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyBuildSignal, + Name: "buildsignal", + ConsumerGroup: "orchestrator-buildsignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return buildsignal.NewController(d.Logger, d.Scope, d.Storage, d.BuildRunner, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBuildSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyMerge, + Name: "submitqueue-merge", + ConsumerGroup: "orchestrator-merge", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return merge.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, runwaymq.TopicKeyMerge, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: runwaymq.TopicKeyMergeSignal, + Name: "merge-signal", + ConsumerGroup: "orchestrator-mergesignal", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return mergesignal.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQMergeSignalController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, + { + Key: topickey.TopicKeyConclude, + Name: "conclude", + ConsumerGroup: "orchestrator-conclude", + New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return conclude.NewController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) { + return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil + }, + }, +} + +// PublishOnlyTopics declares topics the orchestrator publishes to but does +// not consume. These are registered in the TopicRegistry so controllers +// can look up topic names for publishing. +var PublishOnlyTopics = []pipeline.PublishOnlyTopic{ + // Log: the orchestrator emits request-log entries; the gateway consumes them. + {Key: topickey.TopicKeyLog, Name: "log"}, + // Merge-conflict check: the orchestrator publishes check requests to runway. + {Key: runwaymq.TopicKeyMergeConflictCheck, Name: "merge-conflict-check"}, + // Merge: the orchestrator publishes merge requests to runway. + {Key: runwaymq.TopicKeyMerge, Name: "runway-merge"}, +} + +// Controllers holds the orchestrator's RPC-facing controllers, constructed +// but NOT bound to any wire contract. Binding to a proto service + transport +// is host glue, because deployers may use different protos or transports. +type Controllers struct { + // Ping is the health-check controller. + Ping *controller.PingController +} + +// NewControllers creates the orchestrator's RPC controllers from the given Deps. +// The PingController takes a base *zap.Logger, so we desugar the SugaredLogger. +func NewControllers(d Deps) Controllers { + return Controllers{ + Ping: controller.NewPingController(d.Logger.Desugar(), d.Scope), + } +}