sotw: Ordered ADS mode (#544)

* initial ADS code path split so we can order responses for target modes
* implement functional opts pattern to support xds options
* multiplex responses over a single channel rather than selecting over dynamic channels in ordered ads
* add cache ordering on ads mode

Signed-off-by: Alec Holmes <alecholmez@me.com>

Author: alecholmez

Dockerfile.ci

@@ -1,4 +1,3 @@
 FROM golang:1.19
 
-# TODO(mattklein123): Update this to envoy-dev:latest and fix tests.
 COPY --from=envoyproxy/envoy-dev:latest /usr/local/bin/envoy /usr/local/bin/envoy

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/prometheus/client_model v0.4.0
 	github.com/stretchr/testify v1.8.3
 	go.opentelemetry.io/proto/otlp v0.19.0
-	google.golang.org/genproto v0.0.0-20230526203410-71b5a4ffd15e
 	google.golang.org/genproto/googleapis/api v0.0.0-20230526203410-71b5a4ffd15e
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230526203410-71b5a4ffd15e
 	google.golang.org/grpc v1.55.0
@@ -20,9 +19,13 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/kr/pretty v0.3.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.9.0 // indirect
 	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/sys v0.8.0 // indirect
 	golang.org/x/text v0.9.0 // indirect
+	google.golang.org/genproto v0.0.0-20230526203410-71b5a4ffd15e // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -38,11 +38,15 @@ func (e SkipFetchError) Error() string {
 // ResponseType enumeration of supported response types
 type ResponseType int
 
+// NOTE: The order of this enum MATTERS!
+// https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#aggregated-discovery-service
+// ADS expects things to be returned in a specific order.
+// See the following issue for details: https://github.com/envoyproxy/go-control-plane/issues/526
 const (
 	Endpoint ResponseType = iota
 	Cluster
-	Route
 	ScopedRoute
+	Route
 	VirtualHost
 	Listener
 	Secret

pkg/cache/types/types.go

@@ -44,6 +44,10 @@ type DeltaRequest = discovery.DeltaDiscoveryRequest
 // ConfigWatcher implementation must be thread-safe.
 type ConfigWatcher interface {
 	// CreateWatch returns a new open watch from a non-empty request.
+	// This is the entrypoint to propagate configuration changes the
+	// provided Response channel. State from the gRPC server is utilized
+	// to make sure consuming cache implementations can see what the server has sent to clients.
+	//
 	// An individual consumer normally issues a single open watch by each type URL.
 	//
 	// The provided channel produces requested resources as responses, once they are available.
@@ -53,6 +57,9 @@ type ConfigWatcher interface {
 	CreateWatch(*Request, stream.StreamState, chan Response) (cancel func())
 
 	// CreateDeltaWatch returns a new open incremental xDS watch.
+	// This is the entrypoint to propagate configuration changes the
+	// provided DeltaResponse channel. State from the gRPC server is utilized
+	// to make sure consuming cache implementations can see what the server has sent to clients.
 	//
 	// The provided channel produces requested resources as responses, or spontaneous updates in accordance
 	// with the incremental xDS specification.

pkg/cache/v3/cache.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.go

@@ -0,0 +1,72 @@
+package cache
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"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,
+		},
+	}
+	expected := keys{
+		{
+			ID:      4,
+			TypeURL: resource.ListenerType,
+		},
+		{
+			ID:      3,
+			TypeURL: resource.RouteType,
+		},
+		{
+			ID:      5,
+			TypeURL: resource.ScopedRouteType,
+		},
+		{
+			ID:      2,
+			TypeURL: resource.ClusterType,
+		},
+		{
+			ID:      1,
+			TypeURL: resource.EndpointType,
+		},
+	}
+
+	orderedKeys := unorderedKeys
+	sort.Sort(orderedKeys)
+
+	assert.True(t, sort.IsSorted(orderedKeys))
+	assert.NotEqual(t, unorderedKeys, &orderedKeys)
+	assert.Equal(t, expected, orderedKeys)
+
+	// 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/order_test.go

@@ -220,7 +220,7 @@ func (cache *snapshotCache) sendHeartbeats(ctx context.Context, node string) {
 	}
 }
 
-// SetSnapshotCacheContext updates a snapshot for a node.
+// SetSnapshotCache updates a snapshot for a node.
 func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapshot ResourceSnapshot) error {
 	cache.mu.Lock()
 	defer cache.mu.Unlock()
@@ -232,20 +232,41 @@ 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 {
+
+		// responder callback for SOTW watches
+		respond := func(watch ResponseWatch, id int64) error {
 			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)
-
 				resources := snapshot.GetResourcesAndTTL(watch.Request.TypeUrl)
 				err := cache.respond(ctx, watch.Request, watch.Response, resources, version, false)
 				if err != nil {
 					return err
 				}
-
 				// discard the watch
 				delete(info.watches, id)
 			}
+			return nil
+		}
+
+		// If ADS is enabled we need to order response watches so we guarantee
+		// sending them in the correct order. Go's default implementation
+		// of maps are randomized order when ranged over.
+		if cache.ads {
+			info.orderResponseWatches()
+			for _, key := range info.orderedWatches {
+				err := respond(info.watches[key.ID], key.ID)
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			for id, watch := range info.watches {
+				err := respond(watch, id)
+				if err != nil {
+					return err
+				}
+			}
 		}
 
 		// We only calculate version hashes when using delta. We don't
@@ -258,7 +279,8 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
 			}
 		}
 
