feat: support backendtrafficpolicy for httproute

Author: AlinsRan

api/adc/types.go

@@ -128,9 +128,9 @@ type Route struct {
 }
 
 type Timeout struct {
-	Connect *int64 `json:"connect,omitempty"`
-	Read    *int64 `json:"read,omitempty"`
-	Send    *int64 `json:"send,omitempty"`
+	Connect int64 `json:"connect"`
+	Read    int64 `json:"read"`
+	Send    int64 `json:"send"`
 }
 
 type StreamRoute struct {

api/v1alpha1/backendtrafficpolicy_types.go

@@ -72,9 +72,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

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

@@ -168,10 +168,13 @@ spec:
                   upstream.
                 properties:
                   connect:
+                    default: 60s
                     type: string
                   read:
+                    default: 60s
                     type: string
                   send:
+                    default: 60s
                     type: string
                 type: object
               upstream_host:

internal/controller/httproute_controller.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/go-logr/logr"
+	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1 "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -23,6 +24,7 @@ import (
 	"github.com/api7/api7-ingress-controller/api/v1alpha1"
 	"github.com/api7/api7-ingress-controller/internal/controller/indexer"
 	"github.com/api7/api7-ingress-controller/internal/provider"
+	"github.com/api7/gopkg/pkg/log"
 )
 
 // HTTPRouteReconciler reconciles a GatewayClass object.
@@ -65,6 +67,9 @@ func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				},
 			),
 		).
+		Watches(&v1alpha1.BackendTrafficPolicy{},
+			handler.EnqueueRequestsFromMapFunc(r.listHTTPRoutesForBackendTrafficPolicy),
+		).
 		Complete(r)
 }
 
@@ -126,6 +131,8 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		}
 	}
 
