internal/grpc: adds SDS and registers xds for sds

Addresses the first high-level design bullet point, creatign a gRPC server for SDS by adding an SDS cache interface, implementing FetchSecrets and StreamSecrets, and registering the server as an SDS handler. Updates projectcontour#898 Signed-off-by: Matt Alberts <[email protected]>
mattalberts · Apr 30, 2019 · 44dfa61 · 44dfa61
1 parent d678cd9
commit 44dfa61
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 2 deletions.
diff --git a/internal/grpc/resources.go b/internal/grpc/resources.go
@@ -17,6 +17,7 @@ import (
 	"sort"
 
 	"github.com/envoyproxy/go-control-plane/envoy/api/v2"
+	"github.com/envoyproxy/go-control-plane/envoy/api/v2/auth"
 
 	"github.com/gogo/protobuf/proto"
 )
@@ -29,6 +30,7 @@ const (
 	clusterType  = typePrefix + "Cluster"
 	routeType    = typePrefix + "RouteConfiguration"
 	listenerType = typePrefix + "Listener"
+	secretType   = typePrefix + "auth.Secret"
 )
 
 // cache represents a source of proto.Message valus that can be registered
@@ -127,3 +129,25 @@ func (r routeConfigurationsByName) Swap(i, j int) { r[i], r[j] = r[j], r[i] }
 func (r routeConfigurationsByName) Less(i, j int) bool {
 	return r[i].(*v2.RouteConfiguration).Name < r[j].(*v2.RouteConfiguration).Name
 }
+
+// SDS implements the RDS v2 gRPC API.
+type SDS struct {
+	Cache
+}
+
+// Values returns a sorted list of RouteConfigurations.
+func (s *SDS) Values(filter func(string) bool) []proto.Message {
+	v := s.Cache.Values(filter)
+	sort.Stable(secretsByName(v))
+	return v
+}
+
+func (s *SDS) TypeURL() string { return secretType }
+
+type secretsByName []proto.Message
+
+func (s secretsByName) Len() int      { return len(s) }
+func (s secretsByName) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
+func (s secretsByName) Less(i, j int) bool {
+	return s[i].(*auth.Secret).Name < s[j].(*auth.Secret).Name
+}
diff --git a/internal/grpc/server.go b/internal/grpc/server.go
@@ -20,7 +20,8 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/envoyproxy/go-control-plane/envoy/api/v2"
-	envoy_service_v2 "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v2"
+	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
+	loadstats "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v2"
 	"github.com/sirupsen/logrus"
 )
 
@@ -65,6 +66,7 @@ func NewAPI(log logrus.FieldLogger, cacheMap map[string]Cache) *grpc.Server {
 	v2.RegisterEndpointDiscoveryServiceServer(g, s)
 	v2.RegisterListenerDiscoveryServiceServer(g, s)
 	v2.RegisterRouteDiscoveryServiceServer(g, s)
+	discovery.RegisterSecretDiscoveryServiceServer(g, s)
 	return g
 }
 
@@ -97,6 +99,10 @@ func (s *grpcServer) FetchRoutes(_ context.Context, req *v2.DiscoveryRequest) (*
 	return s.fetch(req)
 }
 
+func (s *grpcServer) FetchSecrets(_ context.Context, req *v2.DiscoveryRequest) (*v2.DiscoveryResponse, error) {
+	return s.fetch(req)
+}
+
 func (s *grpcServer) StreamClusters(srv v2.ClusterDiscoveryService_StreamClustersServer) error {
 	return s.stream(srv)
 }
@@ -105,7 +111,7 @@ func (s *grpcServer) StreamEndpoints(srv v2.EndpointDiscoveryService_StreamEndpo
 	return s.stream(srv)
 }
 
-func (s *grpcServer) StreamLoadStats(srv envoy_service_v2.LoadReportingService_StreamLoadStatsServer) error {
+func (s *grpcServer) StreamLoadStats(srv loadstats.LoadReportingService_StreamLoadStatsServer) error {
 	return status.Errorf(codes.Unimplemented, "StreamLoadStats unimplemented")
 }
 
@@ -124,3 +130,7 @@ func (s *grpcServer) StreamListeners(srv v2.ListenerDiscoveryService_StreamListe
 func (s *grpcServer) StreamRoutes(srv v2.RouteDiscoveryService_StreamRoutesServer) error {
 	return s.stream(srv)
 }
+
+func (s *grpcServer) StreamSecrets(srv discovery.SecretDiscoveryService_StreamSecretsServer) error {
+	return s.stream(srv)
+}