Add RateLimitFilter support in k8s provider

Relates to envoyproxy#670

Signed-off-by: Arko Dasgupta <arko@tetrate.io>

api/v1alpha1/ratelimitfilter_types.go

@@ -9,6 +9,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	// KindRateLimitFilter is the name of the RateLimitFilter kind.
+	KindRateLimitFilter = "RateLimitFilter"
+)
+
 // +kubebuilder:object:root=true
 // +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
 
@@ -162,3 +167,7 @@ type RateLimitValue struct {
 //
 // +kubebuilder:validation:Enum=Second;Minute;Hour;Day
 type RateLimitUnit string
+
+func init() {
+	SchemeBuilder.Register(&RateLimitFilter{}, &RateLimitFilterList{})
+}

internal/gatewayapi/helpers.go

@@ -196,10 +196,12 @@ func ValidateHTTPRouteFilter(filter *v1beta1.HTTPRouteFilter) error {
 			return errors.New("extensionRef field must be specified for an extended filter")
 		case string(filter.ExtensionRef.Group) != egv1a1.GroupVersion.Group:
 			return fmt.Errorf("invalid group; must be %s", egv1a1.GroupVersion.Group)
-		case string(filter.ExtensionRef.Kind) != egv1a1.KindAuthenticationFilter:
-			return fmt.Errorf("invalid kind; must be %s", egv1a1.KindAuthenticationFilter)
-		default:
+		case string(filter.ExtensionRef.Kind) == egv1a1.KindAuthenticationFilter:
+			return nil
+		case string(filter.ExtensionRef.Kind) == egv1a1.KindRateLimitFilter:
 			return nil
+		default:
+			return fmt.Errorf("unknown %s kind", string(filter.ExtensionRef.Kind))
 		}
 	}
 

internal/gatewayapi/helpers_test.go

@@ -109,6 +109,18 @@ func TestValidateAuthenFilterRef(t *testing.T) {
 			},
 			expected: true,
 		},
