cache: fix per-response caching and switch to pointers for requests and responses (#340)

* cache: fix per response caching and switch to pointers for response

Signed-off-by: Joshua Rutherford <[email protected]>

* cache: ensure cached discovery response logic is platform agnostic

Signed-off-by: Joshua Rutherford <[email protected]>

Co-authored-by: Joshua Rutherford <[email protected]>

Author: Joshua Rutherford

build/integration.sh

@@ -37,8 +37,8 @@ echo Envoy log: ${ENVOY_LOG}
 ENVOY_PID=$!
 
 function cleanup() {
-  kill ${ENVOY_PID}
-  kill ${UPSTREAM_PID}
+  kill ${ENVOY_PID} ${UPSTREAM_PID}
+  wait ${ENVOY_PID} ${UPSTREAM_PID} 2> /dev/null || true
 }
 trap cleanup EXIT
 

pkg/cache/v2/cache.go

@@ -18,6 +18,7 @@ package cache
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 
 	discovery "github.com/envoyproxy/go-control-plane/envoy/api/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
@@ -43,15 +44,15 @@ type ConfigWatcher interface {
 	//
 	// Cancel is an optional function to release resources in the producer. If
 	// provided, the consumer may call this function multiple times.
-	CreateWatch(Request) (value chan Response, cancel func())
+	CreateWatch(*Request) (value chan Response, cancel func())
 }
 
 // Cache is a generic config cache with a watcher.
 type Cache interface {
 	ConfigWatcher
 
 	// Fetch implements the polling method of the config cache using a non-empty request.
-	Fetch(context.Context, Request) (Response, error)
+	Fetch(context.Context, *Request) (Response, error)
 }
 
 // Response is a wrapper around Envoy's DiscoveryResponse.
@@ -70,7 +71,7 @@ type Response interface {
 // be included in the final Discovery Response.
 type RawResponse struct {
 	// Request is the original request.
-	Request discovery.DiscoveryRequest
+	Request *discovery.DiscoveryRequest
 
 	// Version of the resources as tracked by the cache for the given type.
 	// Proxy responds with this version as an acknowledgement.
@@ -79,21 +80,16 @@ type RawResponse struct {
 	// Resources to be included in the response.
 	Resources []types.Resource
 
-	// isResourceMarshaled indicates whether the resources have been marshaled.
-	// This is internally maintained by go-control-plane to prevent future
-	// duplication in marshaling efforts.
-	isResourceMarshaled bool
-
-	// marshaledResponse holds the serialized discovery response.
-	marshaledResponse *discovery.DiscoveryResponse
+	// marshaledResponse holds an atomic reference to the serialized discovery response.
+	marshaledResponse atomic.Value
 }
 
 var _ Response = &RawResponse{}
 
 // PassthroughResponse is a pre constructed xDS response that need not go through marshalling transformations.
 type PassthroughResponse struct {
 	// Request is the original request.
-	Request discovery.DiscoveryRequest
+	Request *discovery.DiscoveryRequest
 
 	// The discovery response that needs to be sent as is, without any marshalling transformations.
 	DiscoveryResponse *discovery.DiscoveryResponse
@@ -104,55 +100,59 @@ var _ Response = &PassthroughResponse{}
 // GetDiscoveryResponse performs the marshalling the first time its called and uses the cached response subsequently.
 // This is necessary because the marshalled response does not change across the calls.
 // This caching behavior is important in high throughput scenarios because grpc marshalling has a cost and it drives the cpu utilization under load.
-func (r RawResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, error) {
-	if r.isResourceMarshaled {
-		return r.marshaledResponse, nil
-	}
+func (r *RawResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, error) {
+
+	marshaledResponse := r.marshaledResponse.Load()
+
+	if marshaledResponse == nil {
 
-	marshaledResources := make([]*any.Any, len(r.Resources))
+		marshaledResources := make([]*any.Any, len(r.Resources))
 
-	for i, resource := range r.Resources {
-		marshaledResource, err := MarshalResource(resource)
-		if err != nil {
-			return nil, err
+		for i, resource := range r.Resources {
+			marshaledResource, err := MarshalResource(resource)
+			if err != nil {
+				return nil, err
+			}
+			marshaledResources[i] = &any.Any{
+				TypeUrl: r.Request.TypeUrl,
+				Value:   marshaledResource,
+			}
 		}
-		marshaledResources[i] = &any.Any{
-			TypeUrl: r.Request.TypeUrl,
-			Value:   marshaledResource,
+
+		marshaledResponse = &discovery.DiscoveryResponse{
+			VersionInfo: r.Version,
+			Resources:   marshaledResources,
+			TypeUrl:     r.Request.TypeUrl,
 		}
-	}
 
-	r.isResourceMarshaled = true
+		r.marshaledResponse.Store(marshaledResponse)
+	}
 
-	return &discovery.DiscoveryResponse{
-		VersionInfo: r.Version,
-		Resources:   marshaledResources,
-		TypeUrl:     r.Request.TypeUrl,
-	}, nil
+	return marshaledResponse.(*discovery.DiscoveryResponse), nil
 }
 
 // GetRequest returns the original Discovery Request.
-func (r RawResponse) GetRequest() *discovery.DiscoveryRequest {
-	return &r.Request
+func (r *RawResponse) GetRequest() *discovery.DiscoveryRequest {
+	return r.Request
 }
 
 // GetVersion returns the response version.
-func (r RawResponse) GetVersion() (string, error) {
+func (r *RawResponse) GetVersion() (string, error) {
 	return r.Version, nil
 }
 
 // GetDiscoveryResponse returns the final passthrough Discovery Response.
-func (r PassthroughResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, error) {
+func (r *PassthroughResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, error) {
 	return r.DiscoveryResponse, nil
 }
 
 // GetRequest returns the original Discovery Request
-func (r PassthroughResponse) GetRequest() *discovery.DiscoveryRequest {
-	return &r.Request
+func (r *PassthroughResponse) GetRequest() *discovery.DiscoveryRequest {
+	return r.Request
 }
 
 // GetVersion returns the response version.
-func (r PassthroughResponse) GetVersion() (string, error) {
+func (r *PassthroughResponse) GetVersion() (string, error) {
 	if r.DiscoveryResponse != nil {
 		return r.DiscoveryResponse.VersionInfo, nil
 	}

pkg/cache/v2/cache_test.go

@@ -20,7 +20,7 @@ const (
 func TestResponseGetDiscoveryResponse(t *testing.T) {
 	routes := []types.Resource{&route.RouteConfiguration{Name: resourceName}}
 	resp := cache.RawResponse{
-		Request:   discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
+		Request:   &discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
 		Version:   "v",
 		Resources: routes,
 	}
@@ -30,6 +30,10 @@ func TestResponseGetDiscoveryResponse(t *testing.T) {
 	assert.Equal(t, discoveryResponse.VersionInfo, resp.Version)
 	assert.Equal(t, len(discoveryResponse.Resources), 1)
 
+	cachedResponse, err := resp.GetDiscoveryResponse()
+	assert.Nil(t, err)
+	assert.Same(t, discoveryResponse, cachedResponse)
+
 	r := &route.RouteConfiguration{}
 	err = ptypes.UnmarshalAny(discoveryResponse.Resources[0], r)
 	assert.Nil(t, err)
@@ -46,7 +50,7 @@ func TestPassthroughResponseGetDiscoveryResponse(t *testing.T) {
 		VersionInfo: "v",
 	}
 	resp := cache.PassthroughResponse{
-		Request:           discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
+		Request:           &discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
 		DiscoveryResponse: dr,
 	}
 

pkg/cache/v2/linear.go

@@ -84,8 +84,8 @@ func (cache *LinearCache) respond(value chan Response, staleResources []string)
 			}
 		}
 	}
-	value <- RawResponse{
-		Request:   Request{TypeUrl: cache.typeURL},
+	value <- &RawResponse{
+		Request:   &Request{TypeUrl: cache.typeURL},
 		Resources: resources,
 		Version:   strconv.FormatUint(cache.version, 10),
 	}
@@ -141,7 +141,7 @@ func (cache *LinearCache) DeleteResource(name string) error {
 	return nil
 }
 
-func (cache *LinearCache) CreateWatch(request Request) (chan Response, func()) {
+func (cache *LinearCache) CreateWatch(request *Request) (chan Response, func()) {
 	value := make(chan Response, 1)
 	if request.TypeUrl != cache.typeURL {
 		close(value)
@@ -207,7 +207,7 @@ func (cache *LinearCache) CreateWatch(request Request) (chan Response, func()) {
 	}
 }
 
-func (cache *LinearCache) Fetch(ctx context.Context, request Request) (Response, error) {
+func (cache *LinearCache) Fetch(ctx context.Context, request *Request) (Response, error) {
 	return nil, errors.New("not implemented")
 }
 

pkg/cache/v2/linear_test.go

@@ -71,9 +71,9 @@ func mustBlock(t *testing.T, w <-chan Response) {
 
 func TestLinearInitialResources(t *testing.T) {
 	c := NewLinearCache(testType, map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})
-	w, _ := c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType})
+	w, _ := c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType})
 	verifyResponse(t, w, "0", 1)
-	w, _ = c.CreateWatch(Request{TypeUrl: testType})
+	w, _ = c.CreateWatch(&Request{TypeUrl: testType})
 	verifyResponse(t, w, "0", 2)
 }
 
@@ -84,7 +84,7 @@ func TestLinearCornerCases(t *testing.T) {
 		t.Error("expected error on nil resource")
 	}
 	// create an incorrect type URL request
-	w, _ := c.CreateWatch(Request{TypeUrl: "test"})
+	w, _ := c.CreateWatch(&Request{TypeUrl: "test"})
 	select {
 	case _, more := <-w:
 		if more {
@@ -99,9 +99,9 @@ func TestLinearBasic(t *testing.T) {
 	c := NewLinearCache(testType, nil)
 
 	// Create watches before a resource is ready
-	w1, _ := c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
+	w1, _ := c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
 	mustBlock(t, w1)
-	w, _ := c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "0"})
+	w, _ := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"})
 	mustBlock(t, w)
 	checkWatchCount(t, c, "a", 2)
 	checkWatchCount(t, c, "b", 1)
@@ -112,44 +112,44 @@ func TestLinearBasic(t *testing.T) {
 	verifyResponse(t, w, "1", 1)
 
 	// Request again, should get same response
-	w, _ = c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
+	w, _ = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
 	checkWatchCount(t, c, "a", 0)
 	verifyResponse(t, w, "1", 1)
-	w, _ = c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "0"})
+	w, _ = c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"})
 	checkWatchCount(t, c, "a", 0)
 	verifyResponse(t, w, "1", 1)
 
 	// Add another element and update the first, response should be different
 	c.UpdateResource("b", testResource("b"))
 	c.UpdateResource("a", testResource("aa"))
-	w, _ = c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
+	w, _ = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
 	verifyResponse(t, w, "3", 1)
-	w, _ = c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "0"})
+	w, _ = c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"})
 	verifyResponse(t, w, "3", 2)
 }
 
 func TestLinearDeletion(t *testing.T) {
 	c := NewLinearCache(testType, map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})
-	w, _ := c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
+	w, _ := c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "0"})
 	mustBlock(t, w)
 	checkWatchCount(t, c, "a", 1)
 	c.DeleteResource("a")
 	verifyResponse(t, w, "1", 0)
 	checkWatchCount(t, c, "a", 0)
-	w, _ = c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "0"})
+	w, _ = c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"})
 	verifyResponse(t, w, "1", 1)
 	checkWatchCount(t, c, "b", 0)
 	c.DeleteResource("b")
-	w, _ = c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "1"})
+	w, _ = c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"})
 	verifyResponse(t, w, "2", 0)
 	checkWatchCount(t, c, "b", 0)
 }
 
 func TestLinearWatchTwo(t *testing.T) {
 	c := NewLinearCache(testType, map[string]types.Resource{"a": testResource("a"), "b": testResource("b")})
-	w, _ := c.CreateWatch(Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: "0"})
+	w, _ := c.CreateWatch(&Request{ResourceNames: []string{"a", "b"}, TypeUrl: testType, VersionInfo: "0"})
 	mustBlock(t, w)
-	w1, _ := c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "0"})
+	w1, _ := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "0"})
 	mustBlock(t, w1)
 	c.UpdateResource("a", testResource("aa"))
 	// should only get the modified resource
