Incremental Server Implementation

Makefile

@@ -12,11 +12,19 @@ PKG 		:= github.com/envoyproxy/go-control-plane
 build:
 	@go build ./pkg/... ./envoy/...
 
+.PHONY: clean
+clean:
+	@echo "--> cleaning compiled objects and binaries"
+	@go clean -tags netgo -i ./...
+	@go mod tidy
+	@rm -rf $(BINDIR)
+	@rm -rf *.log
+
 # TODO(mattklein123): See the note in TestLinearConcurrentSetWatch() for why we set -parallel here
 # This should be removed.
 .PHONY: test
 test:
-	@go test -race -v -timeout 30s -parallel 100 ./pkg/...
+	@go test -race -v -timeout 30s -count=1 -parallel 100 ./pkg/...
 
 .PHONY: cover
 cover:
@@ -33,15 +41,16 @@ examples:
 #-----------------
 #-- integration
 #-----------------
-.PHONY: $(BINDIR)/test $(BINDIR)/upstream integration integration.ads integration.xds integration.rest integration.xds.mux
+.PHONY: $(BINDIR)/test $(BINDIR)/upstream integration integration.ads integration.xds integration.rest integration.xds.mux integration.xds.delta integration.ads.delta
 
 $(BINDIR)/upstream:
 	@go build -race -o $@ internal/upstream/main.go
 
 $(BINDIR)/test:
+	@echo "Building test binary"
 	@go build -race -o $@ pkg/test/main/main.go
 
-integration: integration.xds integration.ads integration.rest integration.xds.mux
+integration: integration.xds integration.ads integration.rest integration.xds.mux integration.xds.delta integration.ads.delta
 
 integration.ads: $(BINDIR)/test $(BINDIR)/upstream
 	env XDS=ads build/integration.sh
@@ -55,6 +64,12 @@ integration.rest: $(BINDIR)/test $(BINDIR)/upstream
 integration.xds.mux: $(BINDIR)/test $(BINDIR)/upstream
 	env XDS=xds build/integration.sh -mux
 
+integration.xds.delta: $(BINDIR)/test $(BINDIR)/upstream
+	env XDS=delta build/integration.sh
+
+integration.ads.delta: $(BINDIR)/test $(BINDIR)/upstream
+	env XDS=delta-ads build/integration.sh
+
 #--------------------------------------
 #-- example xDS control plane server
 #--------------------------------------

build/integration.sh

@@ -10,7 +10,7 @@ set -o pipefail
 
 MESSAGE=$'Hi, there!\n'
 
-# Management server type. Valid values are "ads", "xds", "rest"
+# Management server type. Valid values are "ads", "xds", "rest", "delta", or "delta-ads"
 XDS=${XDS:-ads}
 
 # Number of RTDS layers.
