Merge pull request #499 from freehan/backoff

Add a Backofff Handler utils

pkg/neg/syncers/backoff.go

@@ -0,0 +1,81 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed 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 syncers
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+var ErrRetriesExceeded = fmt.Errorf("maximum retry exceeded")
+
+// backoffHandler handles delays for back off retry
+type backoffHandler interface {
+	// NextRetryDelay returns the delay for next retry or error if maximum number of retries exceeded.
+	NextRetryDelay() (time.Duration, error)
+	// ResetRetryDelay resets the retry delay
+	ResetRetryDelay()
+}
+
+// exponentialBackOffHandler is a backoff handler that returns retry delays semi-exponentially with random jitter within boundary.
+// exponentialBackOffHandler returns ErrRetriesExceeded when maximum number of retries has reached.
+type exponentialBackOffHandler struct {
+	lock           sync.Mutex
+	lastRetryDelay time.Duration
+	retryCount     int
+	maxRetries     int
+	minRetryDelay  time.Duration
+	maxRetryDelay  time.Duration
+}
+
+func NewExponentialBackendOffHandler(maxRetries int, minRetryDelay, maxRetryDelay time.Duration) *exponentialBackOffHandler {
+	return &exponentialBackOffHandler{
+		lastRetryDelay: time.Duration(0),
+		retryCount:     0,
+		maxRetries:     maxRetries,
+		minRetryDelay:  minRetryDelay,
+		maxRetryDelay:  maxRetryDelay,
+	}
+}
+
+// NextRetryDelay returns the next back off delay for retry.
+func (handler *exponentialBackOffHandler) NextRetryDelay() (time.Duration, error) {
+	handler.lock.Lock()
+	defer handler.lock.Unlock()
+	handler.retryCount += 1
+	if handler.maxRetries > 0 && handler.retryCount > handler.maxRetries {
+		return 0, ErrRetriesExceeded
+	}
+	handler.lastRetryDelay = wait.Jitter(handler.lastRetryDelay, 1.0)
+	if handler.lastRetryDelay < handler.minRetryDelay {
+		handler.lastRetryDelay = handler.minRetryDelay
+	} else if handler.lastRetryDelay > handler.maxRetryDelay {
+		handler.lastRetryDelay = handler.maxRetryDelay
+	}
+	return handler.lastRetryDelay, nil
+}
+
+// ResetRetryDelay resets the retry delay.
+func (handler *exponentialBackOffHandler) ResetRetryDelay() {
+	handler.lock.Lock()
+	defer handler.lock.Unlock()
+	handler.retryCount = 0
+	handler.lastRetryDelay = time.Duration(0)
+}

pkg/neg/syncers/backoff_test.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed 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 syncers
+
+import (
+	"testing"
+	"time"
+)
+
+const (
+	testRetry         = 15
+	testMinRetryDelay = 5 * time.Second
+	testMaxRetryDelay = 5 * time.Minute
+)
+
+func TestExponentialBackendOffHandler(t *testing.T) {
+	handler := NewExponentialBackendOffHandler(testRetry, testMinRetryDelay, testMaxRetryDelay)
+	expectDelay := testMinRetryDelay
+
+	for i := 0; i < testRetry; i++ {
+		delay, err := handler.NextRetryDelay()
+		if err != nil {
+			t.Errorf("Expect error to be nil, but got %v", err)
+		}
+
+		if !(delay >= expectDelay && delay <= 2*expectDelay) {
+			t.Errorf("Expect retry delay >= %v and delay <= %v, but got %v", expectDelay, 2*expectDelay, delay)
+		}
+
+		if delay > testMaxRetryDelay {
+			t.Errorf("Expect delay to be <= %v, but got %v", testMaxRetryDelay, delay)
+		}
+		expectDelay = delay
+	}
+
+	_, err := handler.NextRetryDelay()
+	if err != ErrRetriesExceeded {
+		t.Errorf("Expect error to be %v, but got %v", ErrRetriesExceeded, err)
+	}
+
+	handler.ResetRetryDelay()
+
+	delay, err := handler.NextRetryDelay()
+	if err != nil {
+		t.Errorf("Expect error to be nil, but got %v", err)
+	}
+
+	if testMinRetryDelay != delay {
+		t.Errorf("Expect retry delay = %v, but got %v", expectDelay, delay)
+	}
+}

