fix: full sync during restart results in loss of dataplane traffic (#2489)

internal/controller/apisixconsumer_controller.go

@@ -39,6 +39,7 @@ import (
 	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
@@ -51,10 +52,12 @@ type ApisixConsumerReconciler struct {
 
 	Provider provider.Provider
 	Updater  status.Updater
+	Readier  readiness.ReadinessManager
 }
 
 // Reconcile FIXME: implement the reconcile logic (For now, it dose nothing other than directly accepting)
 func (r *ApisixConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&apiv2.ApisixConsumer{}, req.NamespacedName)
 	r.Log.Info("reconcile", "request", req.NamespacedName)
 
 	ac := &apiv2.ApisixConsumer{}

internal/controller/apisixglobalrule_controller.go

@@ -38,6 +38,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/internal/controller/config"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
@@ -49,10 +50,13 @@ type ApisixGlobalRuleReconciler struct {
 	Log      logr.Logger
 	Provider provider.Provider
 	Updater  status.Updater
+
+	Readier readiness.ReadinessManager
 }
 
 // Reconcile implements the reconciliation logic for ApisixGlobalRule
 func (r *ApisixGlobalRuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&apiv2.ApisixGlobalRule{}, req.NamespacedName)
 	var globalRule apiv2.ApisixGlobalRule
 	if err := r.Get(ctx, req.NamespacedName, &globalRule); err != nil {
 		if client.IgnoreNotFound(err) == nil {

internal/controller/apisixroute_controller.go

@@ -44,6 +44,7 @@ import (
 	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/types"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
@@ -57,6 +58,7 @@ type ApisixRouteReconciler struct {
 	Log      logr.Logger
 	Provider provider.Provider
 	Updater  status.Updater
+	Readier  readiness.ReadinessManager
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -97,6 +99,7 @@ func (r *ApisixRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 }
 
 func (r *ApisixRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&apiv2.ApisixRoute{}, req.NamespacedName)
 	var ar apiv2.ApisixRoute
 	if err := r.Get(ctx, req.NamespacedName, &ar); err != nil {
 		if client.IgnoreNotFound(err) == nil {

internal/controller/apisixtls_controller.go

@@ -39,6 +39,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/internal/controller/config"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
@@ -50,6 +51,7 @@ type ApisixTlsReconciler struct {
 	Log      logr.Logger
 	Provider provider.Provider
 	Updater  status.Updater
+	Readier  readiness.ReadinessManager
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -85,6 +87,7 @@ func (r *ApisixTlsReconciler) SetupWithManager(mgr ctrl.Manager) error {
 
 // Reconcile implements the reconciliation logic for ApisixTls
 func (r *ApisixTlsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&apiv2.ApisixTls{}, req.NamespacedName)
 	var tls apiv2.ApisixTls
 	if err := r.Get(ctx, req.NamespacedName, &tls); err != nil {
 		if client.IgnoreNotFound(err) == nil {

internal/controller/consumer_controller.go

@@ -38,6 +38,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/api/v1alpha1"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
@@ -51,6 +52,7 @@ type ConsumerReconciler struct { //nolint:revive
 	Provider provider.Provider
 
 	Updater status.Updater
+	Readier readiness.ReadinessManager
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -181,6 +183,7 @@ func (r *ConsumerReconciler) listConsumersForGatewayProxy(ctx context.Context, o
 }
 
 func (r *ConsumerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&v1alpha1.Consumer{}, req.NamespacedName)
 	consumer := new(v1alpha1.Consumer)
 	if err := r.Get(ctx, req.NamespacedName, consumer); err != nil {
 		if client.IgnoreNotFound(err) == nil {
@@ -303,31 +306,5 @@ func (r *ConsumerReconciler) checkGatewayRef(object client.Object) bool {
 	if !ok {
 		return false
 	}
-	if consumer.Spec.GatewayRef.Name == "" {
-		return false
-	}
-	if consumer.Spec.GatewayRef.Kind != nil && *consumer.Spec.GatewayRef.Kind != KindGateway {
-		return false
-	}
-	if consumer.Spec.GatewayRef.Group != nil && *consumer.Spec.GatewayRef.Group != gatewayv1.GroupName {
-		return false
-	}
-	ns := consumer.GetNamespace()
-	if consumer.Spec.GatewayRef.Namespace != nil {
-		ns = *consumer.Spec.GatewayRef.Namespace
-	}
-	gateway := &gatewayv1.Gateway{}
-	if err := r.Get(context.Background(), client.ObjectKey{
-		Name:      consumer.Spec.GatewayRef.Name,
-		Namespace: ns,
-	}, gateway); err != nil {
-		r.Log.Error(err, "failed to get gateway", "gateway", consumer.Spec.GatewayRef.Name)
-		return false
-	}
-	gatewayClass := &gatewayv1.GatewayClass{}
-	if err := r.Get(context.Background(), client.ObjectKey{Name: string(gateway.Spec.GatewayClassName)}, gatewayClass); err != nil {
-		r.Log.Error(err, "failed to get gateway class", "gateway", gateway.GetName(), "gatewayclass", gateway.Spec.GatewayClassName)
-		return false
-	}
-	return matchesController(string(gatewayClass.Spec.ControllerName))
+	return MatchConsumerGatewayRef(context.Background(), r.Client, r.Log, consumer)
 }

internal/controller/httproute_controller.go

@@ -48,6 +48,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/api/v1alpha1"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/types"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
@@ -65,6 +66,7 @@ type HTTPRouteReconciler struct { //nolint:revive
 	genericEvent chan event.GenericEvent
 
 	Updater status.Updater
+	Readier readiness.ReadinessManager
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -130,6 +132,7 @@ func (r *HTTPRouteReconciler) SetupWithManager(mgr ctrl.Manager) error {
 }
 
 func (r *HTTPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&gatewayv1.HTTPRoute{}, req.NamespacedName)
 	hr := new(gatewayv1.HTTPRoute)
 	if err := r.Get(ctx, req.NamespacedName, hr); err != nil {
 		if client.IgnoreNotFound(err) == nil {

internal/controller/indexer/indexer.go

@@ -47,6 +47,7 @@ const (
 	GatewayClassIndexRef      = "gatewayClassRef"
 	ApisixUpstreamRef         = "apisixUpstreamRef"
 	PluginConfigIndexRef      = "pluginConfigRefs"
+	ControllerName            = "controllerName"
 )
 
 func SetupIndexer(mgr ctrl.Manager) error {
@@ -59,11 +60,11 @@ func SetupIndexer(mgr ctrl.Manager) error {
 		setupIngressClassIndexer,
 		setupGatewayProxyIndexer,
 		setupGatewaySecretIndex,
-		setupGatewayClassIndexer,
 		setupApisixRouteIndexer,
 		setupApisixPluginConfigIndexer,
 		setupApisixTlsIndexer,
 		setupApisixConsumerIndexer,
+		setupGatewayClassIndexer,
 	} {
 		if err := setup(mgr); err != nil {
 			return err
@@ -81,6 +82,17 @@ func setupGatewayIndexer(mgr ctrl.Manager) error {
 	); err != nil {
 		return err
 	}
+
+	if err := mgr.GetFieldIndexer().IndexField(
+		context.Background(),
+		&gatewayv1.Gateway{},
+		GatewayClassIndexRef,
+		func(obj client.Object) (requests []string) {
+			return []string{string(obj.(*gatewayv1.Gateway).Spec.GatewayClassName)}
+		},
+	); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -273,10 +285,10 @@ func setupGatewaySecretIndex(mgr ctrl.Manager) error {
 func setupGatewayClassIndexer(mgr ctrl.Manager) error {
 	return mgr.GetFieldIndexer().IndexField(
 		context.Background(),
-		&gatewayv1.Gateway{},
-		GatewayClassIndexRef,
-		func(obj client.Object) (requests []string) {
-			return []string{string(obj.(*gatewayv1.Gateway).Spec.GatewayClassName)}
+		&gatewayv1.GatewayClass{},
+		ControllerName,
+		func(obj client.Object) []string {
+			return []string{string(obj.(*gatewayv1.GatewayClass).Spec.ControllerName)}
 		},
 	)
 }

internal/controller/ingress_controller.go

@@ -45,6 +45,7 @@ import (
 	"github.com/apache/apisix-ingress-controller/api/v1alpha1"
 	"github.com/apache/apisix-ingress-controller/internal/controller/indexer"
 	"github.com/apache/apisix-ingress-controller/internal/controller/status"
+	"github.com/apache/apisix-ingress-controller/internal/manager/readiness"
 	"github.com/apache/apisix-ingress-controller/internal/provider"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 )
@@ -59,6 +60,7 @@ type IngressReconciler struct { //nolint:revive
 	genericEvent chan event.GenericEvent
 
 	Updater status.Updater
+	Readier readiness.ReadinessManager
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -117,6 +119,7 @@ func (r *IngressReconciler) SetupWithManager(mgr ctrl.Manager) error {
 
 // Reconcile handles the reconciliation of Ingress resources
 func (r *IngressReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	defer r.Readier.Done(&networkingv1.Ingress{}, req.NamespacedName)
 	ingress := new(networkingv1.Ingress)
 	if err := r.Get(ctx, req.NamespacedName, ingress); err != nil {
 		if client.IgnoreNotFound(err) == nil {

internal/controller/status/updater.go

@@ -150,6 +150,10 @@ func (u *UpdateHandler) Start(ctx context.Context) error {
 	}
 }
 
+func (u *UpdateHandler) NeedsLeaderElection() bool {
+	return true
+}
+
 func (u *UpdateHandler) Writer() Updater {
 	return &UpdateWriter{
 		updateChannel: u.updateChannel,

internal/controller/utils.go

@@ -35,6 +35,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1 "k8s.io/api/discovery/v1"
 	networkingv1 "k8s.io/api/networking/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -1448,3 +1449,37 @@ func TypePredicate[T client.Object]() func(obj client.Object) bool {
 		return ok
 	}
 }
+
+func MatchConsumerGatewayRef(ctx context.Context, c client.Client, log logr.Logger, consumer *v1alpha1.Consumer) bool {
+	if consumer.Spec.GatewayRef.Name == "" {
+		return false
+	}
+	if consumer.Spec.GatewayRef.Kind != nil && *consumer.Spec.GatewayRef.Kind != KindGateway {
+		return false
+	}
+	if consumer.Spec.GatewayRef.Group != nil && *consumer.Spec.GatewayRef.Group != gatewayv1.GroupName {
+		return false
+	}
+	ns := consumer.GetNamespace()
+	if consumer.Spec.GatewayRef.Namespace != nil {
+		ns = *consumer.Spec.GatewayRef.Namespace
+	}
+	gateway := &gatewayv1.Gateway{}
+	if err := c.Get(context.Background(), client.ObjectKey{
+		Name:      consumer.Spec.GatewayRef.Name,
+		Namespace: ns,
+	}, gateway); err != nil {
+		if !k8serrors.IsNotFound(err) {
+			log.Error(err, "failed to get gateway", "gateway", consumer.Spec.GatewayRef.Name)
+		}
+		return false
+	}
+	gatewayClass := &gatewayv1.GatewayClass{}
+	if err := c.Get(context.Background(), client.ObjectKey{Name: string(gateway.Spec.GatewayClassName)}, gatewayClass); err != nil {
+		if !k8serrors.IsNotFound(err) {
+			log.Error(err, "failed to get gateway class", "gateway", gateway.GetName(), "gatewayclass", gateway.Spec.GatewayClassName)
+		}
+		return false
+	}
+	return matchesController(string(gatewayClass.Spec.ControllerName))
+}