@@ -162,24 +162,24 @@ func TestLinearCancel(t *testing.T) {
 	c.UpdateResource("a", testResource("a"))
 
 	// cancel watch-all
-	w, cancel := c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "1"})
+	w, cancel := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"})
 	mustBlock(t, w)
 	checkWatchCount(t, c, "a", 1)
 	cancel()
 	checkWatchCount(t, c, "a", 0)
 
 	// cancel watch for "a"
-	w, cancel = c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"})
+	w, cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"})
 	mustBlock(t, w)
 	checkWatchCount(t, c, "a", 1)
 	cancel()
 	checkWatchCount(t, c, "a", 0)
 
 	// open four watches for "a" and "b" and two for all, cancel one of each, make sure the second one is unaffected
-	w, cancel = c.CreateWatch(Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"})
-	w2, cancel2 := c.CreateWatch(Request{ResourceNames: []string{"b"}, TypeUrl: testType, VersionInfo: "1"})
-	w3, cancel3 := c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "1"})
-	w4, cancel4 := c.CreateWatch(Request{TypeUrl: testType, VersionInfo: "1"})
+	w, cancel = c.CreateWatch(&Request{ResourceNames: []string{"a"}, TypeUrl: testType, VersionInfo: "1"})
+	w2, cancel2 := c.CreateWatch(&Request{ResourceNames: []string{"b"}, TypeUrl: testType, VersionInfo: "1"})
+	w3, cancel3 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"})
+	w4, cancel4 := c.CreateWatch(&Request{TypeUrl: testType, VersionInfo: "1"})
 	mustBlock(t, w)
 	mustBlock(t, w2)
 	mustBlock(t, w3)
