chore(example): wire stovepipe orchestrator start controller

Wires the start controller into the runnable orchestrator sample app.

Author: mnoah1

example/stovepipe/orchestrator/server/BUILD.bazel

@@ -12,7 +12,16 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//api/stovepipe/orchestrator/protopb",
+        "//platform/consumer",
+        "//platform/errs",
+        "//platform/errs/generic",
+        "//platform/errs/mysql",
+        "//platform/extension/messagequeue",
+        "//platform/extension/messagequeue/mysql",
+        "//stovepipe/core/topickey",
         "//stovepipe/orchestrator/controller",
+        "//stovepipe/orchestrator/controller/start",
+        "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_uber_go_tally//:tally",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//reflection",

example/stovepipe/orchestrator/server/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"context"
+	"database/sql"
 	"errors"
 	"fmt"
 	"net"
@@ -25,9 +26,18 @@ import (
 	"syscall"
 	"time"
 
+	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
 	pb "github.com/uber/submitqueue/api/stovepipe/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"
+	extqueue "github.com/uber/submitqueue/platform/extension/messagequeue"
+	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
+	"github.com/uber/submitqueue/stovepipe/core/topickey"
 	"github.com/uber/submitqueue/stovepipe/orchestrator/controller"
+	"github.com/uber/submitqueue/stovepipe/orchestrator/controller/start"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
@@ -102,6 +112,62 @@ func run() error {
 		metricsWgDone.Wait()
 	}()
 
+	queueDSN := os.Getenv("QUEUE_MYSQL_DSN")
+	if queueDSN == "" {
+		return fmt.Errorf("QUEUE_MYSQL_DSN environment variable is required")
+	}
+	queueDB, err := sql.Open("mysql", queueDSN)
+	if err != nil {
+		return fmt.Errorf("failed to open queue database: %w", err)
+	}
+	defer queueDB.Close()
+
+	mysqlQueue, err := queueMySQL.NewQueue(queueMySQL.Params{
+		DB:           queueDB,
+		Logger:       logger,
+		MetricsScope: scope.SubScope("queue"),
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create queue: %w", err)
+	}
+	defer mysqlQueue.Close()
+
+	logger.Info("initialized queue", zap.String("dsn", queueDSN))
+
+	subscriberName := os.Getenv("HOSTNAME")
+	if subscriberName == "" {
+		subscriberName = fmt.Sprintf("stovepipe-orchestrator-%d", time.Now().Unix())
+	}
+
+	registry, err := newTopicRegistry(mysqlQueue, subscriberName)
+	if err != nil {
+		return fmt.Errorf("failed to create topic registry: %w", err)
+	}
+
+	primaryConsumer := consumer.New(logger.Sugar(), scope.SubScope("consumer"), registry,
+		errs.NewClassifierProcessor(
+			genericerrs.Classifier,
+			mysqlerrs.Classifier,
+		),
+	)
+
+	startController := start.NewController(start.Params{
+		Logger:        logger.Sugar(),
+		Scope:         scope,
+		Registry:      registry,
+		TopicKey:      topickey.TopicKeyStart,
+		ConsumerGroup: "orchestrator-start",
+	})
+	if err := primaryConsumer.Register(startController); err != nil {
+		return fmt.Errorf("failed to register start controller: %w", err)
+	}
+	logger.Info("controllers registered", zap.Int("primary", 1))
+
+	if err := primaryConsumer.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start primary consumer: %w", err)
+	}
+	logger.Info("consumer started")
+
 	grpcServer := grpc.NewServer()
 
 	pingController := controller.NewPingController(logger, scope)
@@ -140,11 +206,42 @@ func run() error {
 		serverErr = <-serverErrCh
 	case serverErr = <-serverErrCh:
 		fmt.Println("Shutting down stovepipe orchestrator server due to critical GRPC server error...")
+		cancel()
 	}
 
 	if serverErr != nil {
-		err = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+		serverErr = fmt.Errorf("GRPC server exited with error: %w", serverErr)
+	}
+
+	primaryStopErr := primaryConsumer.Stop(30000)
+	if primaryStopErr != nil {
+		primaryStopErr = fmt.Errorf("failed to stop consumer: %w", primaryStopErr)
+	}
+
+	if primaryStopErr != nil || serverErr != nil {
+		err = errors.Join(primaryStopErr, serverErr)
 	}
 
 	return err
 }
+
+func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRegistry, error) {
+	return consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{
+			Key:   topickey.TopicKeyStart,
+			Name:  "start",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "orchestrator-start",
+			),
+		},
+		{
+			Key:   topickey.TopicKeyValidate,
+			Name:  "validate",
+			Queue: q,
+			Subscription: extqueue.DefaultSubscriptionConfig(
+				subscriberName, "orchestrator-validate",
+			),
+		},
+	})
+}