Add peer fields to expander/grpcplugin to replace inlined non proto-message Kubernetes objects

Author: liggitt

cluster-autoscaler/expander/grpcplugin/grpc_client.go

@@ -80,13 +80,13 @@ func (g *grpcclientstrategy) BestOptions(expansionOptions []expander.Option, nod
 
 	// Transform inputs to gRPC inputs
 	grpcOptionsSlice, nodeGroupIDOptionMap := populateOptionsForGRPC(expansionOptions)
-	grpcNodeMap := populateNodeInfoForGRPC(nodeInfo)
+	grpcNodeMap, grpcNodeBytesMap := populateNodeInfoForGRPC(nodeInfo)
 
 	// call gRPC server to get BestOption
 	klog.V(2).Infof("GPRC call of best options to server with %v options", len(nodeGroupIDOptionMap))
 	ctx, cancel := context.WithTimeout(context.Background(), gRPCTimeout)
 	defer cancel()
-	bestOptionsResponse, err := g.grpcClient.BestOptions(ctx, &protos.BestOptionsRequest{Options: grpcOptionsSlice, NodeMap: grpcNodeMap})
+	bestOptionsResponse, err := g.grpcClient.BestOptions(ctx, &protos.BestOptionsRequest{Options: grpcOptionsSlice, NodeMap: grpcNodeMap, NodeBytesMap: grpcNodeBytesMap})
 	if err != nil {
 		klog.V(4).Infof("GRPC call failed, no options filtered: %v", err)
 		return expansionOptions
@@ -117,12 +117,26 @@ func populateOptionsForGRPC(expansionOptions []expander.Option) ([]*protos.Optio
 }
 
 // populateNodeInfoForGRPC looks at the corresponding v1.Node object per NodeInfo object, and populates the grpcNodeInfoMap with these to pass over grpc
-func populateNodeInfoForGRPC(nodeInfos map[string]*framework.NodeInfo) map[string]*v1.Node {
+func populateNodeInfoForGRPC(nodeInfos map[string]*framework.NodeInfo) (map[string]*v1.Node, map[string][]byte) {
 	grpcNodeInfoMap := make(map[string]*v1.Node)
+	grpcNodeBytesMap := make(map[string][]byte)
 	for nodeId, nodeInfo := range nodeInfos {
-		grpcNodeInfoMap[nodeId] = nodeInfo.Node()
+		node := nodeInfo.Node()
+		grpcNodeInfoMap[nodeId] = node
+
+		// if we're still accumulating node bytes
+		if grpcNodeBytesMap != nil {
+			// try to serialize to proto bytes
+			nodeBytes, err := node.Marshal()
+			if err != nil {
+				// unexpected proto serialization error, avoid sending nodeBytes map at all
+				grpcNodeBytesMap = nil
+			} else {
+				grpcNodeBytesMap[nodeId] = nodeBytes
+			}
+		}
 	}
-	return grpcNodeInfoMap
+	return grpcNodeInfoMap, grpcNodeBytesMap
 }
 
 func transformAndSanitizeOptionsFromGRPC(bestOptionsResponseOptions []*protos.Option, nodeGroupIDOptionMap map[string]expander.Option) []expander.Option {
@@ -143,5 +157,17 @@ func transformAndSanitizeOptionsFromGRPC(bestOptionsResponseOptions []*protos.Op
 }
 
 func newOptionMessage(nodeGroupId string, nodeCount int32, debug string, pods []*v1.Pod) *protos.Option {
-	return &protos.Option{NodeGroupId: nodeGroupId, NodeCount: nodeCount, Debug: debug, Pod: pods}
+	var podsBytes [][]byte
+	if len(pods) > 0 {
+		podsBytes = make([][]byte, 0, len(pods))
+		for _, pod := range pods {
+			podBytes, err := pod.Marshal()
+			if err != nil {
+				podsBytes = nil
+				break
+			}
+			podsBytes = append(podsBytes, podBytes)
+		}
+	}
+	return &protos.Option{NodeGroupId: nodeGroupId, NodeCount: nodeCount, Debug: debug, Pod: pods, PodBytes: podsBytes}
 }

cluster-autoscaler/expander/grpcplugin/grpc_client_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/autoscaler/cluster-autoscaler/expander/grpcplugin/protos"
 	"k8s.io/autoscaler/cluster-autoscaler/expander/mocks"
@@ -135,13 +136,17 @@ func makeFakeNodeInfos() map[string]*framework.NodeInfo {
 
 func TestPopulateNodeInfoForGRPC(t *testing.T) {
 	nodeInfos := makeFakeNodeInfos()
-	grpcNodeInfoMap := populateNodeInfoForGRPC(nodeInfos)
+	grpcNodeInfoMap, grpcNodeBytesMap := populateNodeInfoForGRPC(nodeInfos)
 
 	expectedGrpcNodeInfoMap := make(map[string]*v1.Node)
+	expectedGrpcNodeBytesMap := make(map[string][]byte)
 	for i, opt := range options {
 		expectedGrpcNodeInfoMap[opt.NodeGroup.Id()] = nodes[i]
+		expectedGrpcNodeBytesMap[opt.NodeGroup.Id()], _ = nodes[i].Marshal()
 	}
+
 	assert.Equal(t, expectedGrpcNodeInfoMap, grpcNodeInfoMap)
+	assert.Equal(t, expectedGrpcNodeBytesMap, grpcNodeBytesMap)
 }
 
 func TestValidTransformAndSanitizeOptionsFromGRPC(t *testing.T) {
@@ -179,12 +184,15 @@ func TestBestOptionsValid(t *testing.T) {
 
 	nodeInfos := makeFakeNodeInfos()
 	grpcNodeInfoMap := make(map[string]*v1.Node)
+	grpcNodeBytesMap := make(map[string][]byte)
 	for i, opt := range options {
 		grpcNodeInfoMap[opt.NodeGroup.Id()] = nodes[i]
+		grpcNodeBytesMap[opt.NodeGroup.Id()], _ = nodes[i].Marshal()
 	}
 	expectedBestOptionsReq := &protos.BestOptionsRequest{
-		Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
-		NodeMap: grpcNodeInfoMap,
+		Options:      []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
+		NodeMap:      grpcNodeInfoMap,
+		NodeBytesMap: grpcNodeBytesMap,
 	}
 
 	mockClient.EXPECT().BestOptions(
@@ -220,12 +228,14 @@ func TestBestOptionsEmpty(t *testing.T) {
 		},
 	}
 	for _, tc := range testCases {
-		grpcNodeInfoMap := populateNodeInfoForGRPC(makeFakeNodeInfos())
+		grpcNodeInfoMap, grpcNodeBytesMap := populateNodeInfoForGRPC(makeFakeNodeInfos())
+		assert.NotNil(t, grpcNodeBytesMap)
 		mockClient.EXPECT().BestOptions(
 			gomock.Any(), gomock.Eq(
 				&protos.BestOptionsRequest{
-					Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
-					NodeMap: grpcNodeInfoMap,
+					Options:      []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
+					NodeMap:      grpcNodeInfoMap,
+					NodeBytesMap: grpcNodeBytesMap,
 				})).Return(&tc.mockResponse, nil)
 		resp := g.BestOptions(options, makeFakeNodeInfos())
 
@@ -284,13 +294,15 @@ func TestBestOptionsErrors(t *testing.T) {
 		},
 	}
 	for _, tc := range testCases {
-		grpcNodeInfoMap := populateNodeInfoForGRPC(tc.nodeInfo)
+		grpcNodeInfoMap, grpcNodeBytesMap := populateNodeInfoForGRPC(tc.nodeInfo)
+		assert.NotNil(t, grpcNodeBytesMap)
 		if tc.client.grpcClient != nil {
 			mockClient.EXPECT().BestOptions(
 				gomock.Any(), gomock.Eq(
 					&protos.BestOptionsRequest{
-						Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
-						NodeMap: grpcNodeInfoMap,
+						Options:      []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
+						NodeMap:      grpcNodeInfoMap,
+						NodeBytesMap: grpcNodeBytesMap,
 					})).Return(&tc.mockResponse, tc.errResponse)
 		}
 		resp := tc.client.BestOptions(options, tc.nodeInfo)