+		{
+			name: "valid rateLimitfilter",
+			filter: &gwapiv1b1.HTTPRouteFilter{
+				Type: gwapiv1b1.HTTPRouteFilterExtensionRef,
+				ExtensionRef: &gwapiv1b1.LocalObjectReference{
+					Group: gwapiv1b1.Group(egv1a1.GroupVersion.Group),
+					Kind:  egv1a1.KindRateLimitFilter,
+					Name:  "test",
+				},
+			},
+			expected: true,
+		},
 	}
 
 	for _, tc := range testCases {

internal/gatewayapi/resource.go

@@ -20,30 +20,32 @@ type InfraIRMap map[string]*ir.Infra
 // resources that the translators needs as inputs.
 // +k8s:deepcopy-gen=true
 type Resources struct {
-	Gateways        []*v1beta1.Gateway
-	HTTPRoutes      []*v1beta1.HTTPRoute
-	GRPCRoutes      []*v1alpha2.GRPCRoute
-	TLSRoutes       []*v1alpha2.TLSRoute
-	TCPRoutes       []*v1alpha2.TCPRoute
-	UDPRoutes       []*v1alpha2.UDPRoute
-	ReferenceGrants []*v1alpha2.ReferenceGrant
-	Namespaces      []*v1.Namespace
-	Services        []*v1.Service
-	Secrets         []*v1.Secret
-	AuthenFilters   []*egv1a1.AuthenticationFilter
+	Gateways         []*v1beta1.Gateway
+	HTTPRoutes       []*v1beta1.HTTPRoute
+	GRPCRoutes       []*v1alpha2.GRPCRoute
+	TLSRoutes        []*v1alpha2.TLSRoute
+	TCPRoutes        []*v1alpha2.TCPRoute
+	UDPRoutes        []*v1alpha2.UDPRoute
+	ReferenceGrants  []*v1alpha2.ReferenceGrant
+	Namespaces       []*v1.Namespace
+	Services         []*v1.Service
+	Secrets          []*v1.Secret
+	AuthenFilters    []*egv1a1.AuthenticationFilter
+	RateLimitFilters []*egv1a1.RateLimitFilter
 }
 
 func NewResources() *Resources {
 	return &Resources{
-		Gateways:        []*v1beta1.Gateway{},
-		HTTPRoutes:      []*v1beta1.HTTPRoute{},
-		GRPCRoutes:      []*v1alpha2.GRPCRoute{},
-		TLSRoutes:       []*v1alpha2.TLSRoute{},
-		Services:        []*v1.Service{},
-		Secrets:         []*v1.Secret{},
-		ReferenceGrants: []*v1alpha2.ReferenceGrant{},
-		Namespaces:      []*v1.Namespace{},
-		AuthenFilters:   []*egv1a1.AuthenticationFilter{},
+		Gateways:         []*v1beta1.Gateway{},
+		HTTPRoutes:       []*v1beta1.HTTPRoute{},
+		GRPCRoutes:       []*v1alpha2.GRPCRoute{},
+		TLSRoutes:        []*v1alpha2.TLSRoute{},
+		Services:         []*v1.Service{},
+		Secrets:          []*v1.Secret{},
+		ReferenceGrants:  []*v1alpha2.ReferenceGrant{},
+		Namespaces:       []*v1.Namespace{},
+		AuthenFilters:    []*egv1a1.AuthenticationFilter{},
+		RateLimitFilters: []*egv1a1.RateLimitFilter{},
 	}
 }
 

internal/provider/kubernetes/config/rbac/role.yaml

@@ -27,6 +27,7 @@ rules:
   - gateway.envoyproxy.io
   resources:
   - authenticationfilters
+  - ratelimitfilters
   verbs:
   - get
   - list

internal/provider/kubernetes/controller.go

@@ -36,20 +36,21 @@ import (
 )
 
 const (
-	classGatewayIndex          = "classGatewayIndex"
-	gatewayTLSRouteIndex       = "gatewayTLSRouteIndex"
-	gatewayHTTPRouteIndex      = "gatewayHTTPRouteIndex"
-	gatewayGRPCRouteIndex      = "gatewayGRPCRouteIndex"
-	gatewayTCPRouteIndex       = "gatewayTCPRouteIndex"
-	gatewayUDPRouteIndex       = "gatewayUDPRouteIndex"
-	secretGatewayIndex         = "secretGatewayIndex"
-	targetRefGrantRouteIndex   = "targetRefGrantRouteIndex"
-	serviceHTTPRouteIndex      = "serviceHTTPRouteIndex"
-	serviceGRPCRouteIndex      = "serviceGRPCRouteIndex"
-	serviceTLSRouteIndex       = "serviceTLSRouteIndex"
-	serviceTCPRouteIndex       = "serviceTCPRouteIndex"
-	serviceUDPRouteIndex       = "serviceUDPRouteIndex"
-	authenFilterHTTPRouteIndex = "authenHTTPRouteIndex"
+	classGatewayIndex             = "classGatewayIndex"
+	gatewayTLSRouteIndex          = "gatewayTLSRouteIndex"
+	gatewayHTTPRouteIndex         = "gatewayHTTPRouteIndex"
+	gatewayGRPCRouteIndex         = "gatewayGRPCRouteIndex"
+	gatewayTCPRouteIndex          = "gatewayTCPRouteIndex"
+	gatewayUDPRouteIndex          = "gatewayUDPRouteIndex"
+	secretGatewayIndex            = "secretGatewayIndex"
+	targetRefGrantRouteIndex      = "targetRefGrantRouteIndex"
+	serviceHTTPRouteIndex         = "serviceHTTPRouteIndex"
+	serviceGRPCRouteIndex         = "serviceGRPCRouteIndex"
+	serviceTLSRouteIndex          = "serviceTLSRouteIndex"
+	serviceTCPRouteIndex          = "serviceTCPRouteIndex"
+	serviceUDPRouteIndex          = "serviceUDPRouteIndex"
+	authenFilterHTTPRouteIndex    = "authenHTTPRouteIndex"
+	rateLimitFilterHTTPRouteIndex = "rateLimitHTTPRouteIndex"
 )
 
 type gatewayAPIReconciler struct {
@@ -101,6 +102,9 @@ type resourceMappings struct {
 	// authenFilters is a map of AuthenticationFilters, where the key is the
 	// namespaced name of the AuthenticationFilter.
 	authenFilters map[types.NamespacedName]*egv1a1.AuthenticationFilter
+	// rateLimitFilters is a map of RateLimitFilters, where the key is the
+	// namespaced name of the RateLimitFilter.
+	rateLimitFilters map[types.NamespacedName]*egv1a1.RateLimitFilter
 }
 
 func newResourceMapping() *resourceMappings {
@@ -109,6 +113,7 @@ func newResourceMapping() *resourceMappings {
 		allAssociatedBackendRefs: map[types.NamespacedName]struct{}{},
 		allAssociatedRefGrants:   map[types.NamespacedName]*gwapiv1a2.ReferenceGrant{},
 		authenFilters:            map[types.NamespacedName]*egv1a1.AuthenticationFilter{},
+		rateLimitFilters:         map[types.NamespacedName]*egv1a1.RateLimitFilter{},
 	}
 }
 
@@ -343,6 +348,15 @@ func (r *gatewayAPIReconciler) getAuthenticationFilters(ctx context.Context) ([]
 	return authenList.Items, nil
 }
 
+func (r *gatewayAPIReconciler) getRateLimitFilters(ctx context.Context) ([]egv1a1.RateLimitFilter, error) {
+	rateLimitList := new(egv1a1.RateLimitFilterList)
+	if err := r.client.List(ctx, rateLimitList); err != nil {
+		return nil, fmt.Errorf("failed to list RateLimitFilters: %v", err)
+	}
+
+	return rateLimitList.Items, nil
+}
+
 func (r *gatewayAPIReconciler) processGateways(ctx context.Context, acceptedGC *gwapiv1b1.GatewayClass, resourceMap *resourceMappings, resourceTree *gatewayapi.Resources) error {
 	// Find gateways for the acceptedGC
 	// Find the Gateways that reference this Class.
@@ -460,7 +474,7 @@ func addReferenceGrantIndexers(ctx context.Context, mgr manager.Manager) error {
 // addHTTPRouteIndexers adds indexing on HTTPRoute.
 //   - For Service objects that are referenced in HTTPRoute objects via `.spec.rules.backendRefs`.
 //     This helps in querying for HTTPRoutes that are affected by a particular Service CRUD.
-//   - For AuthenticationFilter objects that are referenced in HTTPRoute objects via
+//   - For AuthenticationFilter and RateLimitFilter objects that are referenced in HTTPRoute objects via
 //     `.spec.rules[].filters`. This helps in querying for HTTPRoutes that are affected by a
 //     particular AuthenticationFilter CRUD.
 func addHTTPRouteIndexers(ctx context.Context, mgr manager.Manager) error {
@@ -472,30 +486,56 @@ func addHTTPRouteIndexers(ctx context.Context, mgr manager.Manager) error {
 		return err
 	}
 
-	if err := mgr.GetFieldIndexer().IndexField(ctx, &gwapiv1b1.HTTPRoute{}, authenFilterHTTPRouteIndex, func(obj client.Object) []string {
-		httproute := obj.(*gwapiv1b1.HTTPRoute)
-		var filters []string
-		for _, rule := range httproute.Spec.Rules {
-			for i := range rule.Filters {
-				filter := rule.Filters[i]
-				if filter.Type == gwapiv1b1.HTTPRouteFilterExtensionRef {
-					if err := gatewayapi.ValidateHTTPRouteFilter(&filter); err != nil {
-						filters = append(filters,
-							types.NamespacedName{
-								Namespace: httproute.Namespace,
-								Name:      string(filter.ExtensionRef.Name),
-							}.String(),
-						)
-					}
+	if err := mgr.GetFieldIndexer().IndexField(ctx, &gwapiv1b1.HTTPRoute{}, authenFilterHTTPRouteIndex, authenFilterHTTPRouteIndexFunc); err != nil {
+		return err
+	}
+
+	if err := mgr.GetFieldIndexer().IndexField(ctx, &gwapiv1b1.HTTPRoute{}, rateLimitFilterHTTPRouteIndex, rateLimitFilterHTTPRouteIndexFunc); err != nil {
+		return err
+	}
+	return nil
+}
+
+func authenFilterHTTPRouteIndexFunc(rawObj client.Object) []string {
+	httproute := rawObj.(*gwapiv1b1.HTTPRoute)
+	var filters []string
+	for _, rule := range httproute.Spec.Rules {
+		for i := range rule.Filters {
+			filter := rule.Filters[i]
+			if filter.Type == gwapiv1b1.HTTPRouteFilterExtensionRef && string(filter.ExtensionRef.Kind) == egv1a1.KindAuthenticationFilter {
+				if err := gatewayapi.ValidateHTTPRouteFilter(&filter); err != nil {
+					filters = append(filters,
+						types.NamespacedName{
+							Namespace: httproute.Namespace,
+							Name:      string(filter.ExtensionRef.Name),
+						}.String(),
+					)
 				}
 			}
 		}
-		return filters
-	}); err != nil {
-		return err
 	}
+	return filters
+}
 
-	return nil
+func rateLimitFilterHTTPRouteIndexFunc(rawObj client.Object) []string {
+	httproute := rawObj.(*gwapiv1b1.HTTPRoute)
+	var filters []string
+	for _, rule := range httproute.Spec.Rules {
+		for i := range rule.Filters {
+			filter := rule.Filters[i]
+			if filter.Type == gwapiv1b1.HTTPRouteFilterExtensionRef && string(filter.ExtensionRef.Kind) == egv1a1.KindRateLimitFilter {
+				if err := gatewayapi.ValidateHTTPRouteFilter(&filter); err != nil {
+					filters = append(filters,
+						types.NamespacedName{
+							Namespace: httproute.Namespace,
+							Name:      string(filter.ExtensionRef.Name),
+						}.String(),
+					)
+				}
+			}
+		}
+	}
+	return filters
 }
 
 func gatewayHTTPRouteIndexFunc(rawObj client.Object) []string {
@@ -1110,6 +1150,14 @@ func (r *gatewayAPIReconciler) watchResources(ctx context.Context, mgr manager.M
 		return err
 	}
 
+	// Watch RateLimitFilter CRUDs and enqueue associated HTTPRoute objects.
+	if err := c.Watch(
+		&source.Kind{Type: &egv1a1.RateLimitFilter{}},
+		&handler.EnqueueRequestForObject{},
+		predicate.NewPredicateFuncs(r.httpRoutesForRateLimitFilter)); err != nil {
+		return err
+	}
+
 	r.log.Info("watching gatewayAPI related objects")
 	return nil
 }

internal/provider/kubernetes/kubernetes_test.go

@@ -468,6 +468,14 @@ func testHTTPRoute(ctx context.Context, t *testing.T, provider *Provider, resour
 		require.NoError(t, cli.Delete(ctx, authenFilter))
 	}()
 
+	rateLimitFilter := test.GetRateLimitFilter("test-rateLimit", ns.Name)
+
+	require.NoError(t, cli.Create(ctx, rateLimitFilter))
+
+	defer func() {
+		require.NoError(t, cli.Delete(ctx, rateLimitFilter))
+	}()
+
 	redirectHostname := gwapiv1b1.PreciseHostname("redirect.hostname.local")
 	redirectPort := gwapiv1b1.PortNumber(8443)
 	redirectStatus := 301

internal/provider/kubernetes/predicates.go

@@ -213,6 +213,28 @@ func (r *gatewayAPIReconciler) httpRoutesForAuthenticationFilter(obj client.Obje
 	return len(httpRouteList.Items) != 0
 }
 
+// httpRoutesForRateLimitFilter tries finding HTTPRoute referents of the provided
+// RateLimitFilter and returns true if any exist.
+func (r *gatewayAPIReconciler) httpRoutesForRateLimitFilter(obj client.Object) bool {
+	ctx := context.Background()
+	filter, ok := obj.(*egv1a1.RateLimitFilter)
+	if !ok {
+		r.log.Info("unexpected object type, bypassing reconciliation", "object", obj)
+		return false
+	}
+
+	// Check if the RateLimitFilter belongs to a managed HTTPRoute.
+	httpRouteList := &gwapiv1b1.HTTPRouteList{}
+	if err := r.client.List(ctx, httpRouteList, &client.ListOptions{
+		FieldSelector: fields.OneTermEqualSelector(rateLimitFilterHTTPRouteIndex, utils.NamespacedName(filter).String()),
+	}); err != nil {
+		r.log.Error(err, "unable to find associated HTTPRoutes")
+		return false
+	}
+
+	return len(httpRouteList.Items) != 0
+}
+
 // envoyDeploymentForGateway returns the Envoy Deployment, returning nil if the Deployment doesn't exist.
 func (r *gatewayAPIReconciler) envoyDeploymentForGateway(ctx context.Context, gateway *gwapiv1b1.Gateway) (*appsv1.Deployment, error) {
 	key := types.NamespacedName{

internal/provider/kubernetes/rbac.go

@@ -7,7 +7,7 @@ package kubernetes
 
 // +kubebuilder:rbac:groups="gateway.networking.k8s.io",resources=gatewayclasses;gateways;httproutes;grpcroutes;tlsroutes;tcproutes;udproutes;referencepolicies;referencegrants,verbs=get;list;watch;update
 // +kubebuilder:rbac:groups="gateway.networking.k8s.io",resources=gatewayclasses/status;gateways/status;httproutes/status;grpcroutes/status;tlsroutes/status;tcproutes/status;udproutes/status,verbs=update
-// +kubebuilder:rbac:groups="gateway.envoyproxy.io",resources=authenticationfilters,verbs=get;list;watch;update
+// +kubebuilder:rbac:groups="gateway.envoyproxy.io",resources=authenticationfilters;ratelimitfilters,verbs=get;list;watch;update
 
 // RBAC for watched resources of Gateway API controllers.
 // +kubebuilder:rbac:groups="",resources=secrets;services;namespaces,verbs=get;list;watch