diff --git a/submitqueue/orchestrator/BUILD.bazel b/submitqueue/orchestrator/BUILD.bazel new file mode 100644 index 00000000..975406ed --- /dev/null +++ b/submitqueue/orchestrator/BUILD.bazel @@ -0,0 +1,35 @@ +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/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/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..7e67fc76 --- /dev/null +++ b/submitqueue/orchestrator/pipeline.go @@ -0,0 +1,238 @@ +// 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/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/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 + + // 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 → 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.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), + } +}