feat: support retry in case of sync failure

Author: AlinsRan

internal/provider/apisix/provider.go

@@ -37,6 +37,7 @@ 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"
 )
@@ -236,19 +237,29 @@ func (d *apisixProvider) Start(ctx context.Context) error {
 	}
 	ticker := time.NewTicker(d.SyncPeriod)
 	defer ticker.Stop()
+
+	retrier := common.NewRetrier(common.NewExponentialBackoff(1*time.Second, 1000*time.Second))
 	for {
 		synced := false
 		select {
 		case <-d.syncCh:
+			retrier.Cancel()
 			synced = true
 		case <-ticker.C:
+			retrier.Cancel()
+			synced = true
+		case <-retrier.C():
 			synced = true
 		case <-ctx.Done():
+			retrier.Cancel()
 			return nil
 		}
 		if synced {
 			if err := d.sync(ctx); err != nil {
 				log.Error(err)
+				retrier.Trigger()
+			} else {
+				retrier.Cancel()
 			}
 		}
 	}

internal/provider/common/retrier.go

@@ -0,0 +1,91 @@
+// 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 common
+
+import (
+	"sync"
+	"time"
+)
+
+type Backoff interface {
+	Next() time.Duration
+	Reset()
+}
+
+type ExponentialBackoff struct {
+	base, max, current time.Duration
+}
+
+func NewExponentialBackoff(base, max time.Duration) *ExponentialBackoff {
+	return &ExponentialBackoff{base: base, max: max, current: base}
+}
+
+func (b *ExponentialBackoff) Next() time.Duration {
+	delay := b.current
+	b.current *= 2
+	if b.current > b.max {
+		b.current = b.max
+	}
+	return delay
+}
+
+func (b *ExponentialBackoff) Reset() {
+	b.current = b.base
+}
+
+type Retrier struct {
+	mu      sync.Mutex
+	ch      chan struct{}
+	timer   *time.Timer
+	backoff Backoff
+}
+
+func NewRetrier(b Backoff) *Retrier {
+	return &Retrier{
+		ch:      make(chan struct{}, 1),
+		backoff: b,
+	}
+}
+
+func (r *Retrier) Cancel() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.timer != nil {
+		r.timer.Stop()
+		r.timer = nil
+	}
+	r.backoff.Reset()
+}
+
+func (r *Retrier) Trigger() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	delay := r.backoff.Next()
+	r.timer = time.AfterFunc(delay, func() {
+		select {
+		case r.ch <- struct{}{}:
+		default:
+		}
+	})
+}
+
+func (r *Retrier) C() <-chan struct{} {
+	return r.ch
+}

test/e2e/crds/v2/route.go

@@ -1676,4 +1676,59 @@ spec:
 			})
 		})
 	})
+
+	Context("Exception Test", func() {
+		const apisixRouteSpec = `
+apiVersion: apisix.apache.org/v2
+kind: ApisixRoute
+metadata:
+  name: default
+spec:
+  ingressClassName: %s
+  http:
+  - name: rule0
+    match:
+      hosts:
+      - httpbin
+      paths:
+      - /*
+    backends:
+    - serviceName: httpbin-service-e2e-test
+      servicePort: 80
+`
+		It("try again when sync failed", func() {
+			s.Deployer.ScaleDataplane(0)
+
+			err := s.CreateResourceFromString(fmt.Sprintf(apisixRouteSpec, s.Namespace()))
+			Expect(err).NotTo(HaveOccurred(), "creating ApisixRoute")
+
+			By("check ApisixRoute status")
+			s.RetryAssertion(func() string {
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				return output
+			}).WithTimeout(30 * time.Second).
+				Should(
+					And(
+						ContainSubstring(`status: "False"`),
+						ContainSubstring(`reason: SyncFailed`),
+					),
+				)
+
+			s.Deployer.ScaleDataplane(1)
+
+			s.RetryAssertion(func() string {
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				return output
+			}).WithTimeout(60 * time.Second).
+				Should(ContainSubstring(`status: "True"`))
+
+			By("check route in APISIX")
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method: "GET",
+				Path:   "/get",
+				Host:   "httpbin",
+				Check:  scaffold.WithExpectedStatus(200),
+			})
+		})
+	})
 })

test/e2e/crds/v2/status.go

@@ -24,6 +24,7 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/yaml"
 
