Support EndpointSlice in Kubernetes Provider

* Reconcile relavant EndpointSlices associated with a Service that is referenced in a xRoute * Add the EndpointSlices to the provider message Relates to envoyproxy#1256 Signed-off-by: Arko Dasgupta <[email protected]>
arkodg · Jun 1, 2023 · c5b900b · c5b900b
1 parent 22bd06b
commit c5b900b
Show file tree

Hide file tree

Showing 8 changed files with 175 additions and 7 deletions.
diff --git a/charts/gateway-helm/templates/generated/rbac/roles.yaml b/charts/gateway-helm/templates/generated/rbac/roles.yaml
@@ -33,6 +33,14 @@ rules:
   - list
   - update
   - watch
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - gateway.envoyproxy.io
   resources:

diff --git a/internal/gatewayapi/resource.go b/internal/gatewayapi/resource.go
@@ -7,6 +7,7 @@ package gatewayapi
 
 import (
 	v1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"sigs.k8s.io/gateway-api/apis/v1alpha2"
 	"sigs.k8s.io/gateway-api/apis/v1beta1"
@@ -35,6 +36,7 @@ type Resources struct {
 	ReferenceGrants       []*v1alpha2.ReferenceGrant     `json:"referenceGrants,omitempty"`
 	Namespaces            []*v1.Namespace                `json:"namespaces,omitempty"`
 	Services              []*v1.Service                  `json:"services,omitempty"`
+	EndpointSlices        []*discoveryv1.EndpointSlice   `json:"endpointSlices,omitempty"`
 	Secrets               []*v1.Secret                   `json:"secrets,omitempty"`
 	AuthenticationFilters []*egv1a1.AuthenticationFilter `json:"authenticationFilters,omitempty"`
 	RateLimitFilters      []*egv1a1.RateLimitFilter      `json:"rateLimitFilters,omitempty"`
@@ -49,6 +51,7 @@ func NewResources() *Resources {
 		GRPCRoutes:            []*v1alpha2.GRPCRoute{},
 		TLSRoutes:             []*v1alpha2.TLSRoute{},
 		Services:              []*v1.Service{},
+		EndpointSlices:        []*discoveryv1.EndpointSlice{},
 		Secrets:               []*v1.Secret{},
 		ReferenceGrants:       []*v1alpha2.ReferenceGrant{},
 		Namespaces:            []*v1.Namespace{},

diff --git a/internal/gatewayapi/zz_generated.deepcopy.go b/internal/gatewayapi/zz_generated.deepcopy.go
diff --git a/internal/provider/kubernetes/controller.go b/internal/provider/kubernetes/controller.go
@@ -12,6 +12,7 @@ import (
 	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/fields"
@@ -204,6 +205,26 @@ func (r *gatewayAPIReconciler) Reconcile(ctx context.Context, request reconcile.
 			resourceTree.Services = append(resourceTree.Services, service)
 			r.log.Info("added Service to resource tree", "namespace", serviceNamespaceName.Namespace,
 				"name", serviceNamespaceName.Name)
+
+			// Retrieve the EndpointSlices associated with the service
+			endpointSliceList := new(discoveryv1.EndpointSliceList)
+			opts := []client.ListOption{
+				client.MatchingLabels(map[string]string{
+					discoveryv1.LabelServiceName: serviceNamespaceName.Name,
+				}),
+				client.InNamespace(serviceNamespaceName.Namespace),
+			}
+			if err := r.client.List(ctx, endpointSliceList, opts...); err != nil {
+				r.log.Error(err, "failed to get EndpointSlices", "namespace", serviceNamespaceName.Namespace,
+					"service", serviceNamespaceName.Name)
+			} else {
+				for i := range endpointSliceList.Items {
+					endpointSlice := endpointSliceList.Items[i]
+					r.log.Info("added EndpointSlice to resource tree", "namespace", endpointSlice.Namespace,
+						"name", endpointSlice.Name)
+					resourceTree.EndpointSlices = append(resourceTree.EndpointSlices, &endpointSlice)
+				}
+			}
 		}
 	}
 
@@ -1130,6 +1151,14 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M
 		return err
 	}
 
+	// Watch EndpointSlice CRUDs and process affected *Route objects.
+	if err := c.Watch(
+		source.Kind(mgr.GetCache(), &discoveryv1.EndpointSlice{}),
+		&handler.EnqueueRequestForObject{},
+		predicate.NewPredicateFuncs(r.validateEndpointSliceForReconcile)); err != nil {
+		return err
+	}
+
 	// Watch Node CRUDs to update Gateway Address exposed by Service of type NodePort.
 	// Node creation/deletion and ExternalIP updates would require update in the Gateway
 	// resource address.

diff --git a/internal/provider/kubernetes/predicates.go b/internal/provider/kubernetes/predicates.go
@@ -10,6 +10,7 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/types"
@@ -22,8 +23,6 @@ import (
 	"github.com/envoyproxy/gateway/internal/provider/utils"
 )
 
-// TODO: all predicate functions are unti test candidates.
-
 // hasMatchingController returns true if the provided object is a GatewayClass
 // with a Spec.Controller string matching this Envoy Gateway's controller string,
 // or false otherwise.
@@ -116,41 +115,49 @@ func (r *gatewayAPIReconciler) validateServiceForReconcile(obj client.Object) bo
 		return false
 	}
 
+	nsName := utils.NamespacedName(svc)
+	return r.isRouteReferencingService(&nsName)
+}
+
+// isRouteReferencingService returns true if the service is referenced by any of the xRoutes
+// in the system, else returns false.
+func (r *gatewayAPIReconciler) isRouteReferencingService(nsName *types.NamespacedName) bool {
+	ctx := context.Background()
 	httpRouteList := &gwapiv1b1.HTTPRouteList{}
 	if err := r.client.List(ctx, httpRouteList, &client.ListOptions{
-		FieldSelector: fields.OneTermEqualSelector(serviceHTTPRouteIndex, utils.NamespacedName(svc).String()),
+		FieldSelector: fields.OneTermEqualSelector(serviceHTTPRouteIndex, nsName.String()),
 	}); err != nil {
 		r.log.Error(err, "unable to find associated HTTPRoutes")
 		return false
 	}
 
 	grpcRouteList := &gwapiv1a2.GRPCRouteList{}
 	if err := r.client.List(ctx, grpcRouteList, &client.ListOptions{
-		FieldSelector: fields.OneTermEqualSelector(serviceGRPCRouteIndex, utils.NamespacedName(svc).String()),
+		FieldSelector: fields.OneTermEqualSelector(serviceGRPCRouteIndex, nsName.String()),
 	}); err != nil {
 		r.log.Error(err, "unable to find associated GRPCRoutes")
 		return false
 	}
 
 	tlsRouteList := &gwapiv1a2.TLSRouteList{}
 	if err := r.client.List(ctx, tlsRouteList, &client.ListOptions{
-		FieldSelector: fields.OneTermEqualSelector(serviceTLSRouteIndex, utils.NamespacedName(svc).String()),
+		FieldSelector: fields.OneTermEqualSelector(serviceTLSRouteIndex, nsName.String()),
 	}); err != nil {
 		r.log.Error(err, "unable to find associated TLSRoutes")
 		return false
 	}
 
 	tcpRouteList := &gwapiv1a2.TCPRouteList{}
 	if err := r.client.List(ctx, tcpRouteList, &client.ListOptions{
-		FieldSelector: fields.OneTermEqualSelector(serviceTCPRouteIndex, utils.NamespacedName(svc).String()),
+		FieldSelector: fields.OneTermEqualSelector(serviceTCPRouteIndex, nsName.String()),
 	}); err != nil {
 		r.log.Error(err, "unable to find associated TCPRoutes")
 		return false
 	}
 
 	udpRouteList := &gwapiv1a2.UDPRouteList{}
 	if err := r.client.List(ctx, udpRouteList, &client.ListOptions{
-		FieldSelector: fields.OneTermEqualSelector(serviceUDPRouteIndex, utils.NamespacedName(svc).String()),
+		FieldSelector: fields.OneTermEqualSelector(serviceUDPRouteIndex, nsName.String()),
 	}); err != nil {
 		r.log.Error(err, "unable to find associated UDPRoutes")
 		return false
@@ -166,6 +173,29 @@ func (r *gatewayAPIReconciler) validateServiceForReconcile(obj client.Object) bo
 	return allAssociatedRoutes != 0
 }
 
+// validateEndpointSliceForReconcile returns true if the the endpointSlice references
+// a service that is referenced by a xRoute
+func (r *gatewayAPIReconciler) validateEndpointSliceForReconcile(obj client.Object) bool {
+	ep, ok := obj.(*discoveryv1.EndpointSlice)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", obj)
+		return false
+	}
+
+	svcName, ok := ep.GetLabels()[discoveryv1.LabelServiceName]
+	if !ok {
+		r.log.Info("endpointslice is missing kubernetes.io/service-name label", "object", obj)
+		return false
+	}
+
+	nsName := types.NamespacedName{
+		Namespace: obj.GetNamespace(),
+		Name:      svcName,
+	}
+
+	return r.isRouteReferencingService(&nsName)
+}
+
 // validateDeploymentForReconcile tries finding the owning Gateway of the Deployment
 // if it exists, finds the Gateway's Service, and further updates the Gateway
 // status Ready condition. No Deployments are pushed for reconciliation.

