Introduce Passthrough Response type

pkg/cache/v2/cache.go

@@ -20,6 +20,7 @@ import (
 
 	discovery "github.com/envoyproxy/go-control-plane/envoy/api/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
+	"github.com/golang/protobuf/ptypes/any"
 )
 
 // Request is an alias for the discovery request type.
@@ -51,21 +52,66 @@ type Cache interface {
 	Fetch(context.Context, Request) (*Response, error)
 }
 
+type ResponseType interface {
+	GetMarshalled() ([]*any.Any, error)
+}
+
 // Response is a pre-serialized xDS response.
 type Response struct {
+	ResponseType
 	// Request is the original request.
 	Request discovery.DiscoveryRequest
 
 	// Version of the resources as tracked by the cache for the given type.
 	// Proxy responds with this version as an acknowledgement.
 	Version string
 
-	// The value indicating whether the resource is marshaled, and only one of `Resources` and `MarshaledResources` is available.
-	ResourceMarshaled bool
-
 	// Resources to be included in the response.
 	Resources []types.Resource
 
+	// The value indicating whether the resource is marshaled, and only one of `Resources` and `MarshaledResources` is available.
+	isResourceMarshaled bool
+
 	// Marshaled Resources to be included in the response.
-	MarshaledResources []types.MarshaledResource
+	marshaledResources []*any.Any
+}
+
+type PassthroughResponse struct {
+	ResponseType
+	// Request is the original request.
+	Request discovery.DiscoveryRequest
+
+	// Version of the resources as tracked by the cache for the given type.
+	// Proxy responds with this version as an acknowledgement.
+	Version string
+
+	// Resources to be included in the response.
+	Resources []*any.Any
+}
+
+func (r Response) GetMarshalled() ([]*any.Any, error) {
+	if r.isResourceMarshaled {
+		return r.marshaledResources, nil
+	}
+
+	r.marshaledResources = make([]*any.Any, len(r.Resources))
+
+	for i, resource := range r.Resources {
+		marshaledResource, err := MarshalResource(resource)
+		if err != nil {
+			return nil, err
+		}
+		r.marshaledResources[i] = &any.Any{
+			TypeUrl: r.Request.TypeUrl,
+			Value:   marshaledResource,
+		}
+	}
+
+	r.isResourceMarshaled = true
+
+	return r.marshaledResources, nil
+}
+
+func (r PassthroughResponse) GetMarshalled() ([]*any.Any, error) {
+	return r.Resources, nil
 }

pkg/server/v2/gateway_test.go

@@ -22,9 +22,11 @@ import (
 	"testing"
 	"testing/iotest"
 
+	discovery "github.com/envoyproxy/go-control-plane/envoy/api/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/resource/v2"
+	rsrc "github.com/envoyproxy/go-control-plane/pkg/resource/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/server/v2"
 )
 
@@ -40,17 +42,20 @@ func (log logger) Errorf(format string, args ...interface{}) { log.t.Logf(format
 func TestGateway(t *testing.T) {
 	config := makeMockConfigWatcher()
 	config.responses = map[string][]cache.Response{
-		resource.ClusterType: []cache.Response{{
+		resource.ClusterType: {{
 			Version:   "2",
 			Resources: []types.Resource{cluster},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.ClusterType},
 		}},
-		resource.RouteType: []cache.Response{{
+		resource.RouteType: {{
 			Version:   "3",
 			Resources: []types.Resource{route},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.RouteType},
 		}},
-		resource.ListenerType: []cache.Response{{
+		resource.ListenerType: {{
 			Version:   "4",
 			Resources: []types.Resource{listener},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.ListenerType},
 		}},
 	}
 	gtw := server.HTTPGateway{Log: logger{t: t}, Server: server.NewServer(context.Background(), config, nil)}

pkg/server/v2/server.go

@@ -21,7 +21,6 @@ import (
 	"strconv"
 	"sync/atomic"
 
-	"github.com/golang/protobuf/ptypes/any"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -145,36 +144,13 @@ func createResponse(resp *cache.Response, typeURL string) (*discovery.DiscoveryR
 		return nil, errors.New("missing response")
 	}
 
-	var resources []*any.Any
-	if resp.ResourceMarshaled {
-		resources = make([]*any.Any, len(resp.MarshaledResources))
-	} else {
-		resources = make([]*any.Any, len(resp.Resources))
-	}
-
-	for i := 0; i < len(resources); i++ {
-		// Envoy relies on serialized protobuf bytes for detecting changes to the resources.
-		// This requires deterministic serialization.
-		if resp.ResourceMarshaled {
-			resources[i] = &any.Any{
-				TypeUrl: typeURL,
-				Value:   resp.MarshaledResources[i],
-			}
-		} else {
-			marshaledResource, err := cache.MarshalResource(resp.Resources[i])
-			if err != nil {
-				return nil, err
-			}
-
-			resources[i] = &any.Any{
-				TypeUrl: typeURL,
-				Value:   marshaledResource,
-			}
-		}
+	marshalledResources, err := resp.GetMarshalled()
+	if err != nil {
+		return nil, err
 	}
 	out := &discovery.DiscoveryResponse{
 		VersionInfo: resp.Version,
-		Resources:   resources,
+		Resources:   marshalledResources,
 		TypeUrl:     typeURL,
 	}
 	return out, nil

pkg/server/v2/server_test.go

@@ -179,21 +179,25 @@ var (
 
 func makeResponses() map[string][]cache.Response {
 	return map[string][]cache.Response{
-		rsrc.EndpointType: []cache.Response{{
+		rsrc.EndpointType: {{
 			Version:   "1",
 			Resources: []types.Resource{endpoint},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.EndpointType},
 		}},
-		rsrc.ClusterType: []cache.Response{{
+		rsrc.ClusterType: {{
 			Version:   "2",
 			Resources: []types.Resource{cluster},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.ClusterType},
 		}},
-		rsrc.RouteType: []cache.Response{{
+		rsrc.RouteType: {{
 			Version:   "3",
 			Resources: []types.Resource{route},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.RouteType},
 		}},
-		rsrc.ListenerType: []cache.Response{{
+		rsrc.ListenerType: {{
 			Version:   "4",
 			Resources: []types.Resource{listener},
+			Request:   discovery.DiscoveryRequest{TypeUrl: rsrc.ListenerType},
 		}},
 	}
 }
@@ -250,7 +254,7 @@ func TestResponseHandlers(t *testing.T) {
 
 			// make a request
 			resp := makeMockStream(t)
-			resp.recv <- &discovery.DiscoveryRequest{Node: node}
+			resp.recv <- &discovery.DiscoveryRequest{Node: node, TypeUrl: typ}
 			go func() {
 				var err error
 				switch typ {