add utils for send server messages to controller

scheduler/pkg/server/server.go

@@ -508,6 +508,19 @@ func createServerStatusResponse(s *store.ServerSnapshot) *pb.ServerStatusRespons
 	return resp
 }
 
+func createServerScaleResponse(s *store.ServerSnapshot, expectedReplicas uint32) *pb.ServerStatusResponse {
+	// we dont care about populating the other fields as they should not be used by the controller, reconsider if this changes
+
+	resp := &pb.ServerStatusResponse{
+		Type:             pb.ServerStatusResponse_ScalingRequest,
+		ServerName:       s.Name,
+		ExpectedReplicas: int32(expectedReplicas),
+		KubernetesMeta:   s.KubernetesMeta,
+	}
+
+	return resp
+}
+
 func (s *SchedulerServer) StartExperiment(ctx context.Context, req *pb.StartExperimentRequest) (*pb.StartExperimentResponse, error) {
 	err := s.experimentServer.StartExperiment(experiment.CreateExperimentFromRequest(req.Experiment))
 	if err != nil {

scheduler/pkg/server/server_status.go

@@ -15,6 +15,7 @@ import (
 	pb "github.com/seldonio/seldon-core/apis/go/v2/mlops/scheduler"
 
 	"github.com/seldonio/seldon-core/scheduler/v2/pkg/coordinator"
+	"github.com/seldonio/seldon-core/scheduler/v2/pkg/store"
 )
 
 func (s *SchedulerServer) SubscribeModelStatus(req *pb.ModelSubscriptionRequest, stream pb.Scheduler_SubscribeModelStatusServer) error {
@@ -247,20 +248,33 @@ func (s *SchedulerServer) sendServerStatus() {
 			continue
 		}
 		ssr := createServerStatusResponse(server)
+		s.sendServerResponse(ssr)
+	}
 
-		for stream, subscription := range s.serverEventStream.streams {
-			hasExpired, err := sendWithTimeout(func() error { return stream.Send(ssr) }, s.timeout)
-			if hasExpired {
-				// this should trigger a reconnect from the client
-				close(subscription.fin)
-				delete(s.serverEventStream.streams, stream)
-			}
-			if err != nil {
-				logger.WithError(err).Errorf("Failed to send server status event to %s", subscription.name)
-			}
+}
+
+func (s *SchedulerServer) sendServerScale(server *store.ServerSnapshot, expectedReplicas uint32) {
+	// TODO: should there be some sort of velocity check ?
+	logger := s.logger.WithField("func", "sendServerScale")
+	logger.Debugf("will attempt to scale servers to %d for %v", expectedReplicas, server.Name)
+
+	ssr := createServerScaleResponse(server, expectedReplicas)
+	s.sendServerResponse(ssr)
+}
+
+func (s *SchedulerServer) sendServerResponse(ssr *pb.ServerStatusResponse) {
+	logger := s.logger.WithField("func", "sendServerStatusResponse")
+	for stream, subscription := range s.serverEventStream.streams {
+		hasExpired, err := sendWithTimeout(func() error { return stream.Send(ssr) }, s.timeout)
+		if hasExpired {
+			// this should trigger a reconnect from the client
+			close(subscription.fin)
+			delete(s.serverEventStream.streams, stream)
+		}
+		if err != nil {
+			logger.WithError(err).Errorf("Failed to send server status response to %s", subscription.name)
 		}
 	}
-
 }
 
 // initial send of server statuses to a new controller