feat: support sync when update gateway proxy (#109)

Signed-off-by: ashing <[email protected]>

Author: ronething

.gitignore

@@ -33,4 +33,4 @@ api7-ingress-controller-conformance-report.yaml
 
 *.mdx
 .cursor/
-
+.env

Makefile

@@ -140,6 +140,15 @@ kind-load-images: pull-infra-images kind-load-ingress-image
 	@kind load docker-image kennethreitz/httpbin:latest --name $(KIND_NAME) 
 	@kind load docker-image jmalloc/echo-server:latest --name $(KIND_NAME)
 
+.PHONY: kind-load-gateway-image
+kind-load-gateway-image:
+	@kind load docker-image hkccr.ccs.tencentyun.com/api7-dev/api7-ee-3-gateway:dev --name $(KIND_NAME) 
+
+.PHONY: kind-load-dashboard-images
+kind-load-dashboard-images:
+	@kind load docker-image hkccr.ccs.tencentyun.com/api7-dev/api7-ee-dp-manager:$(DASHBOARD_VERSION)  --name $(KIND_NAME)
+	@kind load docker-image hkccr.ccs.tencentyun.com/api7-dev/api7-ee-3-integrated:$(DASHBOARD_VERSION)  --name $(KIND_NAME)
+
 .PHONY: kind-load-ingress-image
 kind-load-ingress-image:
 	@kind load docker-image $(IMG) --name $(KIND_NAME)

config/samples/config.yaml

@@ -7,10 +7,10 @@ controller_name: apisix.apache.org/api7-ingress-controller  # The controller nam
 leader_election_id: "api7-ingress-controller-leader" # The leader election ID for the API7 Ingress Controller.
                                                         # The default value is "api7-ingress-controller-leader".
 leader_election:
-  lease_duration: 15s                   # lease_duration is the duration that non-leader candidates will wait
+  lease_duration: 30s                   # lease_duration is the duration that non-leader candidates will wait
                                         # after observing a leadership renewal until attempting to acquire leadership of a
                                         # leader election.
-  renew_deadline: 10s                   # renew_deadline is the time in seconds that the acting controller
+  renew_deadline: 20s                   # renew_deadline is the time in seconds that the acting controller
                                         # will retry refreshing leadership before giving up.
   retry_period: 2s                      # retry_period is the time in seconds that the acting controller
                                         # will wait between tries of actions with the controller.

internal/controller/config/config.go

@@ -37,8 +37,8 @@ func NewDefaultConfig() *Config {
 
 func NewLeaderElection() *LeaderElection {
 	return &LeaderElection{
-		LeaseDuration: types.TimeDuration{Duration: 15 * time.Second},
-		RenewDeadline: types.TimeDuration{Duration: 10 * time.Second},
+		LeaseDuration: types.TimeDuration{Duration: 30 * time.Second},
+		RenewDeadline: types.TimeDuration{Duration: 20 * time.Second},
 		RetryPeriod:   types.TimeDuration{Duration: 2 * time.Second},
 		Disable:       false,
 	}

internal/controller/config/types.go

@@ -50,8 +50,8 @@ type ControlPlaneConfig struct {
 }
 
 type LeaderElection struct {
-	LeaseDuration types.TimeDuration `json:"leaseDuration,omitempty" yaml:"leaseDuration,omitempty"`
-	RenewDeadline types.TimeDuration `json:"renewDeadline,omitempty" yaml:"renewDeadline,omitempty"`
-	RetryPeriod   types.TimeDuration `json:"retryPeriod,omitempty" yaml:"retryPeriod,omitempty"`
+	LeaseDuration types.TimeDuration `json:"lease_duration,omitempty" yaml:"lease_duration,omitempty"`
+	RenewDeadline types.TimeDuration `json:"renew_deadline,omitempty" yaml:"renew_deadline,omitempty"`
+	RetryPeriod   types.TimeDuration `json:"retry_period,omitempty" yaml:"retry_period,omitempty"`
 	Disable       bool               `json:"disable,omitempty" yaml:"disable,omitempty"`
 }

internal/controller/consumer_controller.go

@@ -62,6 +62,9 @@ func (r *ConsumerReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Watches(&corev1.Secret{},
 			handler.EnqueueRequestsFromMapFunc(r.listConsumersForSecret),
 		).
+		Watches(&v1alpha1.GatewayProxy{},
+			handler.EnqueueRequestsFromMapFunc(r.listConsumersForGatewayProxy),
+		).
 		Complete(r)
 }
 
@@ -115,6 +118,49 @@ func (r *ConsumerReconciler) listConsumersForGateway(ctx context.Context, obj cl
 	return requests
 }
 
+func (r *ConsumerReconciler) listConsumersForGatewayProxy(ctx context.Context, obj client.Object) []reconcile.Request {
+	gatewayProxy, ok := obj.(*v1alpha1.GatewayProxy)
+	if !ok {
+		r.Log.Error(nil, "failed to convert to GatewayProxy", "object", obj)
+		return nil
+	}
+
+	namespace := gatewayProxy.GetNamespace()
+	name := gatewayProxy.GetName()
+
+	// find all gateways that reference this gateway proxy
+	gatewayList := &gatewayv1.GatewayList{}
+	if err := r.List(ctx, gatewayList, client.MatchingFields{
+		indexer.ParametersRef: indexer.GenIndexKey(namespace, name),
+	}); err != nil {
+		r.Log.Error(err, "failed to list gateways for gateway proxy", "gatewayproxy", gatewayProxy.GetName())
+		return nil
+	}
+
+	var requests []reconcile.Request
+
+	for _, gateway := range gatewayList.Items {
+		consumerList := &v1alpha1.ConsumerList{}
+		if err := r.List(ctx, consumerList, client.MatchingFields{
+			indexer.ConsumerGatewayRef: indexer.GenIndexKey(gateway.Namespace, gateway.Name),
+		}); err != nil {
+			r.Log.Error(err, "failed to list consumers for gateway", "gateway", gateway.Name)
+			continue
+		}
+
+		for _, consumer := range consumerList.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: client.ObjectKey{
+					Namespace: consumer.Namespace,
+					Name:      consumer.Name,
+				},
+			})
+		}
+	}
+
+	return requests
+}
+
 func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	consumer := new(v1alpha1.Consumer)
 	if err := r.Get(ctx, req.NamespacedName, consumer); err != nil {

internal/controller/httproute_controller.go

@@ -82,6 +82,9 @@ func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			handler.EnqueueRequestsFromMapFunc(r.listHTTPRouteByHTTPRoutePolicy),
 			builder.WithPredicates(httpRoutePolicyPredicateFuncs(r.genericEvent)),
 		).