@@ -40,4 +40,4 @@ function cleanup() {
 trap cleanup EXIT
 
 # run the test suite (which also contains the control plane)
-bin/test --xds=${XDS} --runtimes=${RUNTIMES} -debug -message="$MESSAGE" "$@"
+bin/test --xds=${XDS} --runtimes=${RUNTIMES} -debug -message="$MESSAGE" "$@"

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/golang/protobuf v1.4.3
 	github.com/google/go-cmp v0.5.0
 	github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4
-	github.com/stretchr/testify v1.5.1
+	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/proto/otlp v0.7.0
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013
 	google.golang.org/grpc v1.36.0

go.sum

@@ -46,8 +46,9 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 h1:gQz4mCb
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 go.opentelemetry.io/proto/otlp v0.7.0 h1:rwOQPCuKAKmwGKq2aVNnYIibI6wnV7EvzgfTCzcdGg8=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -111,7 +112,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.3 h1:fvjTMHxHEw/mxHbtzPi3JCcKXQRAnQTBRo6YCJSVHKI=
 gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

pkg/cache/v3/cache.go

@@ -63,7 +63,7 @@ type ConfigWatcher interface {
 	//
 	// Cancel is an optional function to release resources in the producer. If
 	// provided, the consumer may call this function multiple times.
-	CreateDeltaWatch(*DeltaRequest, *stream.StreamState) (value chan DeltaResponse, cancel func())
+	CreateDeltaWatch(*DeltaRequest, stream.StreamState) (value chan DeltaResponse, cancel func())
 }
 
 // ConfigFetcher fetches configuration resources from cache

pkg/cache/v3/delta.go

@@ -21,8 +21,8 @@ import (
 )
 
 // Respond to a delta watch with the provided snapshot value. If the response is nil, there has been no state change.
-func respondDelta(request *DeltaRequest, value chan DeltaResponse, st *stream.StreamState, snapshot Snapshot, log log.Logger) *RawDeltaResponse {
-	resp, err := createDeltaResponse(request, st, snapshot)
+func respondDelta(request *DeltaRequest, value chan DeltaResponse, state stream.StreamState, snapshot Snapshot, log log.Logger) *RawDeltaResponse {
+	resp, err := createDeltaResponse(request, state, snapshot, log)
 	if err != nil {
 		if log != nil {
 			log.Errorf("Error creating delta response: %v", err)
@@ -34,15 +34,15 @@ func respondDelta(request *DeltaRequest, value chan DeltaResponse, st *stream.St
 	if len(resp.Resources) > 0 || len(resp.RemovedResources) > 0 {
 		if log != nil {
 			log.Debugf("node: %s, sending delta response with resources: %v removed resources %v wildcard: %t",
-				request.GetNode().GetId(), resp.Resources, resp.RemovedResources, st.IsWildcard)
+				request.GetNode().GetId(), resp.Resources, resp.RemovedResources, state.Wildcard)
 		}
 		value <- resp
 		return resp
 	}
 	return nil
 }
 
-func createDeltaResponse(req *DeltaRequest, st *stream.StreamState, snapshot Snapshot) (*RawDeltaResponse, error) {
+func createDeltaResponse(req *DeltaRequest, state stream.StreamState, snapshot Snapshot, log log.Logger) (*RawDeltaResponse, error) {
 	resources := snapshot.GetResources((req.TypeUrl))
 
 	// variables to build our response with
@@ -51,20 +51,21 @@ func createDeltaResponse(req *DeltaRequest, st *stream.StreamState, snapshot Sna
 	toRemove := make([]string, 0)
 
 	// If we are handling a wildcard request, we want to respond with all resources
-	if st.IsWildcard {
+	switch {
+	case state.Wildcard:
 		for name, r := range resources {
 			// Since we've already precomputed the version hashes of the new snapshot,
 			// we can just set it here to be used for comparison later
 			version := snapshot.GetVersionMap()[req.TypeUrl][name]
 			nextVersionMap[name] = version
-			prevVersion, found := st.ResourceVersions[name]
+			prevVersion, found := state.ResourceVersions[name]
 			if !found || (prevVersion != nextVersionMap[name]) {
 				filtered = append(filtered, r)
 			}
 		}
-	} else {
+	default:
 		// Reply only with the requested resources
-		for name, prevVersion := range st.ResourceVersions {
+		for name, prevVersion := range state.ResourceVersions {
 			if r, ok := resources[name]; ok {
 				nextVersion := snapshot.GetVersionMap()[req.TypeUrl][name]
 				if prevVersion != nextVersion {
@@ -76,7 +77,7 @@ func createDeltaResponse(req *DeltaRequest, st *stream.StreamState, snapshot Sna
 	}
 
 	// Compute resources for removal regardless of the request type
-	for name := range st.ResourceVersions {
+	for name := range state.ResourceVersions {
 		if _, ok := resources[name]; !ok {
 			toRemove = append(toRemove, name)
 		}

pkg/cache/v3/delta_test.go

@@ -27,14 +27,14 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) {
 			},
 			TypeUrl:                typ,
 			ResourceNamesSubscribe: names[typ],
-		}, &stream.StreamState{IsWildcard: true, ResourceVersions: nil})
+		}, stream.StreamState{ResourceVersions: nil, Wildcard: true})
 	}
 
 	if err := c.SetSnapshot(key, snapshot); err != nil {
 		t.Fatal(err)
 	}
 
-	vm := make(map[string]map[string]string)
+	versionMap := make(map[string]map[string]string)
 	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			select {
@@ -43,7 +43,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) {
 					t.Errorf("got resources %v, want %v", out.(*cache.RawDeltaResponse).Resources, snapshot.GetResources(typ))
 				}
 				vMap := out.GetNextVersionMap()
-				vm[typ] = vMap
+				versionMap[typ] = vMap
 			case <-time.After(time.Second):
 				t.Fatal("failed to receive snapshot response")
 			}
@@ -59,7 +59,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) {
 			},
 			TypeUrl:                typ,
 			ResourceNamesSubscribe: names[typ],
-		}, &stream.StreamState{IsWildcard: false, ResourceVersions: vm[typ]})
+		}, stream.StreamState{ResourceVersions: versionMap[typ]})
 	}
 
 	if count := c.GetStatusInfo(key).GetNumDeltaWatches(); count != len(testTypes) {
@@ -83,7 +83,7 @@ func TestSnapshotCacheDeltaWatch(t *testing.T) {
 			t.Fatalf("got resources %v, want %v", out.(*cache.RawDeltaResponse).Resources, snapshot2.GetResources(rsrc.EndpointType))
 		}
 		vMap := out.GetNextVersionMap()
-		vm[testTypes[0]] = vMap
+		versionMap[testTypes[0]] = vMap
 	case <-time.After(time.Second):
 		t.Fatal("failed to receive snapshot response")
 	}
@@ -101,7 +101,7 @@ func TestDeltaRemoveResources(t *testing.T) {
 				Id: "node",
 			},
 			TypeUrl: typ,
-		}, &stream.StreamState{IsWildcard: true, ResourceVersions: nil})
+		}, stream.StreamState{ResourceVersions: make(map[string]string), Wildcard: true})
 	}
 
 	if err := c.SetSnapshot(key, snapshot); err != nil {
@@ -132,7 +132,7 @@ func TestDeltaRemoveResources(t *testing.T) {
 				Id: "node",
 			},
 			TypeUrl: typ,
-		}, &stream.StreamState{IsWildcard: true, ResourceVersions: versionMap[typ]})
+		}, stream.StreamState{ResourceVersions: versionMap[typ], Wildcard: true})
 	}
 
 	if count := c.GetStatusInfo(key).GetNumDeltaWatches(); count != len(testTypes) {
@@ -187,7 +187,7 @@ func TestConcurrentSetDeltaWatch(t *testing.T) {
 						},
 						TypeUrl:                rsrc.EndpointType,
 						ResourceNamesSubscribe: []string{clusterName},
-					}, &stream.StreamState{IsWildcard: true, ResourceVersions: nil})
+					}, stream.NewStreamState())
 				}
 			})
 		}(i)
