feat: support backendtrafficpolicy for httproute

AlinsRan · AlinsRan · commit 97e302b7516b · 2025-04-17T14:58:04.000+08:00
diff --git a/api/adc/types.go b/api/adc/types.go
@@ -126,9 +126,9 @@ type Route struct {
 }
 
 type Timeout struct {
-	Connect int64 `json:"connect"`
-	Read    int64 `json:"read"`
-	Send    int64 `json:"send"`
+	Connect int `json:"connect"`
+	Read    int `json:"read"`
+	Send    int `json:"send"`
 }
 
 type StreamRoute struct {
diff --git a/api/v1alpha1/backendtrafficpolicy_types.go b/api/v1alpha1/backendtrafficpolicy_types.go
@@ -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"`
 }
 
diff --git a/config/crd/bases/gateway.apisix.io_backendtrafficpolicies.yaml b/config/crd/bases/gateway.apisix.io_backendtrafficpolicies.yaml
@@ -164,12 +164,15 @@ 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:
diff --git a/internal/controller/gateway_controller.go b/internal/controller/gateway_controller.go
@@ -76,6 +76,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
 	}
@@ -222,6 +223,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(),
@@ -232,7 +236,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(
@@ -256,6 +260,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,
diff --git a/internal/controller/httproute_controller.go b/internal/controller/httproute_controller.go
@@ -6,7 +6,6 @@ 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"
@@ -19,12 +18,12 @@ 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"
 	"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.
@@ -35,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{}).
@@ -69,10 +72,91 @@ 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
+
+						// 将旧引用转换为 Map
+						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 {
+	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 {
+			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 {
@@ -131,7 +215,7 @@ func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		}
 	}
 
-	ProcessBackendTrafficPolicy(r.Client, tctx)
+	ProcessBackendTrafficPolicy(r.Client, r.Log, tctx)
 
 	if err := r.Provider.Update(ctx, tctx, hr); err != nil {
 		acceptStatus.status = false
@@ -156,6 +240,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
 }
 
@@ -253,7 +338,10 @@ func (r *HTTPRouteReconciler) listHTTPRoutesForBackendTrafficPolicy(ctx context.
 			},
 		})
 	}
-	log.Errorw("list httproutes for backend traffic policy", zap.Any("httproutes", requests))
+	if !policy.GetDeletionTimestamp().IsZero() {
+		// If the policy is deleted, we need to list all HTTPRoutes that reference this policy
+		// and add them to the requests.
+	}
 	return requests
 }
 
diff --git a/internal/controller/source.go b/internal/controller/source.go
@@ -0,0 +1,121 @@
+package controller
+
+/*
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/api7/api7-ingress-controller/api/v1alpha1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
+)
+
+type TargetRefsChangedPredicate struct {
+	predicate.Funcs
+	DeletedRefs map[string]struct{} // 存储被删除的引用
+}
+
+// Update 事件处理
+func (t *TargetRefsChangedPredicate) Update(e event.UpdateEvent) bool {
+	oldPolicy, okOld := e.ObjectOld.(*v1alpha1.BackendTrafficPolicy)
+	newPolicy, okNew := e.ObjectNew.(*v1alpha1.BackendTrafficPolicy)
+	if !okOld || !okNew {
+		return false
+	}
+
+	// 计算被删除的引用
+	oldRefs := oldPolicy.Spec.TargetRefs
+	newRefs := newPolicy.Spec.TargetRefs
+
+	// 将旧引用转换为 Map
+	oldRefMap := make(map[string]struct{})
+	for _, ref := range oldRefs {
+		key := fmt.Sprintf("%s/%s", ref.Kind, ref.Name)
+		oldRefMap[key] = struct{}{}
+	}
+
+	// 找出被删除的引用
+	t.DeletedRefs = make(map[string]struct{})
+	for _, ref := range newRefs {
+		key := fmt.Sprintf("%s/%s", ref.Kind, ref.Name)
+		delete(oldRefMap, key)
+	}
+	for key := range oldRefMap {
+		t.DeletedRefs[key] = struct{}{}
+	}
+
+	return len(t.DeletedRefs) > 0
+}
+
+type DeletedRefEventSource struct {
+	Client    client.Client
+	Predicate *TargetRefsChangedPredicate
+}
+
+// Start 实现 Source 接口
+func (s *DeletedRefEventSource) Start(
+	ctx context.Context,
+	handler handler.EventHandler,
+	queue workqueue.RateLimitingInterface,
+	predicates ...predicate.Predicate,
+) error {
+	// 监听 BackendTrafficPolicy 的 Update 事件（已通过 Predicate 过滤）
+	// 此处假设 Predicate 已捕获到被删除的引用
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				if len(s.Predicate.DeletedRefs) == 0 {
+					time.Sleep(1 * time.Second)
+					continue
+				}
+
+				// 遍历被删除的引用，查找关联的 HTTPRoute
+				for refKey := range s.Predicate.DeletedRefs {
+					parts := strings.Split(refKey, "/")
+					if len(parts) != 2 {
+						continue
+					}
+					kind, name := parts[0], parts[1]
+
+					// 查找关联的 HTTPRoute
+					var routes gatewayv1.HTTPRouteList
+					if err := s.Client.List(
+						context.Background(),
+						&routes,
+						client.MatchingFields{"targetRefs": refKey},
+					); err != nil {
+						continue
+					}
+
+					// 生成调和请求
+					for _, route := range routes.Items {
+						req := reconcile.Request{
+							NamespacedName: types.NamespacedName{
+								Name:      route.Name,
+								Namespace: route.Namespace,
+							},
+						}
+						handler.Generic(ctx, event.GenericEvent{Object: &route}, queue)
+					}
+				}
+
+				// 清空已处理的引用
+				s.Predicate.DeletedRefs = make(map[string]struct{})
+				time.Sleep(5 * time.Second) // 避免高频触发
+			}
+		}
+	}()
+	return nil
+}
+*/
diff --git a/internal/provider/adc/adc.go b/internal/provider/adc/adc.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 	"os"
 	"os/exec"
+	"runtime/debug"
 
 	"go.uber.org/zap"
 	networkingv1 "k8s.io/api/networking/v1"
@@ -90,7 +91,7 @@ func (d *adcClient) Update(ctx context.Context, tctx *provider.TranslateContext,
 }
 
 func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
-	log.Debugw("deleting object", zap.Any("object", obj))
+	log.Debugw("deleting object", zap.Any("object", obj), zap.String("stack", string(debug.Stack())))
 
 	var resourceTypes []string
 	var labels map[string]string
diff --git a/internal/provider/adc/translator/httproute.go b/internal/provider/adc/translator/httproute.go
diff --git a/internal/provider/adc/translator/policies.go b/internal/provider/adc/translator/policies.go

Original file line number	Diff line number	Diff line change
`@@ -126,9 +126,9 @@ type Route struct {`
`126`	`126`	`}`
`127`	`127`
`128`	`128`	`type Timeout struct {`
`129`		- Connect int64 `json:"connect"`
`130`		- Read int64 `json:"read"`
`131`		- Send int64 `json:"send"`
	`129`	+ Connect int `json:"connect"`
	`130`	+ Read int `json:"read"`
	`131`	+ Send int `json:"send"`
`132`	`132`	`}`
`133`	`133`
`134`	`134`	`type StreamRoute struct {`