add SDS support (#56)

Makefile

@@ -89,6 +89,9 @@ integration.docker: docker
 	docker run -it -e "XDS=ads" test -debug
 	docker run -it -e "XDS=xds" test -debug
 	docker run -it -e "XDS=rest" test -debug
+	docker run -it -e "XDS=ads" test -debug -tls
+	docker run -it -e "XDS=xds" test -debug -tls
+	docker run -it -e "XDS=rest" test -debug -tls
 
 #-----------------
 #-- code generaion

pkg/cache/resource.go

@@ -18,6 +18,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 
 	v2 "github.com/envoyproxy/go-control-plane/envoy/api/v2"
+	"github.com/envoyproxy/go-control-plane/envoy/api/v2/auth"
 	hcm "github.com/envoyproxy/go-control-plane/envoy/config/filter/network/http_connection_manager/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/util"
 )
@@ -35,6 +36,7 @@ const (
 	ClusterType  = typePrefix + "Cluster"
 	RouteType    = typePrefix + "RouteConfiguration"
 	ListenerType = typePrefix + "Listener"
+	SecretType   = typePrefix + "auth.Secret"
 
 	// AnyType is used only by ADS
 	AnyType = ""
@@ -47,6 +49,7 @@ var (
 		ClusterType,
 		RouteType,
 		ListenerType,
+		SecretType,
 	}
 )
 
@@ -61,6 +64,8 @@ func GetResourceName(res Resource) string {
 		return v.GetName()
 	case *v2.Listener:
 		return v.GetName()
+	case *auth.Secret:
+		return v.GetName()
 	default:
 		return ""
 	}

pkg/cache/simple_test.go

@@ -56,6 +56,13 @@ var (
 		cache.RouteType:    []string{routeName},
 		cache.ListenerType: nil,
 	}
+
+	testTypes = []string{
+		cache.EndpointType,
+		cache.ClusterType,
+		cache.RouteType,
+		cache.ListenerType,
+	}
 )
 
 type logger struct {
@@ -81,7 +88,7 @@ func TestSnapshotCache(t *testing.T) {
 	case <-time.After(time.Second / 4):
 	}
 
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			value, _ := c.CreateWatch(v2.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]})
 			select {
@@ -105,7 +112,7 @@ func TestSnapshotCacheFetch(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			resp, err := c.Fetch(context.Background(), v2.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]})
 			if err != nil || resp == nil {
@@ -133,13 +140,13 @@ func TestSnapshotCacheFetch(t *testing.T) {
 func TestSnapshotCacheWatch(t *testing.T) {
 	c := cache.NewSnapshotCache(true, group{}, logger{t: t})
 	watches := make(map[string]chan cache.Response)
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		watches[typ], _ = c.CreateWatch(v2.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]})
 	}
 	if err := c.SetSnapshot(key, snapshot); err != nil {
 		t.Fatal(err)
 	}
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			select {
 			case out := <-watches[typ]:
@@ -156,10 +163,10 @@ func TestSnapshotCacheWatch(t *testing.T) {
 	}
 
 	// open new watches with the latest version
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		watches[typ], _ = c.CreateWatch(v2.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ], VersionInfo: version})
 	}