@@ -203,7 +203,7 @@ func TestSnapshotCacheDeltaWatchCancel(t *testing.T) {
 			},
 			TypeUrl:                typ,
 			ResourceNamesSubscribe: names[typ],
-		}, &stream.StreamState{IsWildcard: true, ResourceVersions: nil})
+		}, stream.NewStreamState())
 
 		// Cancel the watch
 		cancel()

pkg/cache/v3/linear.go

@@ -239,7 +239,7 @@ func (cache *LinearCache) CreateWatch(request *Request) (chan Response, func())
 	}
 }
 
-func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, st *stream.StreamState) (chan DeltaResponse, func()) {
+func (cache *LinearCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState) (chan DeltaResponse, func()) {
 	return nil, nil
 }
 

pkg/cache/v3/mux.go

@@ -47,7 +47,7 @@ func (mux *MuxCache) CreateWatch(request *Request) (chan Response, func()) {
 	return cache.CreateWatch(request)
 }
 
-func (cache *MuxCache) CreateDeltaWatch(request *DeltaRequest, st *stream.StreamState) (chan DeltaResponse, func()) {
+func (cache *MuxCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState) (chan DeltaResponse, func()) {
 	return nil, nil
 }
 

pkg/cache/v3/simple.go

@@ -392,7 +392,7 @@ func createResponse(request *Request, resources map[string]types.ResourceWithTtl
 }
 
 // CreateDeltaWatch returns a watch for a delta xDS request which implements the Simple SnapshotCache.
-func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, st *stream.StreamState) (chan DeltaResponse, func()) {
+func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream.StreamState) (chan DeltaResponse, func()) {
 	nodeID := cache.hash.ID(request.Node)
 	t := request.GetTypeUrl()
 
@@ -424,17 +424,17 @@ func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, st *stream.S
 			cache.log.Errorf("failed to compute version for snapshot resources inline, waiting for next snapshot update")
 			delayedResponse = true
 		}
-		delayedResponse = respondDelta(request, value, st, snapshot, cache.log) == nil
+		delayedResponse = respondDelta(request, value, state, snapshot, cache.log) == nil
 	}
 
 	if delayedResponse {
 		watchID := cache.nextDeltaWatchID()
 		if cache.log != nil {
 			cache.log.Infof("open delta watch ID:%d for %s Resources:%v from nodeID: %q, system version %q", watchID,
-				t, st.ResourceVersions, nodeID, snapshot.GetVersion(t))
+				t, state.ResourceVersions, nodeID, snapshot.GetVersion(t))
 		}
 
-		info.SetDeltaResponseWatch(watchID, DeltaResponseWatch{Request: request, Response: value, StreamState: st})
+		info.SetDeltaResponseWatch(watchID, DeltaResponseWatch{Request: request, Response: value, StreamState: state})
 
 		return value, cache.cancelDeltaWatch(nodeID, watchID)
 	}

pkg/cache/v3/status.go

@@ -105,7 +105,7 @@ type DeltaResponseWatch struct {
 	Response chan DeltaResponse
 
 	// VersionMap for the stream
-	StreamState *stream.StreamState
+	StreamState stream.StreamState
 }
 
 // newStatusInfo initializes a status info data structure.
@@ -148,8 +148,8 @@ func (info *statusInfo) GetLastDeltaWatchRequestTime() time.Time {
 	return info.lastDeltaWatchRequestTime
 }
 
-// GetDeltaVersionMap will pull the version map out of a specific watch
-func (info *statusInfo) GetDeltaStreamState(watchID int64) *stream.StreamState {
+// GetDeltaStreamState will pull the stream state with the version map out of a specific watch
+func (info *statusInfo) GetDeltaStreamState(watchID int64) stream.StreamState {
 	info.mu.RLock()
 	defer info.mu.RUnlock()
 	return info.deltaWatches[watchID].StreamState