Add backoff retry which implements autorest.SendDecorator interface

Author: feiskyer

staging/src/k8s.io/legacy-cloud-providers/azure/retry/BUILD

@@ -2,18 +2,31 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["azure_error.go"],
+    srcs = [
+        "azure_error.go",
+        "azure_retry.go",
+        "doc.go",
+    ],
     importmap = "k8s.io/kubernetes/vendor/k8s.io/legacy-cloud-providers/azure/retry",
     importpath = "k8s.io/legacy-cloud-providers/azure/retry",
     visibility = ["//visibility:public"],
-    deps = ["//vendor/k8s.io/klog:go_default_library"],
+    deps = [
+        "//vendor/github.com/Azure/go-autorest/autorest:go_default_library",
+        "//vendor/k8s.io/klog:go_default_library",
+    ],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["azure_error_test.go"],
+    srcs = [
+        "azure_error_test.go",
+        "azure_retry_test.go",
+    ],
     embed = [":go_default_library"],
-    deps = ["//vendor/github.com/stretchr/testify/assert:go_default_library"],
+    deps = [
+        "//vendor/github.com/Azure/go-autorest/autorest/mocks:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
+    ],
 )
 
 filegroup(

staging/src/k8s.io/legacy-cloud-providers/azure/retry/azure_error.go

@@ -19,7 +19,9 @@ limitations under the License.
 package retry
 
 import (
+	"bytes"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"strconv"
 	"strings"
@@ -28,6 +30,11 @@ import (
 	"k8s.io/klog"
 )
 
+const (
+	// RetryAfterHeaderKey is the retry-after header key in ARM responses.
+	RetryAfterHeaderKey = "Retry-After"
+)
+
 var (
 	// The function to get current time.
 	now = time.Now
@@ -57,6 +64,15 @@ func (err *Error) Error() error {
 		err.Retriable, err.RetryAfter.String(), err.HTTPStatusCode, err.RawError)
 }
 
+// IsThrottled returns true the if the request is being throttled.
+func (err *Error) IsThrottled() bool {
+	if err == nil {
+		return false
+	}
+
+	return err.HTTPStatusCode == http.StatusTooManyRequests || err.RetryAfter.After(now())
+}
+
 // NewError creates a new Error.
 func NewError(retriable bool, err error) *Error {
 	return &Error{
@@ -73,6 +89,20 @@ func GetRetriableError(err error) *Error {
 	}
 }
 
+// GetRateLimitError creates a new error for rate limiting.
+func GetRateLimitError(isWrite bool, opName string) *Error {
+	opType := "read"
+	if isWrite {
+		opType = "write"
+	}
+	return GetRetriableError(fmt.Errorf("azure cloud provider rate limited(%s) for operation %q", opType, opName))
+}
+
+// GetThrottlingError creates a new error for throttling.
+func GetThrottlingError(operation, reason string) *Error {
+	return GetRetriableError(fmt.Errorf("azure cloud provider throttled for operation %s with reason %q", operation, reason))
+}
+
 // GetError gets a new Error based on resp and error.
 func GetError(resp *http.Response, err error) *Error {
 	if err == nil && resp == nil {
@@ -88,12 +118,8 @@ func GetError(resp *http.Response, err error) *Error {
 	if retryAfterDuration := getRetryAfter(resp); retryAfterDuration != 0 {
 		retryAfter = now().Add(retryAfterDuration)
 	}
-	rawError := err
-	if err == nil && resp != nil {
-		rawError = fmt.Errorf("HTTP response: %v", resp.StatusCode)
-	}
 	return &Error{
-		RawError:       rawError,
+		RawError:       getRawError(resp, err),
 		RetryAfter:     retryAfter,
 		Retriable:      shouldRetryHTTPRequest(resp, err),
 		HTTPStatusCode: getHTTPStatusCode(resp),
@@ -114,6 +140,27 @@ func isSuccessHTTPResponse(resp *http.Response) bool {
 	return false
 }
 
+func getRawError(resp *http.Response, err error) error {
+	if err != nil {
+		return err
+	}
+
+	if resp == nil || resp.Body == nil {
+		return fmt.Errorf("empty HTTP response")
+	}
+
+	// return the http status if unabled to get response body.
+	defer resp.Body.Close()
+	respBody, _ := ioutil.ReadAll(resp.Body)
+	resp.Body = ioutil.NopCloser(bytes.NewReader(respBody))
+	if len(respBody) == 0 {
+		return fmt.Errorf("HTTP status code (%d)", resp.StatusCode)
+	}
+
+	// return the raw response body.
+	return fmt.Errorf("%s", string(respBody))
+}
+
 func getHTTPStatusCode(resp *http.Response) int {
 	if resp == nil {
 		return -1
@@ -151,7 +198,7 @@ func getRetryAfter(resp *http.Response) time.Duration {
 		return 0
 	}
 
-	ra := resp.Header.Get("Retry-After")
+	ra := resp.Header.Get(RetryAfterHeaderKey)
 	if ra == "" {
 		return 0
 	}

staging/src/k8s.io/legacy-cloud-providers/azure/retry/azure_error_test.go

@@ -19,7 +19,9 @@ limitations under the License.
 package retry
 
 import (
+	"bytes"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"testing"
 	"time"
@@ -44,27 +46,27 @@ func TestGetError(t *testing.T) {
 		},
 		{
 			code: http.StatusOK,
-			err:  fmt.Errorf("some error"),
+			err:  fmt.Errorf("unknown error"),
 			expected: &Error{
 				Retriable:      true,
 				HTTPStatusCode: http.StatusOK,
-				RawError:       fmt.Errorf("some error"),
+				RawError:       fmt.Errorf("unknown error"),
 			},
 		},
 		{
 			code: http.StatusBadRequest,
 			expected: &Error{
 				Retriable:      false,
 				HTTPStatusCode: http.StatusBadRequest,
-				RawError:       fmt.Errorf("HTTP response: 400"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 		{
 			code: http.StatusInternalServerError,
 			expected: &Error{
 				Retriable:      true,
 				HTTPStatusCode: http.StatusInternalServerError,
-				RawError:       fmt.Errorf("HTTP response: 500"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 		{
@@ -83,7 +85,7 @@ func TestGetError(t *testing.T) {
 				Retriable:      true,
 				HTTPStatusCode: http.StatusTooManyRequests,
 				RetryAfter:     now().Add(100 * time.Second),
-				RawError:       fmt.Errorf("HTTP response: 429"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 	}
@@ -92,6 +94,7 @@ func TestGetError(t *testing.T) {
 		resp := &http.Response{
 			StatusCode: test.code,
 			Header:     http.Header{},
+			Body:       ioutil.NopCloser(bytes.NewReader([]byte("some error"))),
 		}
 		if test.retryAfter != 0 {
 			resp.Header.Add("Retry-After", fmt.Sprintf("%d", test.retryAfter))
@@ -138,15 +141,15 @@ func TestGetStatusNotFoundAndForbiddenIgnoredError(t *testing.T) {
 			expected: &Error{
 				Retriable:      false,
 				HTTPStatusCode: http.StatusBadRequest,
-				RawError:       fmt.Errorf("HTTP response: 400"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 		{
 			code: http.StatusInternalServerError,
 			expected: &Error{
 				Retriable:      true,
 				HTTPStatusCode: http.StatusInternalServerError,
-				RawError:       fmt.Errorf("HTTP response: 500"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 		{
@@ -165,7 +168,7 @@ func TestGetStatusNotFoundAndForbiddenIgnoredError(t *testing.T) {
 				Retriable:      true,
 				HTTPStatusCode: http.StatusTooManyRequests,
 				RetryAfter:     now().Add(100 * time.Second),
-				RawError:       fmt.Errorf("HTTP response: 429"),
+				RawError:       fmt.Errorf("some error"),
 			},
 		},
 	}
@@ -174,6 +177,7 @@ func TestGetStatusNotFoundAndForbiddenIgnoredError(t *testing.T) {
 		resp := &http.Response{
 			StatusCode: test.code,
 			Header:     http.Header{},
+			Body:       ioutil.NopCloser(bytes.NewReader([]byte("some error"))),
 		}
 		if test.retryAfter != 0 {
 			resp.Header.Add("Retry-After", fmt.Sprintf("%d", test.retryAfter))
@@ -251,3 +255,38 @@ func TestIsSuccessResponse(t *testing.T) {
 		}
 	}
 }
+
+func TestIsThrottled(t *testing.T) {
+	tests := []struct {
+		err      *Error
+		expected bool
+	}{
+		{
+			err:      nil,
+			expected: false,
+		},
+		{
+			err: &Error{
+				HTTPStatusCode: http.StatusOK,
+			},
+			expected: false,
+		},
+		{
+			err: &Error{
+				HTTPStatusCode: http.StatusTooManyRequests,
+			},
+			expected: true,
+		},
+		{
+			err: &Error{
+				RetryAfter: time.Now().Add(time.Hour),
+			},
+			expected: true,
+		},
+	}
+
+	for _, test := range tests {
+		real := test.err.IsThrottled()
+		assert.Equal(t, test.expected, real)
+	}
+}