+		Watches(&v1alpha1.GatewayProxy{},
+			handler.EnqueueRequestsFromMapFunc(r.listHTTPRoutesForGatewayProxy),
+		).
 		WatchesRawSource(
 			source.Channel(
 				r.genericEvent,
@@ -535,3 +538,48 @@ func httpRoutePolicyPredicateFuncs(channel chan event.GenericEvent) predicate.Pr
 		},
 	}
 }
+
+// listHTTPRoutesForGatewayProxy list all HTTPRoute resources that are affected by a given GatewayProxy
+func (r *HTTPRouteReconciler) listHTTPRoutesForGatewayProxy(ctx context.Context, obj client.Object) []reconcile.Request {
+	gatewayProxy, ok := obj.(*v1alpha1.GatewayProxy)
+	if !ok {
+		r.Log.Error(fmt.Errorf("unexpected object type"), "failed to convert object to GatewayProxy")
+		return nil
+	}
+
+	namespace := gatewayProxy.GetNamespace()
+	name := gatewayProxy.GetName()
+
+	// find all gateways that reference this gateway proxy
+	gatewayList := &gatewayv1.GatewayList{}
+	if err := r.List(ctx, gatewayList, client.MatchingFields{
+		indexer.ParametersRef: indexer.GenIndexKey(namespace, name),
+	}); err != nil {
+		r.Log.Error(err, "failed to list gateways for gateway proxy", "gatewayproxy", gatewayProxy.GetName())
+		return nil
+	}
+
+	var requests []reconcile.Request
+
+	// for each gateway, find all HTTPRoute resources that reference it
+	for _, gateway := range gatewayList.Items {
+		httpRouteList := &gatewayv1.HTTPRouteList{}
+		if err := r.List(ctx, httpRouteList, client.MatchingFields{
+			indexer.ParentRefs: indexer.GenIndexKey(gateway.Namespace, gateway.Name),
+		}); err != nil {
+			r.Log.Error(err, "failed to list httproutes for gateway", "gateway", gateway.Name)
+			continue
+		}
+
+		for _, httpRoute := range httpRouteList.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: client.ObjectKey{
+					Namespace: httpRoute.Namespace,
+					Name:      httpRoute.Name,
+				},
+			})
+		}
+	}
+
+	return requests
+}

