From 585d2b4ee264103cdd61563c4c820993bdea2106 Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Tue, 14 Jul 2026 21:31:13 +0000 Subject: [PATCH 1/2] feat(stovepipe): wire build controller into server --- service/stovepipe/server/BUILD.bazel | 6 +- service/stovepipe/server/main.go | 110 ++++++++++++++++++++++----- 2 files changed, 95 insertions(+), 21 deletions(-) diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index fa4612ff..ea073e1f 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -15,12 +15,16 @@ go_library( "//platform/extension/messagequeue/mysql:go_default_library", "//service/stovepipe/server/mapper:go_default_library", "//stovepipe/controller:go_default_library", + "//stovepipe/controller/build:go_default_library", "//stovepipe/controller/dlq:go_default_library", "//stovepipe/controller/process:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/extension/buildrunner:go_default_library", + "//stovepipe/extension/buildrunner/fake:go_default_library", "//stovepipe/extension/queueconfig/default:go_default_library", "//stovepipe/extension/sourcecontrol:go_default_library", "//stovepipe/extension/sourcecontrol/fake:go_default_library", + "//stovepipe/extension/storage:go_default_library", "//stovepipe/extension/storage/mysql:go_default_library", "@com_github_go_sql_driver_mysql//:go_default_library", "@com_github_uber_go_tally//:go_default_library", @@ -31,7 +35,7 @@ go_library( ) go_binary( - name = "stovepipe", + name = "server", embed = [":go_default_library"], visibility = ["//visibility:public"], ) diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index 7b3aae31..8e40c51f 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -37,12 +37,16 @@ import ( queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" "github.com/uber/submitqueue/service/stovepipe/server/mapper" "github.com/uber/submitqueue/stovepipe/controller" + "github.com/uber/submitqueue/stovepipe/controller/build" "github.com/uber/submitqueue/stovepipe/controller/dlq" "github.com/uber/submitqueue/stovepipe/controller/process" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/extension/buildrunner" + buildrunnerfake "github.com/uber/submitqueue/stovepipe/extension/buildrunner/fake" queueconfigdefault "github.com/uber/submitqueue/stovepipe/extension/queueconfig/default" "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol" sourcecontrolfake "github.com/uber/submitqueue/stovepipe/extension/sourcecontrol/fake" + "github.com/uber/submitqueue/stovepipe/extension/storage" storageMySQL "github.com/uber/submitqueue/stovepipe/extension/storage/mysql" "go.uber.org/zap" "google.golang.org/grpc" @@ -100,6 +104,15 @@ func (fakeSourceControlFactory) For(cfg sourcecontrol.Config) (sourcecontrol.Sou return sourcecontrolfake.New([]string{fmt.Sprintf("git://%s/HEAD", cfg.QueueName)}), nil } +// fakeBuildRunnerFactory is the example BuildRunner factory: every queue shares the same +// stateless fake runner, which succeeds unless a caller embeds a failure marker in the head +// URI. A real deployment supplies a backend-specific factory (e.g. Buildkite, per queue). +type fakeBuildRunnerFactory struct{} + +func (fakeBuildRunnerFactory) For(_ buildrunner.Config) (buildrunner.BuildRunner, error) { + return buildrunnerfake.New(), nil +} + func main() { code := 0 if err := run(); err != nil { @@ -225,24 +238,21 @@ func run() error { errs.AlwaysRetryableProcessor, ) - processController := process.NewController( - logger.Sugar(), - scope, - store, - queueconfigdefault.NewStore(), - fakeSourceControlFactory{}, - registry, - stovepipemq.TopicKeyProcess, - "stovepipe-process", - ) - if err := primaryConsumer.Register(processController); err != nil { - return fmt.Errorf("failed to register process controller: %w", err) - } + // Each factory is constructed once and threaded through every consumer of + // it, so a real (stateful) backend introduced later is shared rather than + // silently duplicated across controllers. + scf := fakeSourceControlFactory{} + brf := fakeBuildRunnerFactory{} - processDLQController := dlq.NewController(logger.Sugar(), scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") - if err := dlqConsumer.Register(processDLQController); err != nil { - return fmt.Errorf("failed to register process dlq controller: %w", err) + primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, store, registry, scf, brf) + if err != nil { + return err } + dlqCount, err := registerDLQControllers(dlqConsumer, logger.Sugar(), scope, store, registry) + 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 primary consumer then fails to start, the half we @@ -266,7 +276,7 @@ func run() error { logger.Sugar(), scope, newInMemoryCounter(), - fakeSourceControlFactory{}, + scf, store, registry, ) @@ -338,10 +348,67 @@ func run() error { return err } +// registerPrimaryControllers creates the primary-pipeline queue controllers and +// registers them with c, returning how many were registered. +func registerPrimaryControllers( + c consumer.Consumer, + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, + scf sourcecontrol.Factory, + brf buildrunner.Factory, +) (int, error) { + var count int + + processController := process.NewController( + logger, + scope, + store, + queueconfigdefault.NewStore(), + scf, + registry, + stovepipemq.TopicKeyProcess, + "stovepipe-process", + ) + if err := c.Register(processController); err != nil { + return count, fmt.Errorf("failed to register process controller: %w", err) + } + count++ + + buildController := build.NewController(logger, scope, store, brf, registry, stovepipemq.TopicKeyBuild, "stovepipe-build") + if err := c.Register(buildController); err != nil { + return count, fmt.Errorf("failed to register build controller: %w", err) + } + count++ + + return count, nil +} + +// registerDLQControllers creates one DLQ reconciler per primary stage and +// registers them with c, returning how many were registered. +func registerDLQControllers( + c consumer.Consumer, + logger *zap.SugaredLogger, + scope tally.Scope, + store storage.Storage, + registry consumer.TopicRegistry, +) (int, error) { + var count int + + processDLQController := dlq.NewController(logger, scope, store, dlq.TopicKey(stovepipemq.TopicKeyProcess), "stovepipe-process-dlq") + if err := c.Register(processDLQController); err != nil { + return count, fmt.Errorf("failed to register process dlq controller: %w", err) + } + count++ + + return count, nil +} + // newTopicRegistry builds the TopicRegistry for Stovepipe's internal pipeline queues. ingest -// publishes to process; process publishes admitted requests to the publish-only build topic. -// The process_dlq topic is the dead-letter destination the queue backend routes to (per -// DefaultSubscriptionConfig's DLQ.TopicSuffix) when the process controller exhausts retries. +// publishes to the process topic and the process consumer subscribes to it; process publishes +// to the build topic and the build consumer subscribes to it. The buildsignal topic is added +// once the buildsignal controller lands to consume it. func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) { return consumer.NewTopicRegistry([]consumer.TopicConfig{ { @@ -356,6 +423,9 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: q, + Subscription: extqueue.DefaultSubscriptionConfig( + subscriberName, "stovepipe-build", + ), }, { Key: dlq.TopicKey(stovepipemq.TopicKeyProcess), From fcbdc3c8df887ad897e9034b99e6f72cb65bf95e Mon Sep 17 00:00:00 2001 From: "chenghan.ying" Date: Mon, 20 Jul 2026 23:41:50 +0000 Subject: [PATCH 2/2] fix(stovepipe): restore go_binary name to match repo convention --- service/stovepipe/server/BUILD.bazel | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/stovepipe/server/BUILD.bazel b/service/stovepipe/server/BUILD.bazel index ea073e1f..49158b2c 100644 --- a/service/stovepipe/server/BUILD.bazel +++ b/service/stovepipe/server/BUILD.bazel @@ -35,7 +35,7 @@ go_library( ) go_binary( - name = "server", + name = "stovepipe", embed = [":go_default_library"], visibility = ["//visibility:public"], )