feat: support translate BackendTrafficPolicy CRD (#92)

Author: AlinsRan

api/v1alpha1/backendtrafficpolicy_types.go

@@ -2,8 +2,6 @@ package v1alpha1
 
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
 )
 
 // +kubebuilder:object:root=true
@@ -12,22 +10,18 @@ type BackendTrafficPolicy struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 
-	Spec   BackendTrafficPolicySpec     `json:"spec,omitempty"`
-	Status gatewayv1alpha2.PolicyStatus `json:"status,omitempty"`
+	Spec   BackendTrafficPolicySpec `json:"spec,omitempty"`
+	Status PolicyStatus             `json:"status,omitempty"`
 }
 
 type BackendTrafficPolicySpec struct {
 	// TargetRef identifies an API object to apply policy to.
 	// Currently, Backends (i.e. Service, ServiceImport, or any
 	// implementation-specific backendRef) are the only valid API
 	// target references.
-	// +listType=map
-	// +listMapKey=group
-	// +listMapKey=kind
-	// +listMapKey=name
 	// +kubebuilder:validation:MinItems=1
 	// +kubebuilder:validation:MaxItems=16
-	TargetRefs []gatewayv1alpha2.LocalPolicyTargetReferenceWithSectionName `json:"targetRefs"`
+	TargetRefs []BackendPolicyTargetReferenceWithSectionName `json:"targetRefs"`
 	// LoadBalancer represents the load balancer configuration for Kubernetes Service.
 	// The default strategy is round robin.
 	LoadBalancer *LoadBalancer `json:"loadbalancer,omitempty" yaml:"loadbalancer,omitempty"`
@@ -74,9 +68,12 @@ type LoadBalancer struct {
 }
 
 type Timeout struct {
+	// +kubebuilder:default="60s"
 	Connect metav1.Duration `json:"connect,omitempty" yaml:"connect,omitempty"`
-	Send    metav1.Duration `json:"send,omitempty" yaml:"send,omitempty"`
-	Read    metav1.Duration `json:"read,omitempty" yaml:"read,omitempty"`
+	// +kubebuilder:default="60s"
+	Send metav1.Duration `json:"send,omitempty" yaml:"send,omitempty"`
+	// +kubebuilder:default="60s"
+	Read metav1.Duration `json:"read,omitempty" yaml:"read,omitempty"`
 }
 
 // +kubebuilder:object:root=true

api/v1alpha1/policies_type.go

@@ -0,0 +1,8 @@
+package v1alpha1
+
+import gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2"
+
+type PolicyStatus gatewayv1alpha2.PolicyStatus
+
+// +kubebuilder:validation:XValidation:rule="self.kind == 'Service' && self.group == \"\""
+type BackendPolicyTargetReferenceWithSectionName gatewayv1alpha2.LocalPolicyTargetReferenceWithSectionName

api/v1alpha1/zz_generated.deepcopy.go

@@ -153,23 +153,23 @@ spec:
                   - kind
                   - name
                   type: object
+                  x-kubernetes-validations:
+                  - rule: self.kind == 'Service' && self.group == ""
                 maxItems: 16
                 minItems: 1
                 type: array
-                x-kubernetes-list-map-keys:
-                - group
-                - kind
-                - name
-                x-kubernetes-list-type: map
               timeout:
                 description: Timeout settings for the read, send and connect to the
                   upstream.
                 properties:
                   connect:
+                    default: 60s
                     type: string
                   read:
+                    default: 60s
                     type: string
                   send:
+                    default: 60s
                     type: string
                 type: object
               upstream_host:

config/crd/bases/gateway.apisix.io_backendtrafficpolicies.yaml

@@ -136,7 +136,7 @@ func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	}
 
 	var statusErr error
-	tctx := provider.NewDefaultTranslateContext()
+	tctx := provider.NewDefaultTranslateContext(ctx)
 
 	if err := r.processSpec(ctx, tctx, consumer); err != nil {
 		r.Log.Error(err, "failed to process consumer spec", "consumer", consumer)

internal/controller/consumer_controller.go

@@ -112,7 +112,7 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		return ctrl.Result{}, nil
 	}
 