@@ -212,7 +212,7 @@ func TestLinearConcurrentSetWatch(t *testing.T) {
 				} else {
 					id2 := fmt.Sprintf("%d", i-1)
 					t.Logf("request resources %q and %q", id, id2)
-					value, _ := c.CreateWatch(Request{
+					value, _ := c.CreateWatch(&Request{
 						// Only expect one to become stale
 						ResourceNames: []string{id, id2},
 						VersionInfo:   "0",

pkg/cache/v2/mux.go

@@ -34,8 +34,8 @@ type MuxCache struct {
 
 var _ Cache = &MuxCache{}
 
-func (mux *MuxCache) CreateWatch(request Request) (chan Response, func()) {
-	key := mux.Classify(request)
+func (mux *MuxCache) CreateWatch(request *Request) (chan Response, func()) {
+	key := mux.Classify(*request)
 	cache, exists := mux.Caches[key]
 	if !exists {
 		value := make(chan Response, 0)
@@ -45,6 +45,6 @@ func (mux *MuxCache) CreateWatch(request Request) (chan Response, func()) {
 	return cache.CreateWatch(request)
 }
 
-func (mux *MuxCache) Fetch(ctx context.Context, request Request) (Response, error) {
+func (mux *MuxCache) Fetch(ctx context.Context, request *Request) (Response, error) {
 	return nil, errors.New("not implemented")
 }