internal/controller/indexer/indexer.go

@@ -14,15 +14,16 @@ import (
 )
 
 const (
-	ServiceIndexRef    = "serviceRefs"
-	ExtensionRef       = "extensionRef"
-	ParametersRef      = "parametersRef"
-	ParentRefs         = "parentRefs"
-	IngressClass       = "ingressClass"
-	SecretIndexRef     = "secretRefs"
-	IngressClassRef    = "ingressClassRef"
-	ConsumerGatewayRef = "consumerGatewayRef"
-	PolicyTargetRefs   = "targetRefs"
+	ServiceIndexRef           = "serviceRefs"
+	ExtensionRef              = "extensionRef"
+	ParametersRef             = "parametersRef"
+	ParentRefs                = "parentRefs"
+	IngressClass              = "ingressClass"
+	SecretIndexRef            = "secretRefs"
+	IngressClassRef           = "ingressClassRef"
+	IngressClassParametersRef = "ingressClassParametersRef"
+	ConsumerGatewayRef        = "consumerGatewayRef"
+	PolicyTargetRefs          = "targetRefs"
 )
 
 func SetupIndexer(mgr ctrl.Manager) error {
@@ -189,6 +190,16 @@ func setupIngressIndexer(mgr ctrl.Manager) error {
 		return err
 	}
 
+	// create IngressClassParametersRef index
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&networkingv1.IngressClass{},
+		IngressClassParametersRef,
+		IngressClassParametersRefIndexFunc,
+	); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -366,3 +377,19 @@ func BackendTrafficPolicyIndexFunc(rawObj client.Object) []string {
 	}
 	return keys
 }
+
+func IngressClassParametersRefIndexFunc(rawObj client.Object) []string {
+	ingressClass := rawObj.(*networkingv1.IngressClass)
+	// check if the IngressClass references this gateway proxy
+	if ingressClass.Spec.Parameters != nil &&
+		ingressClass.Spec.Parameters.APIGroup != nil &&
+		*ingressClass.Spec.Parameters.APIGroup == v1alpha1.GroupVersion.Group &&
+		ingressClass.Spec.Parameters.Kind == "GatewayProxy" {
+		ns := ingressClass.GetNamespace()
+		if ingressClass.Spec.Parameters.Namespace != nil {
+			ns = *ingressClass.Spec.Parameters.Namespace
+		}
+		return []string{GenIndexKey(ns, ingressClass.Spec.Parameters.Name)}
+	}
+	return nil
+}

internal/controller/ingress_controller.go

@@ -83,6 +83,9 @@ func (r *IngressReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			handler.EnqueueRequestsFromMapFunc(r.listIngressesByHTTPRoutePolicy),
 			builder.WithPredicates(httpRoutePolicyPredicateFuncs(r.genericEvent)),
 		).
