Add configurable exponential backoff for certificate apply retries (#137)

* add per key exponential backoff for retries


Signed-off-by: Hiroki Hanada <hiroki-hanada@cybozu.co.jp>

Author: hanapedia

cmd/root.go

@@ -50,6 +50,8 @@ func init() {
 	fs.StringSlice("allowed-dns-namespaces", []string{}, "List of namespaces where DNSEndpoint resources can be created. If empty, no namespaces are allowed")
 	fs.StringSlice("allowed-issuer-namespaces", []string{}, "List of namespaces where Certificate resources can be created. If empty, no namespaces are allowed")
 	fs.Float64("certificate-apply-limit", 0, "Maximum number of certificate apply operations allowed per second (0 disables rate limiting)")
+	fs.Duration("certificate-apply-retry-base-delay", controllers.DefaultRetryBaseDelay, "Base delay for certificate apply exponential backoff retry")
+	fs.Duration("certificate-apply-retry-max-delay", controllers.DefaultRetryMaxDelay, "Maximum delay for certificate apply exponential backoff retry")
 	if err := viper.BindPFlags(fs); err != nil {
 		panic(err)
 	}

cmd/run.go

@@ -83,6 +83,18 @@ func run() error {
 		return errors.New("certificate-apply-limit must be greater than or equal to 0")
 	}
 
+	opts.CertificateApplyRetryBaseDelay = viper.GetDuration("certificate-apply-retry-base-delay")
+	if opts.CertificateApplyRetryBaseDelay <= 0 {
+		return errors.New("certificate-apply-retry-base-delay must be greater than 0")
+	}
+	opts.CertificateApplyRetryMaxDelay = viper.GetDuration("certificate-apply-retry-max-delay")
+	if opts.CertificateApplyRetryMaxDelay <= 0 {
+		return errors.New("certificate-apply-retry-max-delay must be greater than 0")
+	}
+	if opts.CertificateApplyRetryMaxDelay < opts.CertificateApplyRetryBaseDelay {
+		return errors.New("certificate-apply-retry-max-delay must be greater than or equal to certificate-apply-retry-base-delay")
+	}
+
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme: scheme,
 		Metrics: metricsserver.Options{

controllers/certificate_worker.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math"
 	"sync"
+	"time"
 
 	cmv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	projectcontourv1 "github.com/projectcontour/contour/apis/projectcontour/v1"
@@ -34,6 +35,9 @@ const (
 	labelApplyResult                        = "result"
 	applyResultSuccess     applyResultValue = "success"
 	applyResultError       applyResultValue = "error"
+
+	DefaultRetryBaseDelay = 5 * time.Second
+	DefaultRetryMaxDelay  = 10 * time.Minute
 )
 
 type Applier[T client.Object] interface {
@@ -104,8 +108,9 @@ func NewCertificateApplyWorker(client client.Client, opt ReconcilerOptions) *Cer
 	}
 
 	global := &workqueue.TypedBucketRateLimiter[types.NamespacedName]{Limiter: limiter}
-	workqueue := workqueue.NewTypedRateLimitingQueueWithConfig(
-		global,
+	expFailure := workqueue.NewTypedItemExponentialFailureRateLimiter[types.NamespacedName](opt.CertificateApplyRetryBaseDelay, opt.CertificateApplyRetryMaxDelay)
+	certQueue := workqueue.NewTypedRateLimitingQueueWithConfig(
+		workqueue.NewTypedMaxOfRateLimiter(global, expFailure),
 		workqueue.TypedRateLimitingQueueConfig[types.NamespacedName]{
 			Name: certificateApplierName,
 		},
@@ -114,7 +119,7 @@ func NewCertificateApplyWorker(client client.Client, opt ReconcilerOptions) *Cer
 	return &CertificateApplyWorker{
 		ReconcilerOptions: opt,
 		client:            client,
-		workqueue:         workqueue,
+		workqueue:         certQueue,
 		manifests:         make(map[types.NamespacedName]*cmv1.Certificate),
 		limiter:           limiter,
 		retryCh:           retryCh,
@@ -182,7 +187,6 @@ func (w *CertificateApplyWorker) Start(ctx context.Context) error {
 		log.Info("processing cert queue item", "key", objKey.String())
 
 		func() {
-			// there is no need to call .Forget since we are only using BucketRateLimiter
 			defer w.workqueue.Done(objKey)
 
 			w.mu.Lock()
@@ -209,6 +213,9 @@ func (w *CertificateApplyWorker) Start(ctx context.Context) error {
 
 			log.Info("cert applied from queue", "key", objKey.String())
 			w.recordApply(viaQueueYes, applyResultSuccess)
+			// Forget the cert key to reset the exponential backoff counter so that
+			// a future failure starts from the base delay again.
+			w.workqueue.Forget(objKey)
 		}()
 	}
 }
@@ -250,35 +257,33 @@ func (w *CertificateApplyWorker) enqueueCertificate(objKey types.NamespacedName,
 	w.workqueue.AddRateLimited(objKey)
 }
 
-// enqueueHTTPProxy enqueues HTTPProxy in to the workqueue used by the main Reconcile function.
-// It requires the Controller to be setup with .WatchesRawSource using the same channel as w.retryCh
+// enqueueHTTPProxy sends the HTTPProxy that owns obj directly to retryCh so the main
+// reconcile loop picks it up. The exponential backoff is handled by the certQueue's
+// MaxOfRateLimiter when the reconcile loop calls AddRateLimited on subsequent failures.
 func (w *CertificateApplyWorker) enqueueHTTPProxy(ctx context.Context, obj *cmv1.Certificate) {
 	log := crlog.FromContext(ctx)
-	if w.retryCh == nil {
-		return
-	}
 	annotations := obj.GetAnnotations()
 	if annotations == nil {
-		log.Error(fmt.Errorf("annotation does not exist"), "skipping HTTPProxy enqueue", "certificateName", obj.Name, "certificateNamespace", obj.Namespace)
+		log.Error(fmt.Errorf("annotations do not exist on certificate %s/%s", obj.Namespace, obj.Name), "skipping HTTPProxy enqueue", "certificateName", obj.Name, "certificateNamespace", obj.Namespace)
 		return
 	}
 	owner, ok := annotations[ownerAnnotation]
 	if !ok {
-		log.Error(fmt.Errorf("annotation not found for %s", ownerAnnotation), "skipping HTTPProxy enqueue", "certificateName", obj.Name, "certificateNamespace", obj.Namespace)
+		log.Error(fmt.Errorf("annotation %q not found on certificate %s/%s", ownerAnnotation, obj.Namespace, obj.Name), "skipping HTTPProxy enqueue", "certificateName", obj.Name, "certificateNamespace", obj.Namespace)
 		return
 	}
-
 	ns, name, err := cache.SplitMetaNamespaceKey(owner)
 	if err != nil {
 		log.Error(err, "skipping HTTPProxy enqueue", "certificateName", obj.Name, "certificateNamespace", obj.Namespace)
 		return
 	}
-	h := projectcontourv1.HTTPProxy{}
+	h := &projectcontourv1.HTTPProxy{}
 	h.SetNamespace(ns)
 	h.SetName(name)
-	log.Info("re-queueing HTTPProxy", "namespace", ns, "name", name)
-	w.retryCh <- event.TypedGenericEvent[*projectcontourv1.HTTPProxy]{
-		Object: &h,
+	log.Info("re-queueing HTTPProxy for retry", "namespace", ns, "name", name)
+	select {
+	case w.retryCh <- event.TypedGenericEvent[*projectcontourv1.HTTPProxy]{Object: h}:
+	case <-ctx.Done():
 	}
 }
 

controllers/certificate_worker_test.go

@@ -93,7 +93,9 @@ func testCertificateApplyWorker() {
 			baseClient := newFakeClient() // no existing Certificate
 			cl := &applyAsUpdateClient{Client: baseClient}
 			worker := NewCertificateApplyWorker(cl, ReconcilerOptions{
-				CertificateApplyLimit: 10, // avoid rate.Limit(0) semantics
+				CertificateApplyLimit:          10, // avoid rate.Limit(0) semantics
+				CertificateApplyRetryBaseDelay: 1 * time.Millisecond,
+				CertificateApplyRetryMaxDelay:  10 * time.Millisecond,
 			})
 
 			reg := prometheus.NewRegistry()
@@ -120,7 +122,7 @@ func testCertificateApplyWorker() {
 			Expect(k8serrors.IsNotFound(err)).To(BeTrue(), "new object should not be applied immediately, but queued")
 
 			// And the worker should have the key in its internal queue
-			Expect(worker.workqueue.Len()).To(Equal(1))
+			Eventually(func() int { return worker.workqueue.Len() }, 500*time.Millisecond, time.Millisecond).Should(Equal(1))
 
 			queuedKey, shutdown := worker.workqueue.Get()
 			Expect(shutdown).To(BeFalse())
@@ -207,7 +209,9 @@ func testCertificateApplyWorker() {
 			baseClient := newFakeClient(current) // init with a certificate
 			cl := &applyAsUpdateClient{Client: baseClient}
 			worker := NewCertificateApplyWorker(cl, ReconcilerOptions{
-				CertificateApplyLimit: 10,
+				CertificateApplyLimit:          10,
+				CertificateApplyRetryBaseDelay: 1 * time.Millisecond,
+				CertificateApplyRetryMaxDelay:  10 * time.Millisecond,
 			})
 
 			reg := prometheus.NewRegistry()
@@ -221,7 +225,7 @@ func testCertificateApplyWorker() {
 			Expect(worker.Apply(ctx, desired)).To(Succeed())
 
 			// Should be queued, not applied directly
-			Expect(worker.workqueue.Len()).To(Equal(1))
+			Eventually(func() int { return worker.workqueue.Len() }, 500*time.Millisecond, time.Millisecond).Should(Equal(1))
 
 			// The object in the fake client should still be the old spec (no extra DNS yet),
 			// because the queued apply hasn’t run Start() / processed the queue.
@@ -250,7 +254,9 @@ func testCertificateApplyWorker() {
 			baseClient := newFakeClient() // no existing certificate
 			cl := &applyAsUpdateClient{Client: baseClient}
 			worker := NewCertificateApplyWorker(cl, ReconcilerOptions{
-				CertificateApplyLimit: 10, // avoid rate.Limit(0) oddness
+				CertificateApplyLimit:          10, // avoid rate.Limit(0) oddness
+				CertificateApplyRetryBaseDelay: 1 * time.Millisecond,
+				CertificateApplyRetryMaxDelay:  10 * time.Millisecond,
 			})
 
 			reg := prometheus.NewRegistry()
@@ -318,7 +324,9 @@ func testCertificateApplyWorker() {
 			cl := &failingPatchClient{Client: baseClient}
 
 			worker := NewCertificateApplyWorker(cl, ReconcilerOptions{
-				CertificateApplyLimit: 10,
+				CertificateApplyLimit:          10,
+				CertificateApplyRetryBaseDelay: 100 * time.Millisecond,
+				CertificateApplyRetryMaxDelay:  1 * time.Second,
 			})
 
 			reg := prometheus.NewRegistry()
@@ -350,11 +358,12 @@ func testCertificateApplyWorker() {
 			// Apply should enqueue the certificate (RequiresQueue returns true for new object)
 			Expect(worker.Apply(ctx, cert)).To(Succeed())
 
-			// We expect a retry event on the channel because Patch always fails
+			// We expect a retry event on the channel because Patch always fails.
+			// The event is delayed by CertificateApplyRetryBaseDelay (100ms) due to the exponential backoff queue.
 			retryCh := worker.GetRetryChannel()
 
 			var evt event.TypedGenericEvent[*projectcontourv1.HTTPProxy]
-			Eventually(retryCh, 2*time.Second, 100*time.Millisecond).Should(
+			Eventually(retryCh, 5*time.Second, 100*time.Millisecond).Should(
 				Receive(&evt),
 				"expected an HTTPProxy event to be sent on retry channel after apply failure",
 			)

controllers/httpproxy_controller_test.go

@@ -1391,14 +1391,16 @@ func testHTTPProxyReconcile() {
 
 		prefix := "test-"
 		opts := ReconcilerOptions{
-			ServiceKey:            testServiceKey,
-			Prefix:                prefix,
-			DefaultIssuerName:     "test-issuer",
-			DefaultIssuerKind:     IssuerKind,
-			CreateDNSEndpoint:     true,
-			CreateCertificate:     true,
-			PropagatedAnnotations: []string{"foo"}, // must propagate an annotation to update cert metadata
-			CertificateApplyLimit: 1,
+			ServiceKey:                     testServiceKey,
+			Prefix:                         prefix,
+			DefaultIssuerName:              "test-issuer",
+			DefaultIssuerKind:              IssuerKind,
+			CreateDNSEndpoint:              true,
+			CreateCertificate:              true,
+			PropagatedAnnotations:          []string{"foo"}, // must propagate an annotation to update cert metadata
+			CertificateApplyLimit:          1,
+			CertificateApplyRetryBaseDelay: 1 * time.Millisecond,
+			CertificateApplyRetryMaxDelay:  10 * time.Millisecond,
 		}
 
 		err := SetupReconciler(mgr, scm, opts)

controllers/setup.go

@@ -1,6 +1,8 @@
 package controllers
 
 import (
+	"time"
+
 	cmapiv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
 	projectcontourv1 "github.com/projectcontour/contour/apis/projectcontour/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -14,22 +16,24 @@ import (
 
 // ReconcilerOptions is a set of options for reconcilers
 type ReconcilerOptions struct {
-	ServiceKey              client.ObjectKey
-	Prefix                  string
-	DefaultIssuerName       string
-	DefaultIssuerKind       string
-	DefaultDelegatedDomain  string
-	AllowedDelegatedDomains []string
-	AllowCustomDelegations  bool
-	CSRRevisionLimit        uint
-	CreateDNSEndpoint       bool
-	CreateCertificate       bool
-	IngressClassName        string
-	PropagatedAnnotations   []string
-	PropagatedLabels        []string
-	AllowedDNSNamespaces    []string
-	AllowedIssuerNamespaces []string
-	CertificateApplyLimit   float64
+	ServiceKey                     client.ObjectKey
+	Prefix                         string
+	DefaultIssuerName              string
+	DefaultIssuerKind              string
+	DefaultDelegatedDomain         string
+	AllowedDelegatedDomains        []string
+	AllowCustomDelegations         bool
+	CSRRevisionLimit               uint
+	CreateDNSEndpoint              bool
+	CreateCertificate              bool
+	IngressClassName               string
+	PropagatedAnnotations          []string
+	PropagatedLabels               []string
+	AllowedDNSNamespaces           []string
+	AllowedIssuerNamespaces        []string
+	CertificateApplyLimit          float64
+	CertificateApplyRetryBaseDelay time.Duration
+	CertificateApplyRetryMaxDelay  time.Duration
 }
 
 // SetupScheme initializes a schema