Merge pull request #46 from 7ing/fix-retry-logic

Add timeout to renewal issuance logic

Author: jetstack-bot

manager/manager.go

@@ -189,6 +189,7 @@ func NewManager(opts Options) (*Manager, error) {
 		maxRequestsPerVolume: opts.MaxRequestsPerVolume,
 		nodeNameHash:         nodeNameHash,
 		backoffConfig:        *opts.RenewalBackoffConfig,
+		issueRenewalTimeout:  time.Second * 60, // issueRenewalTimeout set to align with NodePublishVolume timeout value
 		requestNameGenerator: func() string {
 			return string(uuid.NewUUID())
 		},
@@ -291,6 +292,9 @@ type Manager struct {
 	// backoffConfig configures the exponential backoff applied to certificate renewal failures.
 	backoffConfig wait.Backoff
 
+	// issueRenewalTimeout defines timeout value for each issue() call in renewal process
+	issueRenewalTimeout time.Duration
+
 	// requestNameGenerator generates a new random name for a certificaterequest object
 	// Defaults to uuid.NewUUID() from k8s.io/apimachinery/pkg/util/uuid.
 	requestNameGenerator func() string
@@ -634,7 +638,9 @@ func (m *Manager) startRenewalRoutine(volumeID string) (started bool) {
 					// we'll immediately stop waiting and 'continue' which will then hit the `case <-stopCh` case in the `select`.
 					if err := wait.ExponentialBackoffWithContext(ctx, m.backoffConfig, func(ctx context.Context) (bool, error) {
 						log.Info("Triggering new issuance")
-						if err := m.issue(ctx, volumeID); err != nil {
+						issueCtx, issueCancel := context.WithTimeout(ctx, m.issueRenewalTimeout)
+						defer issueCancel()
+						if err := m.issue(issueCtx, volumeID); err != nil {
 							log.Error(err, "Failed to issue certificate, retrying after applying exponential backoff")
 							return false, nil
 						}

manager/manager_test.go

@@ -8,6 +8,7 @@ import (
 	"encoding/pem"
 	"fmt"
 	"math/big"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -285,6 +286,71 @@ func TestManager_ManageVolume_beginsManagingAndProceedsIfNotReady(t *testing.T)
 	}
 }
 
+func TestManager_ManageVolume_exponentialBackOffRetryOnIssueErrors(t *testing.T) {
+	expBackOffDuration := 100 * time.Millisecond
+	expBackOffCap := 5 * expBackOffDuration
+	expBackOffFactor := 2.0 // We multiply the 'duration' by 2.0 if the attempt fails/errors
+	expBackOffJitter := 0.0 // No jitter to the 'duration', so we could calculate number of retries easily
+	expBackOffSteps := 100  // The maximum number of backoff attempts
+	issueRenewalTimeout := expBackOffDuration
+
+	// Expected number of retries in each expBackOff cycle :=
+	// 				⌈log base expBackOffFactor of (expBackOffCap/expBackOffDuration)⌉
+	var expectNumOfRetries int32 = 3 // ⌈log2(500/100)⌉
+
+	// Because in startRenewalRoutine, ticker := time.NewTicker(time.Second)
+	// 2 seconds should complete an expBackOff cycle
+	// ticker start time (1s) + expBackOffCap (0.5s) + expectNumOfRetries (3) * issueRenewalTimeout (0.1)
+	expectGlobalTimeout := 2 * time.Second
+
+	var numOfRetries int32 = 0 // init
+
+	opts := newDefaultTestOptions(t)
+	opts.RenewalBackoffConfig = &wait.Backoff{
+		Duration: expBackOffDuration,
+		Cap:      expBackOffCap,
+		Factor:   expBackOffFactor,
+		Jitter:   expBackOffJitter,
+		Steps:    expBackOffSteps,
+	}
+	opts.ReadyToRequest = func(meta metadata.Metadata) (bool, string) {
+		// ReadyToRequest will be called by issue()
+		atomic.AddInt32(&numOfRetries, 1) // run in a goroutine, thus increment it atomically
+		return true, ""                   // AlwaysReadyToRequest
+	}
+	m, err := NewManager(opts)
+	m.issueRenewalTimeout = issueRenewalTimeout
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Register a new volume with the metadata store
+	store := opts.MetadataReader.(storage.Interface)
+	meta := metadata.Metadata{
+		VolumeID:   "vol-id",
+		TargetPath: "/fake/path",
+	}
+	store.RegisterMetadata(meta)
+	// Ensure we stop managing the volume after the test
+	defer func() {
+		store.RemoveVolume(meta.VolumeID)
+		m.UnmanageVolume(meta.VolumeID)
+	}()
+
+	// Put the certificate under management
+	managed := m.ManageVolume(meta.VolumeID)
+	if !managed {
+		t.Errorf("expected management to have started, but it did not")
+	}
+
+	time.Sleep(expectGlobalTimeout)
+
+	actualNumOfRetries := atomic.LoadInt32(&numOfRetries) // read atomically
+	if actualNumOfRetries != expectNumOfRetries {
+		t.Errorf("expect %d of retires, but got %d", expectNumOfRetries, actualNumOfRetries)
+	}
+}
+
 func TestManager_cleanupStaleRequests(t *testing.T) {
 	type fields struct {
 		nodeID               string