diff --git a/internal/provider/kubernetes/predicates_test.go b/internal/provider/kubernetes/predicates_test.go
@@ -165,6 +165,64 @@ func TestValidateSecretForReconcile(t *testing.T) {
 	}
 }
 
+// TestValidateEndpointSliceForReconcile tests the validateEndpointSliceForReconcile
+// predicate function.
+func TestValidateEndpointSliceForReconcile(t *testing.T) {
+	sampleGateway := test.GetGateway(types.NamespacedName{Namespace: "default", Name: "scheduled-status-test"}, "test-gc")
+
+	testCases := []struct {
+		name          string
+		configs       []client.Object
+		endpointSlice client.Object
+		expect        bool
+	}{
+		{
+			name: "route service but no routes exist",
+			configs: []client.Object{
+				test.GetGatewayClass("test-gc", v1alpha1.GatewayControllerName),
+				sampleGateway,
+			},
+			endpointSlice: test.GetEndpointSlice(types.NamespacedName{Name: "endpointslice"}, "service"),
+			expect:        false,
+		},
+		{
+			name: "http route service routes exist",
+			configs: []client.Object{
+				test.GetGatewayClass("test-gc", v1alpha1.GatewayControllerName),
+				sampleGateway,
+				test.GetHTTPRoute(types.NamespacedName{Name: "httproute-test"}, "scheduled-status-test", types.NamespacedName{Name: "service"}),
+			},
+			endpointSlice: test.GetEndpointSlice(types.NamespacedName{Name: "endpointslice"}, "service"),
+			expect:        true,
+		},
+	}
+
+	// Create the reconciler.
+	logger, err := log.NewLogger()
+	require.NoError(t, err)
+	r := gatewayAPIReconciler{
+		classController: v1alpha1.GatewayControllerName,
+		log:             logger,
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+		r.client = fakeclient.NewClientBuilder().
+			WithScheme(envoygateway.GetScheme()).
+			WithObjects(tc.configs...).
+			WithIndex(&gwapiv1b1.HTTPRoute{}, serviceHTTPRouteIndex, serviceHTTPRouteIndexFunc).
+			WithIndex(&gwapiv1a2.GRPCRoute{}, serviceGRPCRouteIndex, serviceGRPCRouteIndexFunc).
+			WithIndex(&gwapiv1a2.TLSRoute{}, serviceTLSRouteIndex, serviceTLSRouteIndexFunc).
+			WithIndex(&gwapiv1a2.TCPRoute{}, serviceTCPRouteIndex, serviceTCPRouteIndexFunc).
+			WithIndex(&gwapiv1a2.UDPRoute{}, serviceUDPRouteIndex, serviceUDPRouteIndexFunc).
+			Build()
+		t.Run(tc.name, func(t *testing.T) {
+			res := r.validateEndpointSliceForReconcile(tc.endpointSlice)
+			require.Equal(t, tc.expect, res)
+		})
+	}
+}
+
 // TestValidateServiceForReconcile tests the validateServiceForReconcile
 // predicate function.
 func TestValidateServiceForReconcile(t *testing.T) {

diff --git a/internal/provider/kubernetes/rbac.go b/internal/provider/kubernetes/rbac.go
@@ -16,3 +16,4 @@ package kubernetes
 // RBAC for watched resources of Gateway API controllers.
 // +kubebuilder:rbac:groups="",resources=secrets;services;namespaces;nodes,verbs=get;list;watch
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch
+// +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch
diff --git a/internal/provider/kubernetes/test/utils.go b/internal/provider/kubernetes/test/utils.go
@@ -8,6 +8,7 @@ package test
 import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	gwapiv1a2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
@@ -284,6 +285,32 @@ func GetService(nsname types.NamespacedName, labels map[string]string, ports map
 	return service
 }
 
+// GetEndpointSlice returns a sample EndpointSlice.
+func GetEndpointSlice(nsName types.NamespacedName, svcName string) *discoveryv1.EndpointSlice {
+	return &discoveryv1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      nsName.Name,
+			Namespace: nsName.Namespace,
+			Labels:    map[string]string{discoveryv1.LabelServiceName: svcName},
+		},
+		Endpoints: []discoveryv1.Endpoint{
+			{
+				Addresses: []string{"10.0.0.1"},
+				Conditions: discoveryv1.EndpointConditions{
+					Ready: &[]bool{true}[0],
+				},
+			},
+		},
+		Ports: []discoveryv1.EndpointPort{
+			{
+				Name:     &[]string{"dummy"}[0],
+				Port:     &[]int32{8080}[0],
+				Protocol: &[]corev1.Protocol{corev1.ProtocolTCP}[0],
+			},
+		},
+	}
+}
+
 // GetAuthenticationFilter returns a pointer to an AuthenticationFilter with the
 // provided ns/name. The AuthenticationFilter uses a JWT provider with dummy issuer,
 // audiences, and remoteJWKS settings.