-		// process our delta watches
+		// this won't run if there are no delta watches
+		// to process.
 		for id, watch := range info.deltaWatches {
 			res, err := cache.respondDelta(
 				ctx,
@@ -281,7 +303,7 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
 	return nil
 }
 
-// GetSnapshots gets the snapshot for a node, and returns an error if not found.
+// GetSnapshot gets the snapshot for a node, and returns an error if not found.
 func (cache *snapshotCache) GetSnapshot(node string) (ResourceSnapshot, error) {
 	cache.mu.RLock()
 	defer cache.mu.RUnlock()
@@ -341,7 +363,6 @@ func (cache *snapshotCache) CreateWatch(request *Request, streamState stream.Str
 	info.mu.Unlock()
 
 	var version string
-
 	snapshot, exists := cache.snapshots[nodeID]
 	if exists {
 		version = snapshot.GetVersion(request.TypeUrl)

pkg/cache/v3/simple.go

@@ -345,7 +345,7 @@ func TestConcurrentSetWatch(t *testing.T) {
 		i := i
 		t.Run(fmt.Sprintf("worker%d", i), func(t *testing.T) {
 			t.Parallel()
-			id := fmt.Sprintf("%d", i%2)
+			id := t.Name()
 			value := make(chan cache.Response, 1)
 			if i < 25 {
 				snap := cache.Snapshot{}
@@ -542,6 +542,7 @@ func (s *singleResourceSnapshot) GetResources(typeURL string) map[string]types.R
 	if typeURL != s.typeurl {
 		return nil
 	}
+
 	return map[string]types.Resource{
 		s.name: s.resource,
 	}

pkg/cache/v3/simple_test.go

@@ -15,6 +15,7 @@
 package cache
 
 import (
+	"sort"
 	"sync"
 	"time"
 
@@ -65,7 +66,8 @@ 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
+	orderedWatches keys
 
 	// deltaWatches are indexed channels for the delta response watches and the original requests
 	deltaWatches map[int64]DeltaResponseWatch
@@ -105,9 +107,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 +158,22 @@ func (info *statusInfo) setDeltaResponseWatch(id int64, drw DeltaResponseWatch)
 	defer info.mu.Unlock()
 	info.deltaWatches[id] = drw
 }
+
+// orderResponseWatches will track a list of watch keys and order them if
+// true is passed.
+func (info *statusInfo) orderResponseWatches() {
+	info.orderedWatches = make(keys, len(info.watches))
+
+	var index int
+	for id, watch := range info.watches {
+		info.orderedWatches[index] = key{
+			ID:      id,
+			TypeURL: watch.Request.TypeUrl,
+		}
+		index++
+	}
+
+	// Sort our list which we can use in the SetSnapshot functions.
+	// This is only run when we enable ADS on the cache.
+	sort.Sort(info.orderedWatches)
+}

pkg/cache/v3/status.go

@@ -32,14 +32,11 @@ func (log logger) Warnf(format string, args ...interface{})  { log.t.Logf(format
 func (log logger) Errorf(format string, args ...interface{}) { log.t.Logf(format, args...) }
 
 func TestTTLResponse(t *testing.T) {
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	snapshotCache := cache.NewSnapshotCacheWithHeartbeating(ctx, false, cache.IDHash{}, logger{t: t}, time.Second)
-
 	server := server.NewServer(ctx, snapshotCache, nil)
-
 	grpcServer := grpc.NewServer()
 	endpointservice.RegisterEndpointDiscoveryServiceServer(grpcServer, server)
 
@@ -53,8 +50,8 @@ func TestTTLResponse(t *testing.T) {
 
 	conn, err := grpc.Dial(":9999", grpc.WithTransportCredentials(insecure.NewCredentials()))
 	assert.NoError(t, err)
-	client := endpointservice.NewEndpointDiscoveryServiceClient(conn)
 
+	client := endpointservice.NewEndpointDiscoveryServiceClient(conn)
 	sclient, err := client.StreamEndpoints(ctx)
 	assert.NoError(t, err)
 
@@ -75,16 +72,16 @@ func TestTTLResponse(t *testing.T) {
 			TTL:      &oneSecond,
 		}},
 	})
+
 	err = snapshotCache.SetSnapshot(context.Background(), "test", snap)
 	assert.NoError(t, err)
 
 	timeout := time.NewTimer(5 * time.Second)
-
 	awaitResponse := func() *envoy_service_discovery_v3.DiscoveryResponse {
 		t.Helper()
 		doneCh := make(chan *envoy_service_discovery_v3.DiscoveryResponse)
-		go func() {
 
+		go func() {
 			r, err := sclient.Recv()
 			assert.NoError(t, err)
 

pkg/integration/ttl_integration_test.go

@@ -1,3 +1,4 @@
+//nolint:all
 package log
 
 // DefaultLogger is enabled when no consuming clients provide

pkg/log/default.go

@@ -22,13 +22,11 @@ import (
 )
 
 func ExampleLoggerFuncs() {
-	logger := log.Logger{}
-
 	xdsLogger := LoggerFuncs{
-		DebugFunc: logger.Printf,
-		InfoFunc:  logger.Printf,
-		WarnFunc:  logger.Printf,
-		ErrorFunc: logger.Printf,
+		DebugFunc: log.Printf,
+		InfoFunc:  log.Printf,
+		WarnFunc:  log.Printf,
+		ErrorFunc: log.Printf,
 	}
 
 	xdsLogger.Debugf("debug")