add cache ordering on ads mode

Signed-off-by: Alec Holmes <[email protected]>

go.sum

@@ -55,6 +55,8 @@ github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc h1:PYXxkRUBGUMa5xgMVMD
 github.com/cncf/xds/go v0.0.0-20220314180256-7f1daf1720fc/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=

pkg/cache/types/types.go

@@ -35,8 +35,8 @@ type ResponseType int
 const (
 	Endpoint ResponseType = iota
 	Cluster
-	Route
 	ScopedRoute
+	Route
 	VirtualHost
 	Listener
 	Secret

pkg/cache/v3/order.go

@@ -0,0 +1,24 @@
+package cache
+
+// Key is an internal sorting data structure we can use to
+// order responses by Type and their associated watch IDs.
+type key struct {
+	ID      int64
+	TypeURL string
+}
+
+// Keys implements Go's sorting.Sort interface
+type keys []key
+
+func (k keys) Len() int {
+	return len(k)
+}
+
+// Less compares the typeURL and determines what order things should be sent.
+func (k keys) Less(i, j int) bool {
+	return GetResponseType(k[i].TypeURL) > GetResponseType(k[j].TypeURL)
+}
+
+func (k keys) Swap(i, j int) {
+	k[i], k[j] = k[j], k[i]
+}

pkg/cache/v3/order_test.go

@@ -0,0 +1,49 @@
+package cache
+
+import (
+	"errors"
+	"sort"
+	"testing"
+
+	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+)
+
+func TestOrderKeys(t *testing.T) {
+	unorderedKeys := keys{
+		{
+			ID:      1,
+			TypeURL: resource.EndpointType,
+		},
+		{
+			ID:      2,
+			TypeURL: resource.ClusterType,
+		},
+		{
+			ID:      4,
+			TypeURL: resource.ListenerType,
+		},
+		{
+			ID:      3,
+			TypeURL: resource.RouteType,
+		},
+		{
+			ID:      5,
+			TypeURL: resource.ScopedRouteType,
+		},
+	}
+
+	orderedKeys := unorderedKeys
+	sort.Sort(orderedKeys)
+
+	if !sort.IsSorted(orderedKeys) {
+		t.Fatal(errors.New("failed to sort keys"))
+	}
+
+	// Ordering:
+	// === RUN   TestOrderKeys
+	// order_test.go:43: {ID:4 TypeURL:type.googleapis.com/envoy.config.listener.v3.Listener}
+	// order_test.go:43: {ID:3 TypeURL:type.googleapis.com/envoy.config.route.v3.RouteConfiguration}
+	// order_test.go:43: {ID:5 TypeURL:type.googleapis.com/envoy.config.route.v3.ScopedRouteConfiguration}
+	// order_test.go:43: {ID:2 TypeURL:type.googleapis.com/envoy.config.cluster.v3.Cluster}
+	// order_test.go:43: {ID:1 TypeURL:type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment}
+}

pkg/cache/v3/simple.go

@@ -228,7 +228,14 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
 	if info, ok := cache.status[node]; ok {
 		info.mu.Lock()
 		defer info.mu.Unlock()
-		for id, watch := range info.watches {
+
+		// If ADS is enabled we need to order response watches so we guarantee
+		// sending them in the correct order since Go's default implementation
+		// of maps are randomized order when ranged over.
+		info.orderResponseWatches(cache.ads)
+
+		for _, key := range info.orderedWatches {
+			watch := info.watches[key.ID]
 			version := snapshot.GetVersion(watch.Request.TypeUrl)
 			if version != watch.Request.VersionInfo {
 				cache.log.Debugf("respond open watch %d %s%v with new version %q", id, watch.Request.TypeUrl, watch.Request.ResourceNames, version)
@@ -240,7 +247,7 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
 				}
 
 				// discard the watch
-				delete(info.watches, id)
+				delete(info.watches, key.ID)
 			}
 		}
 

pkg/cache/v3/status.go

@@ -15,6 +15,7 @@
 package cache
 
 import (
+	"sort"
 	"sync"
 	"time"
 
@@ -65,7 +66,9 @@ type statusInfo struct {
 	node *core.Node
 
 	// watches are indexed channels for the response watches and the original requests.
-	watches map[int64]ResponseWatch
+	// watches map[int64]ResponseWatch
+	watches        map[int64]ResponseWatch
+	orderedWatches keys
 
 	// deltaWatches are indexed channels for the delta response watches and the original requests
 	deltaWatches map[int64]DeltaResponseWatch
@@ -105,9 +108,10 @@ type DeltaResponseWatch struct {
 // newStatusInfo initializes a status info data structure.
 func newStatusInfo(node *core.Node) *statusInfo {
 	out := statusInfo{
-		node:         node,
-		watches:      make(map[int64]ResponseWatch),
-		deltaWatches: make(map[int64]DeltaResponseWatch),
+		node:           node,
+		watches:        make(map[int64]ResponseWatch),
+		orderedWatches: make(keys, 0),
+		deltaWatches:   make(map[int64]DeltaResponseWatch),
 	}
 	return &out
 }
@@ -155,3 +159,24 @@ func (info *statusInfo) setDeltaResponseWatch(id int64, drw DeltaResponseWatch)
 	defer info.mu.Unlock()
 	info.deltaWatches[id] = drw
 }
+
+// right now this is a noop but will need to set the map in an order
+func (info *statusInfo) orderResponseWatches(order bool) {
+	// 0 out our watch list cause watches get deleted in the map.
+	info.orderedWatches = make(keys, 0)
+
+	// This runs in O(n) which could become problematic when we have an extrclemely high watch count.
+	// TODO(alec): revisit this and optimize for speed.
+	for id, watch := range info.watches {
+		info.orderedWatches = append(info.orderedWatches, key{
+			ID:      id,
+			TypeURL: watch.Request.TypeUrl,
+		})
+	}
+
+	// Sort our list which we can use in the SetSnapshot functions.
+	// This is only run when we enable ADS on the cache.
+	if order {
+		sort.Sort(info.orderedWatches)
+	}
+}