-	tctx := provider.NewDefaultTranslateContext()
+	tctx := provider.NewDefaultTranslateContext(ctx)
 
 	if err := r.processHTTPRoute(tctx, hr); err != nil {
 		acceptStatus.status = false
@@ -249,7 +249,7 @@ func (r *HTTPRouteReconciler) processHTTPRouteBackendRefs(tctx *provider.Transla
 		}
 
 		var service corev1.Service
-		if err := r.Get(context.TODO(), client.ObjectKey{
+		if err := r.Get(tctx, client.ObjectKey{
 			Namespace: namespace,
 			Name:      name,
 		}, &service); err != nil {
@@ -268,9 +268,13 @@ func (r *HTTPRouteReconciler) processHTTPRouteBackendRefs(tctx *provider.Transla
 			terr = fmt.Errorf("port %d not found in service %s", *backend.Port, name)
 			continue
 		}
+		tctx.Services[client.ObjectKey{
+			Namespace: namespace,
+			Name:      name,
+		}] = &service
 
 		endpointSliceList := new(discoveryv1.EndpointSliceList)
-		if err := r.List(context.TODO(), endpointSliceList,
+		if err := r.List(tctx, endpointSliceList,
 			client.InNamespace(namespace),
 			client.MatchingLabels{
 				discoveryv1.LabelServiceName: name,

internal/controller/httproute_controller.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/api7/api7-ingress-controller/api/v1alpha1"
 	networkingv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
@@ -19,6 +21,7 @@ const (
 	SecretIndexRef     = "secretRefs"
 	IngressClassRef    = "ingressClassRef"
 	ConsumerGatewayRef = "consumerGatewayRef"
+	PolicyTargetRefs   = "targetRefs"
 )
 
 func SetupIndexer(mgr ctrl.Manager) error {
@@ -34,6 +37,11 @@ func SetupIndexer(mgr ctrl.Manager) error {
 	if err := setupConsumerIndexer(mgr); err != nil {
 		return err
 	}
+	/*
+		if err := setupBackendTrafficPolicyIndexer(mgr); err != nil {
+			return err
+		}
+	*/
 	return nil
 }
 
@@ -175,6 +183,18 @@ func setupIngressIndexer(mgr ctrl.Manager) error {
 	return nil
 }
 
+func SetupBackendTrafficPolicyIndexer(mgr ctrl.Manager) error {
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&v1alpha1.BackendTrafficPolicy{},
+		PolicyTargetRefs,
+		BackendTrafficPolicyIndexFunc,
+	); err != nil {
+		return err
+	}
+	return nil
+}
+
 func IngressClassIndexFunc(rawObj client.Object) []string {
 	ingressClass := rawObj.(*networkingv1.IngressClass)
 	if ingressClass.Spec.Controller == "" {
@@ -227,6 +247,18 @@ func IngressSecretIndexFunc(rawObj client.Object) []string {
 	return secrets
 }
 
+func GenIndexKeyWithGK(group, kind, namespace, name string) string {
+	gvk := schema.GroupKind{
+		Group: group,
+		Kind:  kind,
+	}
+	nsName := types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	}
+	return gvk.String() + "/" + nsName.String()
+}
+
 func GenIndexKey(namespace, name string) string {
 	return client.ObjectKey{
 		Namespace: namespace,
@@ -292,3 +324,19 @@ func GatewayParametersRefIndexFunc(rawObj client.Object) []string {
 	}
 	return nil
 }
+
+func BackendTrafficPolicyIndexFunc(rawObj client.Object) []string {
+	btp := rawObj.(*v1alpha1.BackendTrafficPolicy)
+	keys := make([]string, 0, len(btp.Spec.TargetRefs))
+	for _, ref := range btp.Spec.TargetRefs {
+		keys = append(keys,
+			GenIndexKeyWithGK(
+				string(ref.Group),
+				string(ref.Kind),
+				btp.GetNamespace(),
+				string(ref.Name),
+			),
+		)
+	}
+	return keys
+}

internal/controller/indexer/indexer.go

@@ -93,16 +93,16 @@ func (r *IngressReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	r.Log.Info("reconciling ingress", "ingress", ingress.Name)
 
 	// create a translate context
-	tctx := provider.NewDefaultTranslateContext()
+	tctx := provider.NewDefaultTranslateContext(ctx)
 
 	// process TLS configuration
-	if err := r.processTLS(ctx, tctx, ingress); err != nil {
+	if err := r.processTLS(tctx, ingress); err != nil {
 		r.Log.Error(err, "failed to process TLS configuration", "ingress", ingress.Name)
 		return ctrl.Result{}, err
 	}
 
 	// process backend services
-	if err := r.processBackends(ctx, tctx, ingress); err != nil {
+	if err := r.processBackends(tctx, ingress); err != nil {
 		r.Log.Error(err, "failed to process backend services", "ingress", ingress.Name)
 		return ctrl.Result{}, err
 	}
@@ -295,14 +295,14 @@ func (r *IngressReconciler) listIngressesBySecret(ctx context.Context, obj clien
 }
 
 // processTLS process the TLS configuration of the ingress
-func (r *IngressReconciler) processTLS(ctx context.Context, tctx *provider.TranslateContext, ingress *networkingv1.Ingress) error {
+func (r *IngressReconciler) processTLS(tctx *provider.TranslateContext, ingress *networkingv1.Ingress) error {
 	for _, tls := range ingress.Spec.TLS {
 		if tls.SecretName == "" {
 			continue
 		}
 
 		secret := corev1.Secret{}
-		if err := r.Get(ctx, client.ObjectKey{
+		if err := r.Get(tctx, client.ObjectKey{
 			Namespace: ingress.Namespace,
 			Name:      tls.SecretName,
 		}, &secret); err != nil {
@@ -323,7 +323,7 @@ func (r *IngressReconciler) processTLS(ctx context.Context, tctx *provider.Trans
 }
 
 // processBackends process the backend services of the ingress
-func (r *IngressReconciler) processBackends(ctx context.Context, tctx *provider.TranslateContext, ingress *networkingv1.Ingress) error {
+func (r *IngressReconciler) processBackends(tctx *provider.TranslateContext, ingress *networkingv1.Ingress) error {
 	var terr error
 
 	// process all the backend services in the rules
@@ -336,7 +336,7 @@ func (r *IngressReconciler) processBackends(ctx context.Context, tctx *provider.
 				continue
 			}
 			service := path.Backend.Service
-			if err := r.processBackendService(ctx, tctx, ingress.Namespace, service); err != nil {
+			if err := r.processBackendService(tctx, ingress.Namespace, service); err != nil {
 				terr = err
 			}
 		}
@@ -345,10 +345,10 @@ func (r *IngressReconciler) processBackends(ctx context.Context, tctx *provider.
 }
 
 // processBackendService process a single backend service
-func (r *IngressReconciler) processBackendService(ctx context.Context, tctx *provider.TranslateContext, namespace string, backendService *networkingv1.IngressServiceBackend) error {
+func (r *IngressReconciler) processBackendService(tctx *provider.TranslateContext, namespace string, backendService *networkingv1.IngressServiceBackend) error {
 	// get the service
 	var service corev1.Service
-	if err := r.Get(ctx, client.ObjectKey{
+	if err := r.Get(tctx, client.ObjectKey{
 		Namespace: namespace,
 		Name:      backendService.Name,
 	}, &service); err != nil {
@@ -385,7 +385,7 @@ func (r *IngressReconciler) processBackendService(ctx context.Context, tctx *pro
 
 	// get the endpoint slices
 	endpointSliceList := &discoveryv1.EndpointSliceList{}
-	if err := r.List(ctx, endpointSliceList,
+	if err := r.List(tctx, endpointSliceList,
 		client.InNamespace(namespace),
 		client.MatchingLabels{
 			discoveryv1.LabelServiceName: backendService.Name,