-	if count := c.GetStatusInfo(key).GetNumWatches(); count != len(cache.ResponseTypes) {
+	if count := c.GetStatusInfo(key).GetNumWatches(); count != len(testTypes) {
 		t.Errorf("watches should be created for the latest version: %d", count)
 	}
 
@@ -169,7 +176,7 @@ func TestSnapshotCacheWatch(t *testing.T) {
 	if err := c.SetSnapshot(key, snapshot2); err != nil {
 		t.Fatal(err)
 	}
-	if count := c.GetStatusInfo(key).GetNumWatches(); count != len(cache.ResponseTypes)-1 {
+	if count := c.GetStatusInfo(key).GetNumWatches(); count != len(testTypes)-1 {
 		t.Errorf("watches should be preserved for all but one: %d", count)
 	}
 
@@ -215,7 +222,7 @@ func TestConcurrentSetWatch(t *testing.T) {
 
 func TestSnapshotCacheWatchCancel(t *testing.T) {
 	c := cache.NewSnapshotCache(true, group{}, logger{t: t})
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		_, cancel := c.CreateWatch(v2.DiscoveryRequest{TypeUrl: typ, ResourceNames: names[typ]})
 		cancel()
 	}
@@ -224,7 +231,7 @@ func TestSnapshotCacheWatchCancel(t *testing.T) {
 		t.Error("got 0, want status info for the node")
 	}
 
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		if count := c.GetStatusInfo(key).GetNumWatches(); count > 0 {
 			t.Errorf("watches should be released for %s", typ)
 		}

pkg/cache/snapshot.go

@@ -60,6 +60,9 @@ type Snapshot struct {
 
 	// Listeners are items in the LDS response payload.
 	Listeners Resources
+
+	// Secrets are items in the SDS response payload.
+	Secrets Resources
 }
 
 // NewSnapshot creates a snapshot from response types and a version.
@@ -117,6 +120,8 @@ func (s *Snapshot) GetResources(typ string) map[string]Resource {
 		return s.Routes.Items
 	case ListenerType:
 		return s.Listeners.Items
+	case SecretType:
+		return s.Secrets.Items
 	}
 	return nil
 }
@@ -135,6 +140,8 @@ func (s *Snapshot) GetVersion(typ string) string {
 		return s.Routes.Version
 	case ListenerType:
 		return s.Listeners.Version
+	case SecretType:
+		return s.Secrets.Version
 	}
 	return ""
 }

pkg/server/gateway.go

@@ -50,6 +50,8 @@ func (h *HTTPGateway) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 		typeURL = cache.ListenerType
 	case "/v2/discovery:routes":
 		typeURL = cache.RouteType
+	case "/v2/discovery:secrets":
+		typeURL = cache.SecretType
 	default:
 		http.Error(resp, "no endpoint", http.StatusNotFound)
 		return

pkg/server/server.go

@@ -39,6 +39,7 @@ type Server interface {
 	v2.RouteDiscoveryServiceServer
 	v2.ListenerDiscoveryServiceServer
 	discovery.AggregatedDiscoveryServiceServer
+	discovery.SecretDiscoveryServiceServer
 
 	// Fetch is the universal fetch method.
 	Fetch(context.Context, *v2.DiscoveryRequest) (*v2.DiscoveryResponse, error)
@@ -89,16 +90,19 @@ type watches struct {
 	clusters  chan cache.Response
 	routes    chan cache.Response
 	listeners chan cache.Response
+	secrets   chan cache.Response
 
 	endpointCancel func()
 	clusterCancel  func()
 	routeCancel    func()
 	listenerCancel func()
+	secretCancel   func()
 
 	endpointNonce string
 	clusterNonce  string
 	routeNonce    string
 	listenerNonce string
+	secretNonce   string
 }
 
 // Cancel all watches
@@ -115,6 +119,9 @@ func (values watches) Cancel() {
 	if values.listenerCancel != nil {
 		values.listenerCancel()
 	}
+	if values.secretCancel != nil {
+		values.secretCancel()
+	}
 }
 
 func createResponse(resp *cache.Response, typeURL string) (*v2.DiscoveryResponse, error) {
@@ -223,6 +230,16 @@ func (s *server) process(stream stream, reqCh <-chan *v2.DiscoveryRequest, defau
 			}
 			values.listenerNonce = nonce
 
+		case resp, more := <-values.secrets:
+			if !more {
+				return status.Errorf(codes.Unavailable, "secrets watch failed")
+			}
+			nonce, err := send(resp, cache.SecretType)
+			if err != nil {
+				return err
+			}
+			values.secretNonce = nonce
+
 		case req, more := <-reqCh:
 			// input stream ended or errored out
 			if !more {
@@ -270,6 +287,11 @@ func (s *server) process(stream stream, reqCh <-chan *v2.DiscoveryRequest, defau
 					values.listenerCancel()
 				}
 				values.listeners, values.listenerCancel = s.cache.CreateWatch(*req)
+			case req.TypeUrl == cache.SecretType && (values.secretNonce == "" || values.secretNonce == nonce):
+				if values.secretCancel != nil {
+					values.secretCancel()
+				}
+				values.secrets, values.secretCancel = s.cache.CreateWatch(*req)
 			}
 		}
 	}
@@ -323,6 +345,10 @@ func (s *server) StreamListeners(stream v2.ListenerDiscoveryService_StreamListen
 	return s.handler(stream, cache.ListenerType)
 }
 
+func (s *server) StreamSecrets(stream discovery.SecretDiscoveryService_StreamSecretsServer) error {
+	return s.handler(stream, cache.SecretType)
+}
+
 // Fetch is the universal fetch method.
 func (s *server) Fetch(ctx context.Context, req *v2.DiscoveryRequest) (*v2.DiscoveryResponse, error) {
 	if s.callbacks != nil {
@@ -373,6 +399,14 @@ func (s *server) FetchListeners(ctx context.Context, req *v2.DiscoveryRequest) (
 	return s.Fetch(ctx, req)
 }
 
+func (s *server) FetchSecrets(ctx context.Context, req *v2.DiscoveryRequest) (*v2.DiscoveryResponse, error) {
+	if req == nil {
+		return nil, status.Errorf(codes.Unavailable, "empty request")
+	}
+	req.TypeUrl = cache.SecretType
+	return s.Fetch(ctx, req)
+}
+
 func (s *server) IncrementalAggregatedResources(_ discovery.AggregatedDiscoveryService_IncrementalAggregatedResourcesServer) error {
 	return errors.New("not implemented")
 }

pkg/server/server_test.go

@@ -165,10 +165,16 @@ var (
 		Id:      "test-id",
 		Cluster: "test-cluster",
 	}
-	endpoint = resource.MakeEndpoint(clusterName, 8080)
-	cluster  = resource.MakeCluster(resource.Ads, clusterName)
-	route    = resource.MakeRoute(routeName, clusterName)
-	listener = resource.MakeHTTPListener(resource.Ads, listenerName, 80, routeName)
+	endpoint  = resource.MakeEndpoint(clusterName, 8080)
+	cluster   = resource.MakeCluster(resource.Ads, clusterName)
+	route     = resource.MakeRoute(routeName, clusterName)
+	listener  = resource.MakeHTTPListener(resource.Ads, listenerName, 80, routeName)
+	testTypes = []string{
+		cache.EndpointType,
+		cache.ClusterType,
+		cache.RouteType,
+		cache.ListenerType,
+	}
 )
 
 func makeResponses() map[string][]cache.Response {
@@ -193,7 +199,7 @@ func makeResponses() map[string][]cache.Response {
 }
 
 func TestResponseHandlers(t *testing.T) {
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			config := makeMockConfigWatcher()
 			config.responses = makeResponses()
@@ -304,7 +310,7 @@ func TestFetch(t *testing.T) {
 }
 
 func TestWatchClosed(t *testing.T) {
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			config := makeMockConfigWatcher()
 			config.closeWatch = true
@@ -328,7 +334,7 @@ func TestWatchClosed(t *testing.T) {
 }
 
 func TestSendError(t *testing.T) {
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			config := makeMockConfigWatcher()
 			config.responses = makeResponses()
@@ -353,7 +359,7 @@ func TestSendError(t *testing.T) {
 }
 
 func TestStaleNonce(t *testing.T) {
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			config := makeMockConfigWatcher()
 			config.responses = makeResponses()
@@ -466,7 +472,7 @@ func TestAggregateRequestType(t *testing.T) {
 }
 
 func TestCallbackError(t *testing.T) {
-	for _, typ := range cache.ResponseTypes {
+	for _, typ := range testTypes {
 		t.Run(typ, func(t *testing.T) {
 			config := makeMockConfigWatcher()
 			config.responses = makeResponses()

pkg/test/main/main.go

@@ -17,6 +17,7 @@ package main
 
 import (
 	"context"
+	cryptotls "crypto/tls"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -51,6 +52,7 @@ var (
 	clusters      int
 	httpListeners int
 	tcpListeners  int
+	tls           bool
 
 	nodeID string
 )
@@ -70,6 +72,7 @@ func init() {
 	flag.IntVar(&httpListeners, "http", 2, "Number of HTTP listeners (and RDS configs)")
 	flag.IntVar(&tcpListeners, "tcp", 2, "Number of TCP pass-through listeners")
 	flag.StringVar(&nodeID, "nodeID", "test-id", "Node ID")
+	flag.BoolVar(&tls, "tls", false, "Enable TLS on all listeners and use SDS for secret delivery")
 }
 
 // main returns code 1 if any of the batches failed to pass all requests
@@ -98,6 +101,7 @@ func main() {
 		NumClusters:      clusters,
 		NumHTTPListeners: httpListeners,
 		NumTCPListeners:  tcpListeners,
+		TLS:              tls,
 	}
 
 	// start the xDS server
@@ -164,8 +168,15 @@ func callEcho() (int, int) {
 		go func(i int) {
 			client := http.Client{
 				Timeout: 100 * time.Millisecond,
+				Transport: &http.Transport{
+					TLSClientConfig: &cryptotls.Config{InsecureSkipVerify: true},
+				},
 			}
-			req, err := client.Get(fmt.Sprintf("http://localhost:%d", basePort+uint(i)))
+			proto := "http"
+			if tls {
+				proto = "https"
+			}
+			req, err := client.Get(fmt.Sprintf("%s://127.0.0.1:%d", proto, basePort+uint(i)))
 			if err != nil {
 				ch <- err
 				return