feat: support backendtrafficpolicy for httproute

api/adc/types.go

@@ -126,9 +126,9 @@ type Route struct {
 }
 
 type Timeout struct {
-	Connect float64 `json:"connect"`
-	Read    float64 `json:"read"`
-	Send    float64 `json:"send"`
+	Connect int `json:"connect"`
+	Read    int `json:"read"`
+	Send    int `json:"send"`
 }
 
 type StreamRoute struct {
@@ -149,7 +149,7 @@ type Upstream struct {
 	HashOn       string        `json:"hash_on,omitempty" yaml:"hash_on,omitempty"`
 	Key          string        `json:"key,omitempty" yaml:"key,omitempty"`
 	Nodes        UpstreamNodes `json:"nodes" yaml:"nodes"`
-	PassHost     *PassHost     `json:"pass_host,omitempty" yaml:"pass_host,omitempty"`
+	PassHost     string        `json:"pass_host,omitempty" yaml:"pass_host,omitempty"`
 	Retries      *int64        `json:"retries,omitempty" yaml:"retries,omitempty"`
 	RetryTimeout *float64      `json:"retry_timeout,omitempty" yaml:"retry_timeout,omitempty"`
 	Scheme       string        `json:"scheme,omitempty" yaml:"scheme,omitempty"`
@@ -202,14 +202,6 @@ const (
 	Trace   Method = "TRACE"
 )
 
-type PassHost string
-
-const (
-	Node    PassHost = "node"
-	Pass    PassHost = "pass"
-	Rewrite PassHost = "rewrite"
-)
-
 type Scheme string
 
 const (

api/v1alpha1/backendtrafficpolicy_types.go

@@ -44,11 +44,11 @@ type BackendTrafficPolicySpec struct {
 	//
 	// +kubebuilder:validation:Enum=pass;node;rewrite;
 	// +kubebuilder:default=pass
-	PassHost string `json:"pass_host,omitempty" yaml:"pass_host,omitempty"`
+	PassHost string `json:"passHost,omitempty" yaml:"passHost,omitempty"`
 
 	// Specifies the host of the Upstream request. This is only valid if
-	// the pass_host is set to rewrite
-	Host Hostname `json:"upstream_host,omitempty" yaml:"upstream_host,omitempty"`
+	// the passHost is set to rewrite
+	Host Hostname `json:"upstreamHost,omitempty" yaml:"upstreamHost,omitempty"`
 }
 
 // LoadBalancer describes the load balancing parameters.
@@ -69,10 +69,16 @@ type LoadBalancer struct {
 
 type Timeout struct {
 	// +kubebuilder:default="60s"
+	// +kubebuilder:validation:Pattern=`^[0-9]+s$`
+	// +kubebuilder:validation:Type=string
 	Connect metav1.Duration `json:"connect,omitempty" yaml:"connect,omitempty"`
 	// +kubebuilder:default="60s"
+	// +kubebuilder:validation:Pattern=`^[0-9]+s$`
+	// +kubebuilder:validation:Type=string
 	Send metav1.Duration `json:"send,omitempty" yaml:"send,omitempty"`
 	// +kubebuilder:default="60s"
+	// +kubebuilder:validation:Pattern=`^[0-9]+s$`
+	// +kubebuilder:validation:Type=string
 	Read metav1.Duration `json:"read,omitempty" yaml:"read,omitempty"`
 }
 

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

@@ -70,7 +70,7 @@ spec:
                 type: object
                 x-kubernetes-validations:
                 - rule: '!(has(self.key) && self.type != ''chash'')'
-              pass_host:
+              passHost:
                 default: pass
                 description: |-
                   Configures the host when the request is forwarded to the upstream.
@@ -164,18 +164,21 @@ spec:
                 properties:
                   connect:
                     default: 60s
+                    pattern: ^[0-9]+s$
                     type: string
                   read:
                     default: 60s
+                    pattern: ^[0-9]+s$
                     type: string
                   send:
                     default: 60s
+                    pattern: ^[0-9]+s$
                     type: string
                 type: object
-              upstream_host:
+              upstreamHost:
                 description: |-
                   Specifies the host of the Upstream request. This is only valid if
-                  the pass_host is set to rewrite
+                  the passHost is set to rewrite
                 maxLength: 253
                 minLength: 1
                 pattern: ^(\*\.)?[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$

internal/controller/gateway_controller.go

@@ -75,6 +75,7 @@ func (r *GatewayReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 			if err := r.Provider.Delete(ctx, gateway); err != nil {
 				return ctrl.Result{}, err
 			}
+			return ctrl.Result{}, nil
 		}
 		return ctrl.Result{}, err
 	}
@@ -234,6 +235,9 @@ func (r *GatewayReconciler) listGatewaysForGatewayProxy(ctx context.Context, obj
 
 	recs := make([]reconcile.Request, 0, len(gatewayList.Items))
 	for _, gateway := range gatewayList.Items {
+		if !r.checkGatewayClass(&gateway) {
+			continue
+		}
 		recs = append(recs, reconcile.Request{
 			NamespacedName: client.ObjectKey{
 				Namespace: gateway.GetNamespace(),
@@ -244,7 +248,7 @@ func (r *GatewayReconciler) listGatewaysForGatewayProxy(ctx context.Context, obj
 	return recs
 }
 
-func (r *GatewayReconciler) listGatewaysForHTTPRoute(_ context.Context, obj client.Object) []reconcile.Request {
+func (r *GatewayReconciler) listGatewaysForHTTPRoute(ctx context.Context, obj client.Object) []reconcile.Request {
 	httpRoute, ok := obj.(*gatewayv1.HTTPRoute)
 	if !ok {
 		r.Log.Error(
@@ -268,6 +272,18 @@ func (r *GatewayReconciler) listGatewaysForHTTPRoute(_ context.Context, obj clie
 			gatewayNamespace = string(*parentRef.Namespace)
 		}
 
+		gateway := new(gatewayv1.Gateway)
+		if err := r.Get(ctx, client.ObjectKey{
+			Namespace: gatewayNamespace,
+			Name:      string(parentRef.Name),
+		}, gateway); err != nil {
+			continue
+		}
+
+		if !r.checkGatewayClass(gateway) {
+			continue
+		}
+
 		recs = append(recs, reconcile.Request{
 			NamespacedName: client.ObjectKey{
 				Namespace: gatewayNamespace,

internal/controller/httproute_controller.go

@@ -18,6 +18,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 
 	"github.com/api7/api7-ingress-controller/api/v1alpha1"
@@ -33,10 +34,14 @@ type HTTPRouteReconciler struct { //nolint:revive
 	Log logr.Logger
 
 	Provider provider.Provider
+
+	genericEvent chan event.GenericEvent
 }
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.genericEvent = make(chan event.GenericEvent, 100)
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&gatewayv1.HTTPRoute{}).
 		WithEventFilter(predicate.GenerationChangedPredicate{}).
@@ -65,9 +70,100 @@ func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				},
 			),
 		).
+		Watches(&v1alpha1.BackendTrafficPolicy{},
+			handler.EnqueueRequestsFromMapFunc(r.listHTTPRoutesForBackendTrafficPolicy),
+			builder.WithPredicates(
+				predicate.Funcs{
+					GenericFunc: func(e event.GenericEvent) bool {
+						return false
+					},
+					DeleteFunc: func(e event.DeleteEvent) bool {
+						return true
+					},
+					CreateFunc: func(e event.CreateEvent) bool {
+						return true
+					},
+					UpdateFunc: func(e event.UpdateEvent) bool {
+						oldObj, ok := e.ObjectOld.(*v1alpha1.BackendTrafficPolicy)
+						newObj, ok2 := e.ObjectNew.(*v1alpha1.BackendTrafficPolicy)
+						if !ok || !ok2 {
+							return false
+						}
+						oldRefs := oldObj.Spec.TargetRefs
+						newRefs := newObj.Spec.TargetRefs
+
+						oldRefMap := make(map[string]v1alpha1.BackendPolicyTargetReferenceWithSectionName)
+						for _, ref := range oldRefs {
+							key := fmt.Sprintf("%s/%s/%s", ref.Group, ref.Kind, ref.Name)
+							oldRefMap[key] = ref
+						}
+
+						for _, ref := range newRefs {
+							key := fmt.Sprintf("%s/%s/%s", ref.Group, ref.Kind, ref.Name)
+							delete(oldRefMap, key)
+						}
+						if len(oldRefMap) > 0 {
+							targetRefs := make([]v1alpha1.BackendPolicyTargetReferenceWithSectionName, 0, len(oldRefs))
+							for _, ref := range oldRefMap {
+								targetRefs = append(targetRefs, ref)
+							}
+							dump := oldObj.DeepCopy()
+							dump.Spec.TargetRefs = targetRefs
+							r.genericEvent <- event.GenericEvent{
+								Object: dump,
+							}
+						}
+						return true
+					},
+				},
+			),
+		).
+		WatchesRawSource(
+			source.Channel(
+				r.genericEvent,
+				handler.EnqueueRequestsFromMapFunc(r.listHTTPRouteForGenericEvent),
+			),
+		).
 		Complete(r)
 }
 
+func (r *HTTPRouteReconciler) listHTTPRouteForGenericEvent(ctx context.Context, obj client.Object) []reconcile.Request {
+	var namespacedNameMap = make(map[types.NamespacedName]struct{})
+	requests := []reconcile.Request{}
+	switch v := obj.(type) {
+	case *v1alpha1.BackendTrafficPolicy:
+		httprouteAll := []gatewayv1.HTTPRoute{}
+		for _, ref := range v.Spec.TargetRefs {
+			httprouteList := &gatewayv1.HTTPRouteList{}
+			if err := r.List(ctx, httprouteList, client.MatchingFields{
+				indexer.ServiceIndexRef: indexer.GenIndexKey(v.GetNamespace(), string(ref.Name)),
+			}); err != nil {
+				r.Log.Error(err, "failed to list HTTPRoutes for BackendTrafficPolicy", "namespace", v.GetNamespace(), "ref", ref.Name)
+				return nil
+			}
+			httprouteAll = append(httprouteAll, httprouteList.Items...)
+		}
+		for _, hr := range httprouteAll {
+			key := types.NamespacedName{
+				Namespace: hr.Namespace,
+				Name:      hr.Name,
+			}
+			if _, ok := namespacedNameMap[key]; !ok {
+				namespacedNameMap[key] = struct{}{}
+				requests = append(requests, reconcile.Request{
+					NamespacedName: client.ObjectKey{
+						Namespace: hr.Namespace,
+						Name:      hr.Name,
+					},
+				})
+			}
+		}
+	default:
+		r.Log.Error(fmt.Errorf("unexpected object type"), "failed to convert object to BackendTrafficPolicy")
+	}
+	return requests
+}
+
 func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	hr := new(gatewayv1.HTTPRoute)
 	if err := r.Get(ctx, req.NamespacedName, hr); err != nil {
@@ -114,6 +210,7 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 
 	tctx := provider.NewDefaultTranslateContext(ctx)
 
+	tctx.RouteParentRefs = hr.Spec.ParentRefs
 	rk := provider.ResourceKind{
 		Kind:      hr.Kind,
 		Namespace: hr.Namespace,
@@ -138,6 +235,8 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		}
 	}
 
+	ProcessBackendTrafficPolicy(r.Client, r.Log, tctx)
+
 	if err := r.Provider.Update(ctx, tctx, hr); err != nil {
 		acceptStatus.status = false
 		acceptStatus.msg = err.Error()
@@ -161,6 +260,7 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 	if err := r.Status().Update(ctx, hr); err != nil {
 		return ctrl.Result{}, err
 	}
+	UpdateStatus(r.Client, r.Log, tctx)
 	return ctrl.Result{}, nil
 }
 
@@ -220,6 +320,54 @@ 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...)
+	}
+	var namespacedNameMap = make(map[types.NamespacedName]struct{})
+	requests := make([]reconcile.Request, 0, len(httprouteList))
+	for _, hr := range httprouteList {
+		key := types.NamespacedName{
+			Namespace: hr.Namespace,
+			Name:      hr.Name,
+		}
+		if _, ok := namespacedNameMap[key]; !ok {
+			namespacedNameMap[key] = struct{}{}
+			requests = append(requests, reconcile.Request{
+				NamespacedName: client.ObjectKey{
+					Namespace: hr.Namespace,
+					Name:      hr.Name,
+				},
+			})
+		}
+	}
+	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

@@ -37,11 +37,9 @@ func SetupIndexer(mgr ctrl.Manager) error {
 	if err := setupConsumerIndexer(mgr); err != nil {
 		return err
 	}
-	/*
-		if err := setupBackendTrafficPolicyIndexer(mgr); err != nil {
-			return err
-		}
-	*/
+	if err := setupBackendTrafficPolicyIndexer(mgr); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -183,7 +181,7 @@ func setupIngressIndexer(mgr ctrl.Manager) error {
 	return nil
 }
 
-func SetupBackendTrafficPolicyIndexer(mgr ctrl.Manager) error {
+func setupBackendTrafficPolicyIndexer(mgr ctrl.Manager) error {
 	if err := mgr.GetFieldIndexer().IndexField(
 		context.Background(),
 		&v1alpha1.BackendTrafficPolicy{},