pkg/neg/syncers/retry_handler.go

@@ -0,0 +1,92 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed 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 syncers
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/clock"
+)
+
+var ErrHandlerRetrying = fmt.Errorf("retry handler is retrying")
+
+// retryHandler encapsulates logic that handles retry
+type retryHandler interface {
+	// Retry triggers retry
+	Retry() error
+	// Reset resets handler internals
+	Reset()
+}
+
+// backoffRetryHandler handles retry with back off delay
+// At any time, there is only one ongoing retry flow running
+type backoffRetryHandler struct {
+	// stateLock protects internal states and channels
+	stateLock sync.Mutex
+	// internal states
+	retrying bool
+
+	// backoff delay handling
+	clock   clock.Clock
+	backoff backoffHandler
+
+	// retryFunc called on retry
+	retryFunc func()
+}
+
+func NewDelayRetryHandler(retryFunc func(), backoff backoffHandler) *backoffRetryHandler {
+	return &backoffRetryHandler{
+		retrying:  false,
+		clock:     clock.RealClock{},
+		backoff:   backoff,
+		retryFunc: retryFunc,
+	}
+}
+
+// Retry triggers retry with back off
+// At any time, there is only one onging retry allowed.
+func (h *backoffRetryHandler) Retry() error {
+	h.stateLock.Lock()
+	defer h.stateLock.Unlock()
+
+	if h.retrying {
+		return ErrHandlerRetrying
+	}
+
+	delay, err := h.backoff.NextRetryDelay()
+	if err != nil {
+		return err
+	}
+
+	h.retrying = true
+	go func() {
+		<-h.clock.After(delay)
+		func() {
+			h.stateLock.Lock()
+			defer h.stateLock.Unlock()
+			h.retryFunc()
+			h.retrying = false
+		}()
+	}()
+	return nil
+}
+
+// Reset resets internal back off delay handler
+func (h *backoffRetryHandler) Reset() {
+	h.backoff.ResetRetryDelay()
+}

pkg/neg/syncers/retry_handler_test.go

@@ -0,0 +1,85 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed 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 syncers
+
+import (
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sync"
+	"testing"
+	"time"
+)
+
+const (
+	testMaxRetries      = 10
+	smallTestRetryDelay = 3 * time.Second
+)
+
+type retryHandlerTestHelper struct {
+	lock  sync.Mutex
+	count int
+}
+
+func (h *retryHandlerTestHelper) incrementCount() {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+	h.count += 1
+}
+
+func (h *retryHandlerTestHelper) getCount() int {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+	return h.count
+}
+
+func verifyRetryHandler(t *testing.T, expectCount int, helper *retryHandlerTestHelper) {
+	if err := wait.PollImmediate(time.Second, 2*smallTestRetryDelay, func() (bool, error) {
+		if helper.getCount() == expectCount {
+			return true, nil
+		}
+		return false, nil
+	}); err != nil {
+		t.Fatalf("Failed to retry. Expect counter == %d, but got: %v", expectCount, helper.getCount())
+	}
+}
+
+func TestBackoffRetryHandler_Retry(t *testing.T) {
+	helper := &retryHandlerTestHelper{}
+	handler := NewDelayRetryHandler(helper.incrementCount, NewExponentialBackendOffHandler(testMaxRetries, smallTestRetryDelay, testMaxRetryDelay))
+
+	if err := handler.Retry(); err != nil {
+		t.Fatalf("Expect no error, but got %v", err)
+	}
+
+	if err := handler.Retry(); err != ErrHandlerRetrying {
+		t.Fatalf("Expect error %v, but got %v", ErrHandlerRetrying, err)
+	}
+
+	verifyRetryHandler(t, 1, helper)
+
+	if err := handler.Retry(); err != nil {
+		t.Fatalf("Expect no error, but got %v", err)
+	}
+
+	verifyRetryHandler(t, 2, helper)
+
+	handler.Reset()
+	if err := handler.Retry(); err != nil {
+		t.Fatalf("Expect no error, but got %v", err)
+	}
+
+	verifyRetryHandler(t, 3, helper)
+}