feat: support retry for full synchronization (#367)

Author: AlinsRan

internal/provider/api7ee/provider.go

@@ -39,12 +39,18 @@ import (
 	"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/provider/common"
 	"github.com/apache/apisix-ingress-controller/internal/types"
 	"github.com/apache/apisix-ingress-controller/internal/utils"
 	pkgutils "github.com/apache/apisix-ingress-controller/pkg/utils"
 )
 
-const ProviderTypeAPI7EE = "api7ee"
+const (
+	ProviderTypeAPI7EE = "api7ee"
+
+	RetryBaseDelay = 1 * time.Second
+	RetryMaxDelay  = 1000 * time.Second
+)
 
 type api7eeProvider struct {
 	translator *translator.Translator
@@ -252,20 +258,23 @@ func (d *api7eeProvider) Start(ctx context.Context) error {
 	}
 	ticker := time.NewTicker(d.SyncPeriod)
 	defer ticker.Stop()
+
+	retrier := common.NewRetrier(common.NewExponentialBackoff(RetryBaseDelay, RetryMaxDelay))
+
 	for {
-		synced := false
 		select {
 		case <-d.syncCh:
-			synced = true
+		case <-retrier.C():
 		case <-ticker.C:
-			synced = true
 		case <-ctx.Done():
 			return nil
 		}
-		if synced {
-			if err := d.sync(ctx); err != nil {
-				d.log.Error(err, "failed to sync for startup")
-			}
+
+		if err := d.sync(ctx); err != nil {
+			d.log.Error(err, "failed to sync")
+			retrier.Next()
+		} else {
+			retrier.Reset()
 		}
 	}
 }

test/e2e/api7/gatewayproxy.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package gatewayapi
+package api7
 
 import (
 	"fmt"

test/e2e/api7/route.go

@@ -0,0 +1,138 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package api7
+
+import (
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/apache/apisix-ingress-controller/test/e2e/scaffold"
+)
+
+var _ = Describe("Test apisix.apache.org/v2 Status", Label("apisix.apache.org", "v2", "apisixroute"), func() {
+	s := scaffold.NewDefaultScaffold()
+
+	Context("Test ApisixRoute Sync Status", func() {
+		const (
+			serviceSpec = `
+apiVersion: v1
+kind: Service
+metadata:
+  name: api7ee3-dashboard
+spec:
+  type: ExternalName
+  externalName: %s
+`
+			gatewayProxyYaml = `
+apiVersion: apisix.apache.org/v1alpha1
+kind: GatewayProxy
+metadata:
+  name: apisix-proxy-config
+spec:
+  provider:
+    type: ControlPlane
+    controlPlane:
+      endpoints:
+        - https://api7ee3-dashboard:7443
+      auth:
+        type: AdminKey
+        adminKey:
+          value: "%s"
+`
+			ar = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: default
+  namespace: %s
+spec:
+  ingressClassName: %s
+  http:
+  - name: rule0
+    match:
+      hosts:
+      - httpbin
+      paths:
+      - /*
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+`
+		)
+
+		It("dataplane unavailable", func() {
+			s.Deployer.ScaleIngress(0)
+			By("create GatewayProxy")
+			err := s.CreateResourceFromString(fmt.Sprintf(gatewayProxyYaml, s.AdminKey()))
+			Expect(err).NotTo(HaveOccurred(), "creating GatewayProxy")
+
+			By("create IngressClass")
+			err = s.CreateResourceFromStringWithNamespace(s.GetIngressClassYaml(), "")
+			Expect(err).NotTo(HaveOccurred(), "creating IngressClass")
+
+			By("create Service with invalid host")
+			err = s.CreateResourceFromString(fmt.Sprintf(serviceSpec, "invalid.host"))
+			Expect(err).NotTo(HaveOccurred(), "creating Service")
+
+			By("apply ApisixRoute")
+			err = s.CreateResourceFromString(fmt.Sprintf(ar, s.Namespace(), s.Namespace()))
+			Expect(err).NotTo(HaveOccurred(), "creating ApisixRoute")
+
+			s.Deployer.ScaleIngress(1)
+
+			By("check ApisixRoute status")
+			s.RetryAssertion(func() string {
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				return output
+			}).WithTimeout(5 * time.Minute).
+				Should(
+					And(
+						ContainSubstring(`status: "False"`),
+						ContainSubstring(`reason: SyncFailed`),
+					),
+				)
+
+			By("update service to dashboard")
+			err = s.CreateResourceFromString(fmt.Sprintf(serviceSpec, "api7ee3-dashboard.api7-ee-e2e.svc.cluster.local"))
+			Expect(err).NotTo(HaveOccurred(), "updating Service")
+
+			By("check ApisixRoute status after scaling up")
+			s.RetryAssertion(func() string {
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				return output
+			}).WithTimeout(60 * time.Second).
+				Should(
+					And(
+						ContainSubstring(`status: "True"`),
+						ContainSubstring(`reason: Accepted`),
+					),
+				)
+
+			By("check route in APISIX")
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method: "GET",
+				Path:   "/get",
+				Host:   "httpbin",
+				Check:  scaffold.WithExpectedStatus(200),
+			})
+		})
+	})
+})

test/e2e/scaffold/api7_deployer.go

@@ -193,20 +193,22 @@ func (s *API7Deployer) DeployIngress() {
 	}
 
 	s.Framework.DeployIngress(framework.IngressDeployOpts{
-		ProviderType:   "api7ee",
-		ControllerName: s.runtimeOpts.ControllerName,
-		Namespace:      s.namespace,
-		Replicas:       ptr.To(1),
-		WebhookEnable:  s.runtimeOpts.EnableWebhook,
+		ProviderType:       "api7ee",
+		ControllerName:     s.runtimeOpts.ControllerName,
+		ProviderSyncPeriod: 1 * time.Hour,
+		Namespace:          s.namespace,
+		Replicas:           ptr.To(1),
+		WebhookEnable:      s.runtimeOpts.EnableWebhook,
 	})
 }
 
 func (s *API7Deployer) ScaleIngress(replicas int) {
 	s.Framework.DeployIngress(framework.IngressDeployOpts{
-		ProviderType:   "api7ee",
-		ControllerName: s.runtimeOpts.ControllerName,
-		Namespace:      s.namespace,
-		Replicas:       ptr.To(replicas),
+		ProviderType:       "api7ee",
+		ControllerName:     s.runtimeOpts.ControllerName,
+		ProviderSyncPeriod: 1 * time.Hour,
+		Namespace:          s.namespace,
+		Replicas:           ptr.To(replicas),
 	})
 }