Merge pull request #890 from lec-bit/big-scale-bug

make xds request send async

Author: kmesh-bot

pkg/controller/ads/ads_controller.go

@@ -22,6 +22,7 @@ import (
 
 	service_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	resource_v3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	"istio.io/istio/pkg/channels"
 
 	"kmesh.net/kmesh/pkg/logger"
 )
@@ -31,8 +32,14 @@ var (
 )
 
 type Controller struct {
-	Stream    service_discovery_v3.AggregatedDiscoveryService_StreamAggregatedResourcesClient
 	Processor *processor
+	con       *connection
+}
+
+type connection struct {
+	Stream       service_discovery_v3.AggregatedDiscoveryService_StreamAggregatedResourcesClient
+	requestsChan *channels.Unbounded[*service_discovery_v3.DiscoveryRequest]
+	stopCh       chan struct{}
 }
 
 func NewController() *Controller {
@@ -42,17 +49,26 @@ func NewController() *Controller {
 }
 
 func (c *Controller) AdsStreamCreateAndSend(client service_discovery_v3.AggregatedDiscoveryServiceClient, ctx context.Context) error {
-	var err error
+	if c.con != nil {
+		close(c.con.stopCh)
+	}
 
-	c.Stream, err = client.StreamAggregatedResources(ctx)
+	stream, err := client.StreamAggregatedResources(ctx)
 	if err != nil {
 		return fmt.Errorf("StreamAggregatedResources failed, %s", err)
 	}
 
+	c.con = &connection{
+		Stream:       stream,
+		requestsChan: channels.NewUnbounded[*service_discovery_v3.DiscoveryRequest](),
+		stopCh:       make(chan struct{}),
+	}
+
 	c.Processor.Reset()
-	if err := c.Stream.Send(newAdsRequest(resource_v3.ClusterType, nil, "")); err != nil {
+	if err := stream.Send(newAdsRequest(resource_v3.ClusterType, nil, "")); err != nil {
 		return fmt.Errorf("send request failed, %s", err)
 	}
+	go sendUpstream(c.con)
 
 	return nil
 }
@@ -62,26 +78,39 @@ func (c *Controller) HandleAdsStream() error {
 		err error
 		rsp *service_discovery_v3.DiscoveryResponse
 	)
-
-	if rsp, err = c.Stream.Recv(); err != nil {
+	if rsp, err = c.con.Stream.Recv(); err != nil {
+		_ = c.con.Stream.CloseSend()
 		return fmt.Errorf("stream recv failed, %s", err)
 	}
 
 	c.Processor.processAdsResponse(rsp)
-	defer func() {
+	c.con.requestsChan.Put(c.Processor.ack)
+	if c.Processor.req != nil {
+		c.con.requestsChan.Put(c.Processor.req)
 		c.Processor.req = nil
-		c.Processor.ack = nil
-	}()
-
-	if err = c.Stream.Send(c.Processor.ack); err != nil {
-		return fmt.Errorf("stream send ack failed, %s", err)
 	}
 
-	if c.Processor.req != nil {
-		if err = c.Stream.Send(c.Processor.req); err != nil {
-			return fmt.Errorf("stream send rqt failed, %s", err)
+	return nil
+}
+
+func sendUpstream(con *connection) {
+	for {
+		select {
+		case req := <-con.requestsChan.Get():
+			con.requestsChan.Load()
+			if err := con.Stream.Send(req); err != nil {
+				log.Errorf("send error for type url %s: %v", req.TypeUrl, err)
+				return
+			}
+		case <-con.stopCh:
+			return
 		}
 	}
+}
 
-	return nil
+func (c *Controller) Close() {
+	if c.con != nil {
+		close(c.con.stopCh)
+		_ = c.con.Stream.CloseSend()
+	}
 }

pkg/controller/ads/ads_controller_test.go

@@ -25,9 +25,11 @@ import (
 	"github.com/agiledragon/gomonkey/v2"
 	config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	service_discovery_v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 	resource_v3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/anypb"
+	"istio.io/istio/pkg/channels"
 
 	"kmesh.net/kmesh/pkg/controller/xdstest"
 )
@@ -43,7 +45,7 @@ func TestAdsStreamAdsStreamCreateAndSend(t *testing.T) {
 	defer client.Cleanup()
 
 	adsStream := Controller{
-		Stream:    client.AdsClient,
+		con:       &connection{Stream: client.AdsClient, stopCh: make(chan struct{})},
 		Processor: nil,
 	}
 
@@ -62,7 +64,7 @@ func TestAdsStreamAdsStreamCreateAndSend(t *testing.T) {
 					func(_ discoveryv3.AggregatedDiscoveryServiceClient, ctx context.Context, opts ...grpc.CallOption) (discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient, error) {
 						return client.AdsClient, nil
 					})
-				patches2.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Send",
+				patches2.ApplyMethod(reflect.TypeOf(adsStream.con.Stream), "Send",
 					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient, req *discoveryv3.DiscoveryRequest) error {
 						return errors.New("timeout")
 					})
@@ -116,7 +118,7 @@ func TestHandleAdsStream(t *testing.T) {
 	defer fakeClient.Cleanup()
 
 	adsStream := NewController()
-	adsStream.Stream = fakeClient.AdsClient
+	adsStream.con = &connection{Stream: fakeClient.AdsClient, requestsChan: channels.NewUnbounded[*service_discovery_v3.DiscoveryRequest](), stopCh: make(chan struct{})}
 
 	patches1 := gomonkey.NewPatches()
 	patches2 := gomonkey.NewPatches()
@@ -129,7 +131,7 @@ func TestHandleAdsStream(t *testing.T) {
 		{
 			name: "test1: stream Revc failed, should return error",
 			beforeFunc: func() {
-				patches1.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Recv",
+				patches1.ApplyMethod(reflect.TypeOf(adsStream.con.Stream), "Recv",
 					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient) (*discoveryv3.DiscoveryResponse, error) {
 						return nil, errors.New("failed to recv message")
 					})
@@ -140,9 +142,9 @@ func TestHandleAdsStream(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "test2: stream Send failed, should return error",
+			name: "test2: handle success, should return nil",
 			beforeFunc: func() {
-				patches1.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Recv",
+				patches1.ApplyMethod(reflect.TypeOf(adsStream.con.Stream), "Recv",
 					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient) (*discoveryv3.DiscoveryResponse, error) {
 						// create resource of rsq
 						cluster := &config_cluster_v3.Cluster{
@@ -156,38 +158,6 @@ func TestHandleAdsStream(t *testing.T) {
 							},
 						}, nil
 					})
-				patches2.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Send",
-					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient) error {
-						return errors.New("failed to send message")
-					})
-			},
-			afterFunc: func() {
-				patches1.Reset()
-				patches2.Reset()
-			},
-			wantErr: true,
-		},
-		{
-			name: "test3: handle success, should return nil",
-			beforeFunc: func() {
-				patches1.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Recv",
-					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient) (*discoveryv3.DiscoveryResponse, error) {
-						// create resource of rsq
-						cluster := &config_cluster_v3.Cluster{
-							Name: "ut-cluster",
-						}
-						anyCluster, _ := anypb.New(cluster)
-						return &discoveryv3.DiscoveryResponse{
-							TypeUrl: resource_v3.ClusterType,
-							Resources: []*anypb.Any{
-								anyCluster,
-							},
-						}, nil
-					})
-				patches2.ApplyMethod(reflect.TypeOf(adsStream.Stream), "Send",
-					func(_ discoveryv3.AggregatedDiscoveryService_StreamAggregatedResourcesClient) error {
-						return nil
-					})
 			},
 			afterFunc: func() {
 				patches1.Reset()

pkg/controller/client.go

@@ -124,14 +124,12 @@ func (c *XdsClient) handleUpstream(ctx context.Context) {
 
 			if c.mode == constants.AdsMode {
 				if err = c.AdsController.HandleAdsStream(); err != nil {
-					_ = c.AdsController.Stream.CloseSend()
 					_ = c.grpcConn.Close()
 					reconnect = true
 					continue
 				}
 			} else if c.mode == constants.WorkloadMode {
 				if err = c.WorkloadController.HandleWorkloadStream(); err != nil {
-					_ = c.WorkloadController.Stream.CloseSend()
 					_ = c.grpcConn.Close()
 					reconnect = true
 					continue
@@ -164,8 +162,8 @@ func (c *XdsClient) Run(stopCh <-chan struct{}) error {
 }
 
 func (c *XdsClient) closeStreamClient() {
-	if c.AdsController != nil && c.AdsController.Stream != nil {
-		_ = c.AdsController.Stream.CloseSend()
+	if c.AdsController != nil {
+		c.AdsController.Close()
 	}
 	if c.WorkloadController != nil && c.WorkloadController.Stream != nil {
 		_ = c.WorkloadController.Stream.CloseSend()

pkg/controller/workload/workload_controller.go

@@ -110,6 +110,7 @@ func (c *Controller) HandleWorkloadStream() error {
 	)
 
 	if rspDelta, err = c.Stream.Recv(); err != nil {
+		_ = c.Stream.CloseSend()
 		return fmt.Errorf("stream recv failed, %s", err)
 	}