+	ProcessBackendTrafficPolicy(r.Client, tctx)
+
 	if err := r.Provider.Update(ctx, tctx, hr); err != nil {
 		acceptStatus.status = false
 		acceptStatus.msg = err.Error()
@@ -208,6 +215,48 @@ func (r *HTTPRouteReconciler) listHTTPRoutesByExtensionRef(ctx context.Context,
 	return requests
 }
 
+func (r *HTTPRouteReconciler) listHTTPRoutesForBackendTrafficPolicy(ctx context.Context, obj client.Object) []reconcile.Request {
+	policy, ok := obj.(*v1alpha1.BackendTrafficPolicy)
+	if !ok {
+		r.Log.Error(fmt.Errorf("unexpected object type"), "failed to convert object to BackendTrafficPolicy")
+		return nil
+	}
+
+	httprouteList := []gatewayv1.HTTPRoute{}
+	for _, targetRef := range policy.Spec.TargetRefs {
+		service := &corev1.Service{}
+		if err := r.Get(ctx, client.ObjectKey{
+			Namespace: policy.Namespace,
+			Name:      string(targetRef.Name),
+		}, service); err != nil {
+			if client.IgnoreNotFound(err) != nil {
+				r.Log.Error(err, "failed to get service", "namespace", policy.Namespace, "name", targetRef.Name)
+			}
+			continue
+		}
+		hrList := &gatewayv1.HTTPRouteList{}
+		if err := r.List(ctx, hrList, client.MatchingFields{
+			indexer.ServiceIndexRef: indexer.GenIndexKey(policy.Namespace, string(targetRef.Name)),
+		}); err != nil {
+			r.Log.Error(err, "failed to list httproutes by service reference", "service", targetRef.Name)
+			return nil
+		}
+		httprouteList = append(httprouteList, hrList.Items...)
+	}
+
+	requests := make([]reconcile.Request, 0, len(httprouteList))
+	for _, hr := range httprouteList {
+		requests = append(requests, reconcile.Request{
+			NamespacedName: client.ObjectKey{
+				Namespace: hr.Namespace,
+				Name:      hr.Name,
+			},
+		})
+	}
+	log.Errorw("list httproutes for backend traffic policy", zap.Any("httproutes", requests))
+	return requests
+}
+
 func (r *HTTPRouteReconciler) listHTTPRoutesForGateway(ctx context.Context, obj client.Object) []reconcile.Request {
 	gateway, ok := obj.(*gatewayv1.Gateway)
 	if !ok {

internal/controller/indexer/indexer.go

@@ -131,6 +131,15 @@ func setupHTTPRouteIndexer(mgr ctrl.Manager) error {
 	); err != nil {
 		return err
 	}
+
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&v1alpha1.BackendTrafficPolicy{},
+		PolicyTargetRefs,
+		BackendTrafficPolicyIndexFunc,
+	); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -314,7 +323,7 @@ func BackendTrafficPolicyIndexFunc(rawObj client.Object) []string {
 	for _, ref := range btp.Spec.TargetRefs {
 		keys = append(keys,
 			GenIndexKeyWithGK(
-				v1alpha1.GroupVersion.Group,
+				string(ref.Group),
 				string(ref.Kind),
 				btp.GetNamespace(),
 				string(ref.Name),

internal/controller/policies.go

@@ -10,6 +10,7 @@ import (
 	"github.com/api7/api7-ingress-controller/internal/controller/config"
 	"github.com/api7/api7-ingress-controller/internal/controller/indexer"
 	"github.com/api7/api7-ingress-controller/internal/provider"
+	"github.com/go-logr/logr"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -25,7 +26,9 @@ func (p PolicyTargetKey) String() string {
 	return p.NsName.String() + "/" + p.GroupKind.String()
 }
 
-func ProcessBackendTrafficPolicy(c client.Client, tctx *provider.TranslateContext) {
+func ProcessBackendTrafficPolicy(c client.Client,
+	log logr.Logger,
+	tctx *provider.TranslateContext) {
 	conflicts := map[string]v1alpha1.BackendTrafficPolicy{}
 	for _, service := range tctx.Services {
 		backendTrafficPolicyList := &v1alpha1.BackendTrafficPolicyList{}
@@ -34,10 +37,8 @@ func ProcessBackendTrafficPolicy(c client.Client, tctx *provider.TranslateContex
 				indexer.PolicyTargetRefs: indexer.GenIndexKeyWithGK("", "Service", service.Namespace, service.Name),
 			},
 		); err != nil {
-			if client.IgnoreNotFound(err) == nil {
-				continue
-			}
-			return
+			log.Error(err, "failed to list BackendTrafficPolicy for Service")
+			continue
 		}
 		if len(backendTrafficPolicyList.Items) == 0 {
 			continue

internal/manager/run.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"crypto/tls"
 	"os"
+	"time"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
@@ -100,6 +102,9 @@ func Run(ctx context.Context, logger logr.Logger) error {
 		LeaderElection:          true,
 		LeaderElectionID:        cfg.LeaderElectionID,
 		LeaderElectionNamespace: namespace,
+		LeaseDuration:           ptr.To(time.Second * 15),
+		RenewDeadline:           ptr.To(time.Second * 10),
+		RetryPeriod:             ptr.To(time.Second * 5),
 		// LeaderElectionReleaseOnCancel defines if the leader should step down voluntarily
 		// when the Manager ends. This requires the binary to immediately end when the
 		// Manager is stopped, otherwise, this setting is unsafe. Setting this significantly

internal/provider/adc/translator/httproute.go

@@ -288,6 +288,13 @@ func (t *Translator) TranslateHTTPRoute(tctx *provider.TranslateContext, httpRou
 				backend.Namespace = &namespace
 			}
 			upNodes := t.translateBackendRef(tctx, backend.BackendRef)
+			for _, po := range tctx.BackendTrafficPolicies {
+				log.Errorw("backend traffic policy",
+					zap.String("policy", po.Name),
+					zap.String("namespace", po.Namespace))
+			}
+
+			t.AttachBackendTrafficPolicyToUpstream(backend.BackendRef, tctx.BackendTrafficPolicies, upstream)
 			upstream.Nodes = append(upstream.Nodes, upNodes...)
 		}
 		t.attachBackendTrafficPolicyToUpstream(nil, upstream)

internal/provider/adc/translator/policies.go

@@ -4,18 +4,29 @@ import (
 	"github.com/api7/api7-ingress-controller/api/adc"
 	"github.com/api7/api7-ingress-controller/api/v1alpha1"
 	"k8s.io/apimachinery/pkg/types"
+	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 
 	adctypes "github.com/api7/api7-ingress-controller/api/adc"
 )
 
-func (t *Translator) AttachBackendTrafficPolicyToUpstream(policies map[types.NamespacedName]*v1alpha1.BackendTrafficPolicy, upstream *adctypes.Upstream) {
+func (t *Translator) AttachBackendTrafficPolicyToUpstream(ref gatewayv1.BackendRef, policies map[types.NamespacedName]*v1alpha1.BackendTrafficPolicy, upstream *adctypes.Upstream) {
 	if len(policies) == 0 {
 		return
 	}
-	for _, policy := range policies {
-		t.attachBackendTrafficPolicyToUpstream(policy, upstream)
+	var policy *v1alpha1.BackendTrafficPolicy
+	for _, po := range policies {
+		for _, targetRef := range po.Spec.TargetRefs {
+			if ref.Name == targetRef.Name &&
+				(ref.Namespace != nil && string(*ref.Namespace) == po.Namespace) {
+				policy = po
+				break
+			}
+		}
 	}
-
+	if policy == nil {
+		return
+	}
+	t.attachBackendTrafficPolicyToUpstream(policy, upstream)
 }
 
 func (t *Translator) attachBackendTrafficPolicyToUpstream(policy *v1alpha1.BackendTrafficPolicy, upstream *adctypes.Upstream) {
@@ -29,27 +40,10 @@ func (t *Translator) attachBackendTrafficPolicyToUpstream(policy *v1alpha1.Backe
 		*upstream.Retries = int64(*policy.Spec.Retries)
 	}
 	if policy.Spec.Timeout != nil {
-		var (
-			connect *int64
-			read    *int64
-			send    *int64
-		)
-		if policy.Spec.Timeout.Connect.Duration > 0 {
-			connect = new(int64)
-			*connect = policy.Spec.Timeout.Connect.Duration.Milliseconds()
-		}
-		if policy.Spec.Timeout.Read.Duration > 0 {
-			read = new(int64)
-			*read = policy.Spec.Timeout.Read.Duration.Milliseconds()
-		}
-		if policy.Spec.Timeout.Send.Duration > 0 {
-			send = new(int64)
-			*send = policy.Spec.Timeout.Send.Duration.Milliseconds()
-		}
 		upstream.Timeout = &adctypes.Timeout{
-			Connect: connect,
-			Read:    read,
-			Send:    send,
+			Connect: policy.Spec.Timeout.Connect.Duration.Milliseconds(),
+			Read:    policy.Spec.Timeout.Read.Duration.Milliseconds(),
+			Send:    policy.Spec.Timeout.Send.Duration.Milliseconds(),
 		}
 	}
 	if policy.Spec.LoadBalancer != nil {