feat: support retry in case of sync failure (#2534)

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

Author: AlinsRan

internal/provider/apisix/provider.go

@@ -38,12 +38,20 @@ 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 ProviderTypeAPISIX = "apisix"
+const (
+	ProviderTypeAPISIX = "apisix"
+
+	RetryBaseDelay = 1 * time.Second
+	RetryMaxDelay  = 1000 * time.Second
+
+	MinSyncPeriod = 1 * time.Second
+)
 
 type apisixProvider struct {
 	provider.Options
@@ -229,33 +237,32 @@ func (d *apisixProvider) Start(ctx context.Context) error {
 
 	initalSyncDelay := d.InitSyncDelay
 	if initalSyncDelay > 0 {
-		time.AfterFunc(initalSyncDelay, func() {
-			if err := d.sync(ctx); err != nil {
-				log.Error(err)
-				return
-			}
-		})
+		time.AfterFunc(initalSyncDelay, d.syncNotify)
 	}
 
-	if d.SyncPeriod < 1 {
-		return nil
+	syncPeriod := d.SyncPeriod
+	if syncPeriod < MinSyncPeriod {
+		syncPeriod = MinSyncPeriod
 	}
-	ticker := time.NewTicker(d.SyncPeriod)
+	ticker := time.NewTicker(syncPeriod)
 	defer ticker.Stop()
+
+	retrier := common.NewRetrier(common.NewExponentialBackoff(RetryBaseDelay, RetryMaxDelay))
+
 	for {
-		synced := false
 		select {
 		case <-d.syncCh:
-			synced = true
 		case <-ticker.C:
-			synced = true
+		case <-retrier.C():
 		case <-ctx.Done():
+			retrier.Reset()
 			return nil
 		}
-		if synced {
-			if err := d.sync(ctx); err != nil {
-				log.Error(err)
-			}
+		if err := d.sync(ctx); err != nil {
+			log.Error(err)
+			retrier.Next()
+		} else {
+			retrier.Reset()
 		}
 	}
 }

internal/provider/common/retrier.go

@@ -0,0 +1,96 @@
+// 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) Reset() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.timer != nil {
+		r.timer.Stop()
+		r.timer = nil
+	}
+	r.backoff.Reset()
+}
+
+func (r *Retrier) Next() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.timer != nil {
+		r.timer.Stop()
+		r.timer = nil
+	}
+
+	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

@@ -25,6 +25,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 
@@ -1762,4 +1763,62 @@ 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() {
+			if os.Getenv("PROVIDER_TYPE") == framework.ProviderTypeAPI7EE {
+				Skip("skipping test in API7EE mode")
+			}
+			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),
+			})
+		})
+	})
 })