feat: sync once when startup ingress (#233)

Signed-off-by: Ashing Zheng <[email protected]>

Author: ronething

internal/adc/client/client.go

@@ -39,7 +39,7 @@ import (
 )
 
 type Client struct {
-	syncMu sync.Mutex
+	syncMu sync.RWMutex
 	mu     sync.Mutex
 	*cache.Store
 
@@ -97,6 +97,9 @@ func (d *Client) Update(ctx context.Context, args Task) error {
 	}
 	d.mu.Unlock()
 
+	d.syncMu.RLock()
+	defer d.syncMu.RUnlock()
+
 	if len(deleteConfigs) > 0 {
 		err := d.sync(ctx, Task{
 			Name:          args.Name,
@@ -108,6 +111,7 @@ func (d *Client) Update(ctx context.Context, args Task) error {
 			log.Warnw("failed to sync deleted configs", zap.Error(err))
 		}
 	}
+
 	return d.sync(ctx, args)
 }
 
@@ -154,6 +158,9 @@ func (d *Client) Delete(ctx context.Context, args Task) error {
 	}
 	d.mu.Unlock()
 
+	d.syncMu.RLock()
+	defer d.syncMu.RUnlock()
+
 	return d.sync(ctx, Task{
 		Labels:        args.Labels,
 		ResourceTypes: args.ResourceTypes,

internal/provider/api7ee/provider.go

@@ -19,6 +19,7 @@ package api7ee
 
 import (
 	"context"
+	"sync/atomic"
 	"time"
 
 	"github.com/api7/gopkg/pkg/log"
@@ -56,6 +57,8 @@ type api7eeProvider struct {
 
 	syncCh chan struct{}
 
+	startUpSync atomic.Bool
+
 	client *adcclient.Client
 }
 
@@ -159,6 +162,11 @@ func (d *api7eeProvider) Update(ctx context.Context, tctx *provider.TranslateCon
 		},
 	}
 
+	if !d.startUpSync.Load() {
+		log.Debugw("startup synchronization not completed, skip sync", zap.Any("object", obj))
+		return d.client.UpdateConfig(ctx, task)
+	}
+
 	return d.client.Update(ctx, task)
 }
 
@@ -204,6 +212,12 @@ func (d *api7eeProvider) Delete(ctx context.Context, obj client.Object) error {
 func (d *api7eeProvider) Start(ctx context.Context) error {
 	d.readier.WaitReady(ctx, 5*time.Minute)
 
+	d.startUpSync.Store(true)
+	log.Info("Performing startup synchronization")
+	if err := d.sync(ctx); err != nil {
+		log.Warnw("failed to sync for startup", zap.Error(err))
+	}
+
 	initalSyncDelay := d.InitSyncDelay
 	if initalSyncDelay > 0 {
 		time.AfterFunc(initalSyncDelay, func() {

test/e2e/crds/v2/route.go

@@ -460,6 +460,86 @@ spec:
 		})
 	})
 
+	Context("Ingress Scale and Route Management", func() {
+		It("should handle ApisixRoute management during ingress scaling", func() {
+			const apisixRouteSpec1 = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: route1
+spec:
+  ingressClassName: apisix
+  http:
+  - name: rule1
+    match:
+      hosts:
+      - httpbin
+      paths:
+      - /get
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+`
+
+			const apisixRouteSpec2 = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: route2
+spec:
+  ingressClassName: apisix
+  http:
+  - name: rule2
+    match:
+      hosts:
+      - httpbin
+      paths:
+      - /anything
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+`
+
+			request := func(path string) int {
+				return s.NewAPISIXClient().GET(path).WithHost("httpbin").Expect().Raw().StatusCode
+			}
+
+			By("apply first ApisixRoute")
+			var apisixRoute1 apiv2.ApisixRoute
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "route1"}, &apisixRoute1, apisixRouteSpec1)
+
+			By("apply second ApisixRoute")
+			var apisixRoute2 apiv2.ApisixRoute
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "route2"}, &apisixRoute2, apisixRouteSpec2)
+
+			By("verify both ApisixRoutes work")
+			Eventually(request).WithArguments("/get").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+			Eventually(request).WithArguments("/anything").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+
+			By("scale ingress to 0")
+			s.Deployer.ScaleIngress(0)
+
+			By("delete first ApisixRoute")
+			err := s.DeleteResource("ApisixRoute", "route1")
+			Expect(err).ShouldNot(HaveOccurred(), "deleting ApisixRoute route1")
+
+			By("verify both routes still accessible during scale-down")
+			Eventually(request).WithArguments("/get").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+			Eventually(request).WithArguments("/anything").WithTimeout(8 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+
+			By("scale ingress back to 1")
+			s.Deployer.ScaleIngress(1)
+
+			By("verify first route is gone and second route still accessible after scale-up")
+			Eventually(request).WithArguments("/get").WithTimeout(30 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusNotFound))
+			Eventually(request).WithArguments("/anything").WithTimeout(30 * time.Second).ProbeEvery(time.Second).Should(Equal(http.StatusOK))
+
+			routes, err := s.DefaultDataplaneResource().Route().List(context.Background())
+			Expect(err).NotTo(HaveOccurred(), "list routes error")
+			Expect(routes).To(HaveLen(1), "routes number not expect")
+		})
+	})
+
 	Context("Test ApisixRoute reference ApisixUpstream", func() {
 		It("Test reference ApisixUpstream", func() {
 			const apisixRouteSpec = `

test/e2e/framework/ingress.go

@@ -49,7 +49,7 @@ type IngressDeployOpts struct {
 	ProviderSyncPeriod time.Duration
 	Namespace          string
 	StatusAddress      string
-	Replicas           int
+	Replicas           *int
 	InitSyncDelay      time.Duration
 }
 

test/e2e/framework/manifests/ingress.yaml

@@ -334,6 +334,15 @@ data:
     log_level: "debug"
     controller_name: {{ .ControllerName | default "apisix.apache.org/apisix-ingress-controller" }}
     leader_election_id: "apisix-ingress-controller-leader"
+    leader_election:
+      lease_duration: 10s                   # 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: 5s                   # 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.
+      disable: false                        # Whether to disable leader election.
     exec_adc_timeout: 5s
     provider:
       type: {{ .ProviderType | default "apisix-standalone" }}

test/e2e/gatewayapi/httproute.go

@@ -609,10 +609,11 @@ spec:
 			})
 
 			s.RequestAssert(&scaffold.RequestAssert{
-				Method: "GET",
-				Path:   "/get",
-				Host:   "httpbin2.example",
-				Check:  scaffold.WithExpectedStatus(http.StatusNotFound),
+				Method:  "GET",
+				Path:    "/get",
+				Host:    "httpbin2.example",
+				Timeout: 1 * time.Minute,
+				Check:   scaffold.WithExpectedStatus(http.StatusNotFound),
 			})
 		})
 	})

test/e2e/scaffold/api7_deployer.go

@@ -207,7 +207,7 @@ func (s *API7Deployer) DeployIngress() {
 		ProviderType:   "api7ee",
 		ControllerName: s.opts.ControllerName,
 		Namespace:      s.namespace,
-		Replicas:       1,
+		Replicas:       ptr.To(1),
 	})
 }
 
@@ -216,7 +216,7 @@ func (s *API7Deployer) ScaleIngress(replicas int) {
 		ProviderType:   "api7ee",
 		ControllerName: s.opts.ControllerName,
 		Namespace:      s.namespace,
-		Replicas:       replicas,
+		Replicas:       ptr.To(replicas),
 	})
 }
 

test/e2e/scaffold/apisix_deployer.go

@@ -265,7 +265,7 @@ func (s *APISIXDeployer) DeployIngress() {
 		ProviderType:       framework.ProviderType,
 		ProviderSyncPeriod: syncPeriod,
 		Namespace:          s.namespace,
-		Replicas:           1,
+		Replicas:           ptr.To(1),
 	})
 }
 
@@ -279,7 +279,7 @@ func (s *APISIXDeployer) ScaleIngress(replicas int) {
 		ProviderType:       framework.ProviderType,
 		ProviderSyncPeriod: syncPeriod,
 		Namespace:          s.namespace,
-		Replicas:           replicas,
+		Replicas:           &replicas,
 	})
 }