xds: add LEDS (Locality Endpoint Discovery Service) support

Wire up the LocalityEndpointDiscoveryService which allows discovering
LbEndpoint resources via delta xDS, as an alternative to inlining
endpoints in the EDS ClusterLoadAssignment response.

Signed-off-by: William Dauchy <william.dauchy@datadoghq.com>

Author: wdauchy

internal/example/server.go

@@ -56,6 +56,7 @@ func (s *Server) registerServer(grpcServer *grpc.Server) {
 	// register services
 	discoverygrpc.RegisterAggregatedDiscoveryServiceServer(grpcServer, s.xdsserver)
 	endpointservice.RegisterEndpointDiscoveryServiceServer(grpcServer, s.xdsserver)
+	endpointservice.RegisterLocalityEndpointDiscoveryServiceServer(grpcServer, s.xdsserver)
 	clusterservice.RegisterClusterDiscoveryServiceServer(grpcServer, s.xdsserver)
 	routeservice.RegisterRouteDiscoveryServiceServer(grpcServer, s.xdsserver)
 	listenerservice.RegisterListenerDiscoveryServiceServer(grpcServer, s.xdsserver)
@@ -99,6 +100,7 @@ func registerServer(grpcServer *grpc.Server, server server.Server) {
 	// register services
 	discoverygrpc.RegisterAggregatedDiscoveryServiceServer(grpcServer, server)
 	endpointservice.RegisterEndpointDiscoveryServiceServer(grpcServer, server)
+	endpointservice.RegisterLocalityEndpointDiscoveryServiceServer(grpcServer, server)
 	clusterservice.RegisterClusterDiscoveryServiceServer(grpcServer, server)
 	routeservice.RegisterRouteDiscoveryServiceServer(grpcServer, server)
 	listenerservice.RegisterListenerDiscoveryServiceServer(grpcServer, server)

pkg/cache/types/types.go

@@ -37,6 +37,7 @@ type ResponseType int
 const (
 	Cluster ResponseType = iota
 	Endpoint
+	LbEndpoint
 	Listener
 	Route
 	ScopedRoute

pkg/cache/v3/resource.go

@@ -50,6 +50,8 @@ func GetResponseType(typeURL resource.Type) types.ResponseType {
 		return types.ExtensionConfig
 	case resource.RateLimitConfigType:
 		return types.RateLimitConfig
+	case resource.LbEndpointType:
+		return types.LbEndpoint
 	}
 	return types.UnknownType
 }
@@ -77,6 +79,8 @@ func GetResponseTypeURL(responseType types.ResponseType) (string, error) {
 		return resource.ExtensionConfigType, nil
 	case types.RateLimitConfig:
 		return resource.RateLimitConfigType, nil
+	case types.LbEndpoint:
+		return resource.LbEndpointType, nil
 	default:
 		return "", fmt.Errorf("couldn't map response type %v to known resource type", responseType)
 	}

pkg/resource/v3/resource.go

@@ -29,6 +29,9 @@ const (
 	// Rate Limit service
 	RateLimitConfigType = APITypePrefix + "ratelimit.config.ratelimit.v3.RateLimitConfig"
 
+	// LEDS
+	LbEndpointType = APITypePrefix + "envoy.config.endpoint.v3.LbEndpoint"
+
 	// AnyType is used only by ADS
 	AnyType = ""
 )

pkg/server/v3/server.go

@@ -44,6 +44,7 @@ import (
 // Server is a collection of handlers for streaming discovery requests.
 type Server interface {
 	endpointservice.EndpointDiscoveryServiceServer
+	endpointservice.LocalityEndpointDiscoveryServiceServer
 	clusterservice.ClusterDiscoveryServiceServer
 	routeservice.RouteDiscoveryServiceServer
 	routeservice.ScopedRoutesDiscoveryServiceServer
@@ -350,3 +351,7 @@ func (s *server) DeltaExtensionConfigs(stream extensionconfigservice.ExtensionCo
 func (s *server) DeltaVirtualHosts(stream routeservice.VirtualHostDiscoveryService_DeltaVirtualHostsServer) error {
 	return s.DeltaStreamHandler(stream, resource.VirtualHostType)
 }
+
+func (s *server) DeltaLocalityEndpoints(stream endpointservice.LocalityEndpointDiscoveryService_DeltaLocalityEndpointsServer) error {
+	return s.DeltaStreamHandler(stream, resource.LbEndpointType)
+}

pkg/test/v3/register.go

@@ -46,4 +46,5 @@ func RegisterServer(grpcServer *grpc.Server, server server.Server) {
 	listenerservice.RegisterListenerDiscoveryServiceServer(grpcServer, server)
 	secretservice.RegisterSecretDiscoveryServiceServer(grpcServer, server)
 	runtimeservice.RegisterRuntimeDiscoveryServiceServer(grpcServer, server)
+	endpointservice.RegisterLocalityEndpointDiscoveryServiceServer(grpcServer, server)
 }