Merge pull request #800 from traPtitech/builder-priority

Builder priority

.local-dev/config/ns.yaml

@@ -36,6 +36,7 @@ components:
         platformAPI: "0.11"
     controller:
       url: http://ns-controller:10000
+    priority: 0
 
   controller:
     port: 10000

.local-dev/ext-builder/config.yaml

@@ -39,3 +39,4 @@ components:
         platformAPI: "0.11"
     controller:
       url: http://host.docker.internal:10000
+    priority: 10

.local-dev/manifest/ns-system/config/ns.yaml

@@ -43,6 +43,7 @@ components:
         platformAPI: "0.11"
     controller:
       url: http://ns-controller.ns-system.svc.cluster.local:10000
+    priority: 0
 
   controller:
     port: 10000

api/proto/neoshowcase/protobuf/controller.proto

@@ -21,6 +21,11 @@ message BuilderRequest {
   }
 }
 
+message ConnectedBody {
+  // Larger value means higher priority
+  int64 priority = 1;
+}
+
 message BuildStarted {
   string build_id = 1;
 }
@@ -49,9 +54,10 @@ message BuilderResponse {
   }
   Type type = 1;
   oneof body {
-    BuildStarted started = 2;
-    BuildSettled settled = 3;
-    BuildLogPortion log = 4;
+    ConnectedBody connected = 2;
+    BuildStarted started = 3;
+    BuildSettled settled = 4;
+    BuildLogPortion log = 5;
   }
 }
 

cmd/config.go

@@ -58,6 +58,7 @@ type BuilderConfig struct {
 		K8s     bk8simpl.Config    `mapstructure:"k8s" yaml:"k8s"`
 	}
 	Controller grpc.ControllerServiceClientConfig `mapstructure:"controller" yaml:"controller"`
