Add WithOrededADS option to delta xds server

Signed-off-by: huabing zhao <zhaohuabing@gmail.com>

pkg/server/delta/v3/server.go

@@ -11,6 +11,7 @@ import (
 
 	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/config"
@@ -37,6 +38,13 @@ type Callbacks interface {
 
 var deltaErrorResponse = &cache.RawDeltaResponse{}
 
+// WithOrderedADS enables the internal flag to order responses strictly.
+func WithOrderedADS() config.XDSOption {
+	return func(o *config.Opts) {
+		o.Ordered = true
+	}
+}
+
 type server struct {
 	cache     cache.ConfigWatcher
 	callbacks Callbacks
@@ -66,6 +74,9 @@ func NewServer(ctx context.Context, config cache.ConfigWatcher, callbacks Callba
 }
 
 func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.DeltaDiscoveryRequest, defaultTypeURL string) error {
+	// create a sharedChan for the watches to send responses to
+	sharedChan := make(chan cache.DeltaResponse, types.UnknownType)
+
 	streamID := atomic.AddInt64(&s.streamCount, 1)
 
 	// streamNonce holds a unique nonce for req-resp pairs per xDS stream.
@@ -74,10 +85,24 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 	// a collection of stack allocated watches per request type
 	watches := newWatches()
 
+	// use a single go routine to send responses to the muxedResponses channel to retain resource orders
+	go func() {
+		for {
+			select {
+			case resp, more := <-sharedChan:
+				if !more {
+					return
+				}
+				watches.deltaMuxedResponses <- resp
+			}
+		}
+	}()
+
 	var node = &core.Node{}
 
 	defer func() {
 		watches.Cancel()
+		close(sharedChan)
 		if s.callbacks != nil {
 			s.callbacks.OnDeltaStreamClosed(streamID, node)
 		}
@@ -156,11 +181,15 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 				req.Node = node
 			}
 
+			ordered := false
 			// type URL is required for ADS but is implicit for any other xDS stream
 			if defaultTypeURL == resource.AnyType {
 				if req.TypeUrl == "" {
 					return status.Errorf(codes.InvalidArgument, "type URL is required for ADS")
 				}
+				if s.opts.Ordered {
+					ordered = true
+				}
 			} else if req.TypeUrl == "" {
 				req.TypeUrl = defaultTypeURL
 			}
@@ -184,16 +213,26 @@ func (s *server) processDelta(str stream.DeltaStream, reqCh <-chan *discovery.De
 			s.subscribe(req.GetResourceNamesSubscribe(), &watch.state)
 			s.unsubscribe(req.GetResourceNamesUnsubscribe(), &watch.state)
 
-			watch.responses = make(chan cache.DeltaResponse, 1)
+			if ordered {
+				// Use the shared channel for ordered responses
+				watch.responses = sharedChan
+				watch.isSharedChan = true
+			} else {
+				watch.responses = make(chan cache.DeltaResponse, 1)
+			}
 			watch.cancel = s.cache.CreateDeltaWatch(req, watch.state, watch.responses)
 			watches.deltaWatches[typeURL] = watch
 
-			go func() {
-				resp, more := <-watch.responses
-				if more {
-					watches.deltaMuxedResponses <- resp
-				}
-			}()
+			// just handle normal non-ordered responses here
+			// all ordered responses are handled in a single go routine
+			if !watch.isSharedChan {
+				go func() {
+					resp, more := <-watch.responses
+					if more {
+						watches.deltaMuxedResponses <- resp
+					}
+				}()
+			}
 		}
 	}
 }

pkg/server/delta/v3/watches.go

@@ -32,9 +32,10 @@ func (w *watches) Cancel() {
 
 // watch contains the necessary modifiables for receiving resource responses
 type watch struct {
-	responses chan cache.DeltaResponse
-	cancel    func()
-	nonce     string
+	responses    chan cache.DeltaResponse
+	isSharedChan bool // is this watch using a shared channel
+	cancel       func()
+	nonce        string
 
 	state stream.StreamState
 }
@@ -44,7 +45,7 @@ func (w *watch) Cancel() {
 	if w.cancel != nil {
 		w.cancel()
 	}
-	if w.responses != nil {
+	if w.responses != nil && !w.isSharedChan {
 		// w.responses should never be used by a producer once cancel() has been closed, so we can safely close it here
 		// This is needed to release resources taken by goroutines watching this channel
 		close(w.responses)

pkg/server/v3/delta_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v3"
 	rsrc "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/server/delta/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/server/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/test/resource/v3"
@@ -345,7 +346,9 @@ func TestDeltaAggregatedHandlers(t *testing.T) {
 		resp.recv <- r
 	}
 
-	s := server.NewServer(context.Background(), config, server.CallbackFuncs{})
+	// We create the server with the optional ordered ADS flag so we guarantee resource
+	// ordering over the stream.
+	s := server.NewServer(context.Background(), config, server.CallbackFuncs{}, delta.WithOrderedADS())
 	go func() {
 		err := s.DeltaAggregatedResources(resp)
 		assert.NoError(t, err)