+		Watches(&v1alpha1.GatewayProxy{},
+			handler.EnqueueRequestsFromMapFunc(r.listIngressesForGatewayProxy),
+		).
 		WatchesRawSource(
 			source.Channel(
 				r.genericEvent,
@@ -219,38 +222,8 @@ func (r *IngressReconciler) getIngressClass(obj client.Object) (*networkingv1.In
 
 // checkIngressClass check if the ingress uses the ingress class that we control
 func (r *IngressReconciler) checkIngressClass(obj client.Object) bool {
-	ingress := obj.(*networkingv1.Ingress)
-
-	if ingress.Spec.IngressClassName == nil {
-		// handle the case where IngressClassName is not specified
-		// find all ingress classes and check if any of them is marked as default
-		ingressClassList := &networkingv1.IngressClassList{}
-		if err := r.List(context.Background(), ingressClassList, client.MatchingFields{
-			indexer.IngressClass: config.GetControllerName(),
-		}); err != nil {
-			r.Log.Error(err, "failed to list ingress classes")
-			return false
-		}
-
-		// find the ingress class that is marked as default
-		for _, ic := range ingressClassList.Items {
-			if IsDefaultIngressClass(&ic) && matchesController(ic.Spec.Controller) {
-				log.Debugw("match the default ingress class")
-				return true
-			}
-		}
-
-		log.Debugw("no default ingress class found")
-		return false
-	}
-
-	// check if the ingress class is controlled by us
-	ingressClass := networkingv1.IngressClass{}
-	if err := r.Client.Get(context.Background(), client.ObjectKey{Name: *ingress.Spec.IngressClassName}, &ingressClass); err != nil {
-		return false
-	}
-
-	return matchesController(ingressClass.Spec.Controller)
+	_, err := r.getIngressClass(obj)
+	return err == nil
 }
 
 // matchesIngressController check if the ingress class is controlled by us
@@ -735,3 +708,46 @@ func (r *IngressReconciler) processIngressClassParameters(ctx context.Context, t
 
 	return nil
 }
+
+// listIngressesForGatewayProxy list all ingresses that use a specific gateway proxy
+func (r *IngressReconciler) listIngressesForGatewayProxy(ctx context.Context, obj client.Object) []reconcile.Request {
+	gatewayProxy, ok := obj.(*v1alpha1.GatewayProxy)
+	if !ok {
+		r.Log.Error(fmt.Errorf("unexpected object type"), "failed to convert object to GatewayProxy")
+		return nil
+	}
+
+	// find all ingress classes that reference this gateway proxy
+	ingressClassList := &networkingv1.IngressClassList{}
+	if err := r.List(ctx, ingressClassList, client.MatchingFields{
+		indexer.IngressClassParametersRef: indexer.GenIndexKey(gatewayProxy.GetNamespace(), gatewayProxy.GetName()),
+	}); err != nil {
+		r.Log.Error(err, "failed to list ingress classes for gateway proxy", "gatewayproxy", gatewayProxy.GetName())
+		return nil
+	}
+
+	var requests []reconcile.Request
+
+	for _, ingressClass := range ingressClassList.Items {
+		requests = append(requests, r.listIngressForIngressClass(ctx, &ingressClass)...)
+	}
+
+	// the requests may contain duplicates, distinct the requests
+	requests = distinctRequests(requests)
+
+	return requests
+}
+
+// distinctRequests distinct the requests
+func distinctRequests(requests []reconcile.Request) []reconcile.Request {
+	uniqueRequests := make(map[string]reconcile.Request)
+	for _, request := range requests {
+		uniqueRequests[request.String()] = request
+	}
+
+	distinctRequests := make([]reconcile.Request, 0, len(uniqueRequests))
+	for _, request := range uniqueRequests {
+		distinctRequests = append(distinctRequests, request)
+	}
+	return distinctRequests
+}

internal/provider/adc/adc.go

@@ -32,10 +32,9 @@ type adcConfig struct {
 type adcClient struct {
 	sync.Mutex
 
-	translator   *translator.Translator
-	ServerAddr   string
-	Token        string
-	GatewayGroup string
+	execLock sync.Mutex
+
+	translator *translator.Translator
 	// gateway/ingressclass -> adcConfig
 	configs map[provider.ResourceKind]adcConfig
 	// httproute/consumer/ingress/gateway -> gateway/ingressclass
@@ -164,6 +163,7 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 	}
 
 	configs := d.getConfigs(rk)
+	defer d.deleteConfigs(rk)
 
 	err := d.sync(ctx, Task{
 		Name:          obj.GetName(),
@@ -175,7 +175,6 @@ func (d *adcClient) Delete(ctx context.Context, obj client.Object) error {
 		return err
 	}
 
-	d.deleteConfigs(rk)
 	return nil
 }
 
@@ -230,17 +229,14 @@ func (d *adcClient) sync(ctx context.Context, task Task) error {
 }
 
 func (d *adcClient) execADC(ctx context.Context, config adcConfig, args []string) error {
+	d.execLock.Lock()
+	defer d.execLock.Unlock()
+
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, d.syncTimeout)
 	defer cancel()
 	// todo: use adc config
-	serverAddr := d.ServerAddr
-	if config.ServerAddr != "" {
-		serverAddr = config.ServerAddr
-	}
-	token := d.Token
-	if config.Token != "" {
-		token = config.Token
-	}
+	serverAddr := config.ServerAddr
+	token := config.Token
 
 	adcEnv := []string{
 		"ADC_EXPERIMENTAL_FEATURE_FLAGS=remote-state-file,parallel-backend-request",