+	Priority   int                                `mapstructure:"priority" yaml:"priority"`
 }
 
 type ControllerConfig struct {
@@ -161,6 +162,8 @@ func init() {
 
 	viper.SetDefault("components.builder.controller.url", "http://ns-controller:10000")
 
+	viper.SetDefault("components.builder.priority", 0)
+
 	viper.SetDefault("components.controller.port", 10000)
 	viper.SetDefault("components.controller.mode", "docker")
 

cmd/providers.go

@@ -67,7 +67,7 @@ var providers = wire.NewSet(
 	grpc.NewControllerService,
 	grpc.NewControllerServiceClient,
 	grpc.NewControllerBuilderService,
-	grpc.NewControllerBuilderServiceClient,
+	provideControllerBuilderServiceClient,
 	grpc.NewControllerSSGenService,
 	grpc.NewControllerSSGenServiceClient,
 	healthcheck.NewServer,
@@ -131,6 +131,13 @@ func provideAuthDevServer(c Config) *authdev.Server {
 	return authdev.NewServer(cc.Header, cc.Port, cc.User)
 }
 
+func provideControllerBuilderServiceClient(c Config) domain.ControllerBuilderServiceClient {
+	return grpc.NewControllerBuilderServiceClient(
+		c.Components.Builder.Controller,
+		c.Components.Builder.Priority,
+	)
+}
+
 func provideBuildpackBackend(c Config) (builder.BuildpackBackend, error) {
 	cc := c.Components.Builder
 	switch cc.Buildpack.Backend {

cmd/wire.go

@@ -29,7 +29,6 @@ func NewAuthDev(c Config) (component, error) {
 func NewBuilder(c Config) (component, error) {
 	wire.Build(
 		providers,
-		wire.FieldsOf(new(BuilderConfig), "Controller"),
 		wire.Bind(new(component), new(*builder.Server)),
 		wire.Struct(new(builder.Server), "*"),
 	)

dashboard/src/api/neoshowcase/protobuf/gateway_pb.ts

@@ -1,4 +1,4 @@
-// @generated by protoc-gen-es v1.4.1 with parameter "target=ts"
+// @generated by protoc-gen-es v1.5.0 with parameter "target=ts"
 // @generated from file neoshowcase/protobuf/gateway.proto (package neoshowcase.protobuf, syntax proto3)
 /* eslint-disable */
 // @ts-nocheck

dashboard/src/api/neoshowcase/protobuf/null_pb.ts

@@ -1,4 +1,4 @@
-// @generated by protoc-gen-es v1.4.1 with parameter "target=ts"
+// @generated by protoc-gen-es v1.5.0 with parameter "target=ts"
 // @generated from file neoshowcase/protobuf/null.proto (package neoshowcase.protobuf, syntax proto3)
 /* eslint-disable */
 // @ts-nocheck

pkg/domain/component.go

@@ -36,7 +36,7 @@ type ControllerBuilderService interface {
 	ListenBuilderIdle() (sub <-chan struct{}, unsub func())
 	ListenBuildSettled() (sub <-chan struct{}, unsub func())
 	StartBuilds(buildIDs []string)
-	BroadcastBuilder(req *pb.BuilderRequest)
+	CancelBuild(buildID string)
 }
 
 type ControllerBuilderServiceClient interface {

pkg/infrastructure/grpc/controller_builder_service.go

@@ -1,14 +1,14 @@
 package grpc
 
 import (
-	"context"
-	"io"
-	"sync"
-
 	"connectrpc.com/connect"
+	"context"
 	"github.com/friendsofgo/errors"
 	"github.com/samber/lo"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/exp/slices"
+	"io"
+	"sync"
 
 	"github.com/traPtitech/neoshowcase/pkg/domain"
 	"github.com/traPtitech/neoshowcase/pkg/infrastructure/grpc/pb"
@@ -18,6 +18,27 @@ import (
 
 type builderConnection struct {
 	reqSender chan<- *pb.BuilderRequest
+	priority  int64
+	buildID   string
+}
+
+func (c *builderConnection) Send(req *pb.BuilderRequest) {
+	select {
+	case c.reqSender <- req:
+	default:
+	}
+}
+
+func (c *builderConnection) SetBuildID(id string) {
+	c.buildID = id
+}
+
+func (c *builderConnection) ClearBuildID() {
+	c.buildID = ""
+}
+
+func (c *builderConnection) Busy() bool {
+	return c.buildID != ""
 }
 
 type ControllerBuilderService struct {
@@ -73,11 +94,16 @@ func (s *ControllerBuilderService) ConnectBuilder(ctx context.Context, st *conne
 
 			s.lock.Lock()
 			switch res.Type {
+			case pb.BuilderResponse_CONNECTED:
+				payload := res.Body.(*pb.BuilderResponse_Connected).Connected
+				conn.priority = payload.Priority
 			case pb.BuilderResponse_BUILD_STARTED:
 				payload := res.Body.(*pb.BuilderResponse_Started).Started
+				conn.SetBuildID(payload.BuildId)
 				s.logStream.StartBuildLog(payload.BuildId)
 			case pb.BuilderResponse_BUILD_SETTLED:
 				payload := res.Body.(*pb.BuilderResponse_Settled).Settled
+				conn.ClearBuildID()
 				s.idle.Publish(struct{}{})
 				s.settled.Publish(struct{}{})
 				s.logStream.CloseBuildLog(payload.BuildId)
@@ -113,31 +139,39 @@ func (s *ControllerBuilderService) ListenBuildSettled() (sub <-chan struct{}, un
 	return s.settled.Subscribe()
 }
 
-func (s *ControllerBuilderService) broadcast(req *pb.BuilderRequest) {
-	for _, builder := range s.builderConnections {
-		select {
-		case builder.reqSender <- req:
-		default:
-		}
-	}
-}
-
 func (s *ControllerBuilderService) StartBuilds(buildIDs []string) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	// Send at most n (= number of builders) build requests
-	n := len(s.builderConnections)
-	for _, buildID := range ds.FirstN(buildIDs, n) {
-		s.broadcast(&pb.BuilderRequest{
+	// Select available builders (and copy the slice)
+	conns := lo.Filter(s.builderConnections, func(c *builderConnection, _ int) bool { return !c.Busy() })
+	// Select from higher priority builders
+	slices.SortFunc(conns, ds.MoreFunc(func(c *builderConnection) int64 { return c.priority }))
+
+	// Send builds to available builders
+	for i, conn := range ds.FirstN(conns, len(buildIDs)) {
+		buildID := buildIDs[i]
+		conn.Send(&pb.BuilderRequest{
 			Type: pb.BuilderRequest_START_BUILD,
-			Body: &pb.BuilderRequest_StartBuild{StartBuild: &pb.StartBuildRequest{BuildId: buildID}},
+			Body: &pb.BuilderRequest_StartBuild{StartBuild: &pb.StartBuildRequest{
+				BuildId: buildID,
+			}},
 		})
 	}
 }
 
-func (s *ControllerBuilderService) BroadcastBuilder(req *pb.BuilderRequest) {
+func (s *ControllerBuilderService) CancelBuild(buildID string) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	s.broadcast(req)
+
+	conns := lo.Filter(s.builderConnections, func(c *builderConnection, _ int) bool { return c.buildID == buildID })
+	// assert len(conns) <= 1
+	for _, conn := range conns {
+		conn.Send(&pb.BuilderRequest{
+			Type: pb.BuilderRequest_CANCEL_BUILD,
+			Body: &pb.BuilderRequest_CancelBuild{CancelBuild: &pb.BuildIdRequest{
+				BuildId: buildID,
+			}},
+		})
+	}
 }

pkg/infrastructure/grpc/controller_builder_service_client.go

@@ -14,14 +14,17 @@ import (
 )
 
 type ControllerBuilderServiceClient struct {
-	client pbconnect.ControllerBuilderServiceClient
+	client   pbconnect.ControllerBuilderServiceClient
+	priority int
 }
 
 func NewControllerBuilderServiceClient(
 	c ControllerServiceClientConfig,
+	priority int,
 ) domain.ControllerBuilderServiceClient {
 	return &ControllerBuilderServiceClient{
-		client: pbconnect.NewControllerBuilderServiceClient(web.NewH2CClient(), c.URL),
+		client:   pbconnect.NewControllerBuilderServiceClient(web.NewH2CClient(), c.URL),
+		priority: priority,
 	}
 }
 
@@ -38,7 +41,12 @@ func (c *ControllerBuilderServiceClient) ConnectBuilder(ctx context.Context, onR
 
 		// Need to send one arbitrary event to actually start the connection
 		// not sure if this is a bug with connect protocol or something
-		err := st.Send(&pb.BuilderResponse{Type: pb.BuilderResponse_CONNECTED})
+		err := st.Send(&pb.BuilderResponse{
+			Type: pb.BuilderResponse_CONNECTED,
+			Body: &pb.BuilderResponse_Connected{Connected: &pb.ConnectedBody{
+				Priority: int64(c.priority),
+			}},
+		})
 		if err != nil {
 			log.Errorf("failed to send connected event: %+v", err)
 			return

pkg/infrastructure/grpc/controller_service.go

@@ -113,10 +113,7 @@ loop:
 
 func (s *ControllerService) CancelBuild(_ context.Context, c *connect.Request[pb.BuildIdRequest]) (*connect.Response[emptypb.Empty], error) {
 	buildID := c.Msg.BuildId
-	s.builder.BroadcastBuilder(&pb.BuilderRequest{
-		Type: pb.BuilderRequest_CANCEL_BUILD,
-		Body: &pb.BuilderRequest_CancelBuild{CancelBuild: &pb.BuildIdRequest{BuildId: buildID}},
-	})
+	s.builder.CancelBuild(buildID)
 	res := connect.NewResponse(&emptypb.Empty{})
 	return res, nil
 }