@@ -41,7 +42,7 @@ var _ = Describe("Test CRD Status", Label("apisix.apache.org", "v2", "apisixrout
 	Context("Test ApisixRoute Sync Status", func() {
 		BeforeEach(func() {
 			By("create GatewayProxy")
-			gatewayProxy := s.GetGatewayProxyYaml()
+			gatewayProxy := s.GetGatewayProxyWithServiceYaml()
 			err := s.CreateResourceFromString(gatewayProxy)
 			Expect(err).NotTo(HaveOccurred(), "creating GatewayProxy")
 			time.Sleep(5 * time.Second)
@@ -144,7 +145,8 @@ spec:
 
 		It("dataplane unavailable", func() {
 			By("apply ApisixRoute")
-			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, &apiv2.ApisixRoute{}, fmt.Sprintf(ar, s.Namespace(), s.Namespace()))
+			arYaml := fmt.Sprintf(ar, s.Namespace(), s.Namespace())
+			applier.MustApplyAPIv2(types.NamespacedName{Namespace: s.Namespace(), Name: "default"}, &apiv2.ApisixRoute{}, arYaml)
 
 			By("check route in APISIX")
 			s.RequestAssert(&scaffold.RequestAssert{
@@ -154,27 +156,51 @@ spec:
 				Check:   scaffold.WithExpectedStatus(200),
 			})
 
-			s.Deployer.ScaleDataplane(0)
+			By("get yaml from service")
+			serviceYaml, err := s.GetOutputFromString("svc", framework.ProviderType, "-o", "yaml")
+			Expect(err).NotTo(HaveOccurred(), "getting service yaml")
+			By("update service to type ExternalName with invalid host")
+			var k8sservice corev1.Service
+			err = yaml.Unmarshal([]byte(serviceYaml), &k8sservice)
+			Expect(err).NotTo(HaveOccurred(), "unmarshalling service")
+			oldSpec := k8sservice.Spec
+			k8sservice.Spec = corev1.ServiceSpec{
+				Type:         corev1.ServiceTypeExternalName,
+				ExternalName: "invalid.host",
+			}
+			newServiceYaml, err := yaml.Marshal(k8sservice)
+			Expect(err).NotTo(HaveOccurred(), "marshalling service")
+			err = s.CreateResourceFromString(string(newServiceYaml))
+			Expect(err).NotTo(HaveOccurred(), "creating service")
 
 			By("check ApisixRoute status")
 			s.RetryAssertion(func() string {
-				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml")
 				return output
-			}).WithTimeout(80 * time.Second).
+			}).WithTimeout(60 * time.Second).
 				Should(
 					And(
 						ContainSubstring(`status: "False"`),
 						ContainSubstring(`reason: SyncFailed`),
 					),
 				)
 
-			s.Deployer.ScaleDataplane(1)
+			By("update service to original spec")
+			serviceYaml, err = s.GetOutputFromString("svc", framework.ProviderType, "-o", "yaml")
+			Expect(err).NotTo(HaveOccurred(), "getting service yaml")
+			err = yaml.Unmarshal([]byte(serviceYaml), &k8sservice)
+			Expect(err).NotTo(HaveOccurred(), "unmarshalling service")
+			k8sservice.Spec = oldSpec
+			newServiceYaml, err = yaml.Marshal(k8sservice)
+			Expect(err).NotTo(HaveOccurred(), "marshalling service")
+			err = s.CreateResourceFromString(string(newServiceYaml))
+			Expect(err).NotTo(HaveOccurred(), "creating service")
 
 			By("check ApisixRoute status after scaling up")
 			s.RetryAssertion(func() string {
-				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml", "-n", s.Namespace())
+				output, _ := s.GetOutputFromString("ar", "default", "-o", "yaml")
 				return output
-			}).WithTimeout(80 * time.Second).
+			}).WithTimeout(60 * time.Second).
 				Should(
 					And(
 						ContainSubstring(`status: "True"`),

test/e2e/framework/manifests/ingress.yaml

@@ -333,7 +333,7 @@ data:
                                         # The period between two consecutive syncs.
                                         # The default value is 0 seconds, which means the controller will not sync.
                                         # If you want to enable the sync, set it to a positive value.
-      init_sync_delay: {{ .InitSyncDelay | default "1m" }}
+      init_sync_delay: {{ .InitSyncDelay | default "20m" }}
 ---
 apiVersion: v1
 kind: Service

test/e2e/scaffold/scaffold.go

@@ -460,6 +460,25 @@ spec:
           value: "%s"
 `
 
+const gatewayProxyWithServiceYaml = `
+apiVersion: apisix.apache.org/v1alpha1
+kind: GatewayProxy
+metadata:
+  name: %s
+  namespace: %s
+spec:
+  provider:
+    type: ControlPlane
+    controlPlane:
+      service:
+        name: %s
+        port: 9180
+      auth:
+        type: AdminKey
+        adminKey:
+          value: "%s"
+`
+
 const ingressClassYaml = `
 apiVersion: networking.k8s.io/v1
 kind: IngressClass
@@ -479,6 +498,10 @@ func (s *Scaffold) GetGatewayProxyYaml() string {
 	return fmt.Sprintf(gatewayProxyYaml, s.namespace, s.namespace, s.Deployer.GetAdminEndpoint(), s.AdminKey())
 }
 
+func (s *Scaffold) GetGatewayProxyWithServiceYaml() string {
+	return fmt.Sprintf(gatewayProxyWithServiceYaml, s.namespace, s.namespace, s.dataplaneService.Name, s.AdminKey())
+}
+
 func (s *Scaffold) GetIngressClassYaml() string {
 	return fmt.Sprintf(ingressClassYaml, s.namespace, s.GetControllerName(), s.namespace, s.namespace)
 }