all: standardize ResumableUpload.Upload retry with cloud.google.com/go
Standardizes all Upload retries with cloud.google.com/go by using gax
instead of a default implementation.
The functional change is that Upload initial retry pause period is
reduced from 250ms to 100ms, and the maximum backoff time is increased
from 16s to 30s.
Clients that use ResumableUpload.Upload include:
- storage
- youtube
- drive
- bigquery
- gmail
- androidpublisher
- playcustomapp
- dfareporting
- remotebuildexecution
- groupsmigration
- fusiontables
Change-Id: Ic646c8621e2dbb028c50410129c62463f0aed3c1
Reviewed-on: https://code-review.googlesource.com/c/google-api-go-client/+/43470
Reviewed-by: kokoro <noreply+kokoro@google.com>
Reviewed-by: Chris Broadfoot <cbro@google.com>
diff --git a/gensupport/backoff.go b/gensupport/backoff.go
deleted file mode 100644
index 94b7789..0000000
--- a/gensupport/backoff.go
+++ /dev/null
@@ -1,51 +0,0 @@
-// Copyright 2016 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-package gensupport
-
-import (
- "math/rand"
- "time"
-)
-
-// BackoffStrategy defines the set of functions that a backoff-er must
-// implement.
-type BackoffStrategy interface {
- // Pause returns the duration of the next pause and true if the operation should be
- // retried, or false if no further retries should be attempted.
- Pause() (time.Duration, bool)
-
- // Reset restores the strategy to its initial state.
- Reset()
-}
-
-// ExponentialBackoff performs exponential backoff as per https://en.wikipedia.org/wiki/Exponential_backoff.
-// The initial pause time is given by Base.
-// Once the total pause time exceeds Max, Pause will indicate no further retries.
-type ExponentialBackoff struct {
- Base time.Duration
- Max time.Duration
- total time.Duration
- n uint
-}
-
-// Pause returns the amount of time the caller should wait.
-func (eb *ExponentialBackoff) Pause() (time.Duration, bool) {
- if eb.total > eb.Max {
- return 0, false
- }
-
- // The next pause is selected from randomly from [0, 2^n * Base).
- d := time.Duration(rand.Int63n((1 << eb.n) * int64(eb.Base)))
- eb.total += d
- eb.n++
- return d, true
-}
-
-// Reset resets the backoff strategy such that the next Pause call will begin
-// counting from the start. It is not safe to call concurrently with Pause.
-func (eb *ExponentialBackoff) Reset() {
- eb.n = 0
- eb.total = 0
-}
diff --git a/gensupport/backoff_test.go b/gensupport/backoff_test.go
deleted file mode 100644
index cd1e58c..0000000
--- a/gensupport/backoff_test.go
+++ /dev/null
@@ -1,46 +0,0 @@
-// Copyright 2017 Google LLC
-//
-// 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 gensupport
-
-import (
- "testing"
- "time"
-)
-
-func TestBackoff(t *testing.T) {
- eb := &ExponentialBackoff{Base: time.Millisecond, Max: time.Second}
-
- var total time.Duration
- for n, max := 0, 2*time.Millisecond; ; n, max = n+1, max*2 {
- if n > 100 {
- // There's less than 1 in 10^28 of taking longer than 100 iterations,
- // so this is just to check we don't have an infinite loop.
- t.Fatalf("Failed to timeout after 100 iterations.")
- }
- pause, retry := eb.Pause()
- if !retry {
- break
- }
-
- if 0 > pause || pause >= max {
- t.Errorf("Iteration %d: pause = %v; want in range [0, %v)", n, pause, max)
- }
- total += pause
- }
-
- if total < time.Second {
- t.Errorf("Total time = %v; want > %v", total, time.Second)
- }
-}
diff --git a/gensupport/resumable.go b/gensupport/resumable.go
index 2552a6a..61c90dd 100644
--- a/gensupport/resumable.go
+++ b/gensupport/resumable.go
@@ -12,8 +12,21 @@
"net/http"
"sync"
"time"
+
+ gax "github.com/googleapis/gax-go/v2"
)
+// Backoff is an interface around gax.Backoff's Pause method, allowing tests to provide their
+// own implementation.
+type Backoff interface {
+ Pause() time.Duration
+}
+
+// This is declared as a global variable so that tests can overwrite it.
+var backoff = func() Backoff {
+ return &gax.Backoff{Initial: 100 * time.Millisecond}
+}
+
const (
// statusTooManyRequests is returned by the storage API if the
// per-project limits have been temporarily exceeded. The request
@@ -39,9 +52,6 @@
// Callback is an optional function that will be periodically called with the cumulative number of bytes uploaded.
Callback func(int64)
-
- // If not specified, a default exponential backoff strategy will be used.
- Backoff BackoffStrategy
}
// Progress returns the number of bytes uploaded at this point.
@@ -157,11 +167,25 @@
// Exactly one of resp or err will be nil. If resp is non-nil, the caller must call resp.Body.Close.
func (rx *ResumableUpload) Upload(ctx context.Context) (resp *http.Response, err error) {
var pause time.Duration
- backoff := rx.Backoff
- if backoff == nil {
- backoff = DefaultBackoffStrategy()
+
+ var shouldRetry = func(status int, err error) bool {
+ if 500 <= status && status <= 599 {
+ return true
+ }
+ if status == statusTooManyRequests {
+ return true
+ }
+ if err == io.ErrUnexpectedEOF {
+ return true
+ }
+ if err, ok := err.(interface{ Temporary() bool }); ok {
+ return err.Temporary()
+ }
+ return false
}
+ bo := backoff()
+
for {
// Ensure that we return in the case of cancelled context, even if pause is 0.
if contextDone(ctx) {
@@ -182,21 +206,17 @@
// Check if we should retry the request.
if shouldRetry(status, err) {
- var retry bool
- pause, retry = backoff.Pause()
- if retry {
- if resp != nil && resp.Body != nil {
- resp.Body.Close()
- }
- continue
+ pause = bo.Pause()
+ if resp != nil && resp.Body != nil {
+ resp.Body.Close()
}
+ continue
}
// If the chunk was uploaded successfully, but there's still
// more to go, upload the next chunk without any delay.
if statusResumeIncomplete(resp) {
pause = 0
- backoff.Reset()
resp.Body.Close()
continue
}
diff --git a/gensupport/resumable_test.go b/gensupport/resumable_test.go
index 287ed7c..94848fd 100644
--- a/gensupport/resumable_test.go
+++ b/gensupport/resumable_test.go
@@ -112,6 +112,7 @@
func TestInterruptedTransferChunks(t *testing.T) {
type testCase struct {
+ name string
data string
chunkSize int
events []event
@@ -120,6 +121,7 @@
for _, tc := range []testCase{
{
+ name: "large",
data: strings.Repeat("a", 300),
chunkSize: 90,
events: []event{
@@ -130,10 +132,10 @@
{"bytes 180-269/*", 308},
{"bytes 270-299/300", 200},
},
-
wantProgress: []int64{90, 180, 270, 300},
},
{
+ name: "small",
data: strings.Repeat("a", 20),
chunkSize: 10,
events: []event{
@@ -145,51 +147,56 @@
{"bytes */20", http.StatusServiceUnavailable},
{"bytes */20", 200},
},
-
wantProgress: []int64{10, 20},
},
} {
- media := strings.NewReader(tc.data)
+ t.Run(tc.name, func(t *testing.T) {
+ media := strings.NewReader(tc.data)
- tr := &interruptibleTransport{
- buf: make([]byte, 0, len(tc.data)),
- events: tc.events,
- bodies: bodyTracker{},
- }
-
- pr := progressRecorder{}
- rx := &ResumableUpload{
- Client: &http.Client{Transport: tr},
- Media: NewMediaBuffer(media, tc.chunkSize),
- MediaType: "text/plain",
- Callback: pr.ProgressUpdate,
- Backoff: NoPauseStrategy,
- }
- res, err := rx.Upload(context.Background())
- if err == nil {
- res.Body.Close()
- }
- if err != nil || res == nil || res.StatusCode != http.StatusOK {
- if res == nil {
- t.Errorf("Upload not successful, res=nil: %v", err)
- } else {
- t.Errorf("Upload not successful, statusCode=%v: %v", res.StatusCode, err)
+ tr := &interruptibleTransport{
+ buf: make([]byte, 0, len(tc.data)),
+ events: tc.events,
+ bodies: bodyTracker{},
}
- }
- if !reflect.DeepEqual(tr.buf, []byte(tc.data)) {
- t.Errorf("transferred contents:\ngot %s\nwant %s", tr.buf, tc.data)
- }
- if !reflect.DeepEqual(pr.updates, tc.wantProgress) {
- t.Errorf("progress updates: got %v, want %v", pr.updates, tc.wantProgress)
- }
+ pr := progressRecorder{}
+ rx := &ResumableUpload{
+ Client: &http.Client{Transport: tr},
+ Media: NewMediaBuffer(media, tc.chunkSize),
+ MediaType: "text/plain",
+ Callback: pr.ProgressUpdate,
+ }
- if len(tr.events) > 0 {
- t.Errorf("did not observe all expected events. leftover events: %v", tr.events)
- }
- if len(tr.bodies) > 0 {
- t.Errorf("unclosed request bodies: %v", tr.bodies)
- }
+ oldBackoff := backoff
+ backoff = func() Backoff { return new(NoPauseBackoff) }
+ defer func() { backoff = oldBackoff }()
+
+ res, err := rx.Upload(context.Background())
+ if err == nil {
+ res.Body.Close()
+ }
+ if err != nil || res == nil || res.StatusCode != http.StatusOK {
+ if res == nil {
+ t.Fatalf("Upload not successful, res=nil: %v", err)
+ } else {
+ t.Fatalf("Upload not successful, statusCode=%v, err=%v", res.StatusCode, err)
+ }
+ }
+ if !reflect.DeepEqual(tr.buf, []byte(tc.data)) {
+ t.Fatalf("transferred contents:\ngot %s\nwant %s", tr.buf, tc.data)
+ }
+
+ if !reflect.DeepEqual(pr.updates, tc.wantProgress) {
+ t.Fatalf("progress updates: got %v, want %v", pr.updates, tc.wantProgress)
+ }
+
+ if len(tr.events) > 0 {
+ t.Fatalf("did not observe all expected events. leftover events: %v", tr.events)
+ }
+ if len(tr.bodies) > 0 {
+ t.Errorf("unclosed request bodies: %v", tr.bodies)
+ }
+ })
}
}
@@ -210,16 +217,20 @@
Media: NewMediaBuffer(media, chunkSize),
MediaType: "text/plain",
Callback: pr.ProgressUpdate,
- Backoff: NoPauseStrategy,
}
+
+ oldBackoff := backoff
+ backoff = func() Backoff { return new(NoPauseBackoff) }
+ defer func() { backoff = oldBackoff }()
+
ctx, cancelFunc := context.WithCancel(context.Background())
cancelFunc() // stop the upload that hasn't started yet
res, err := rx.Upload(ctx)
if err != context.Canceled {
- t.Errorf("Upload err: got: %v; want: context cancelled", err)
+ t.Fatalf("Upload err: got: %v; want: context cancelled", err)
}
if res != nil {
- t.Errorf("Upload result: got: %v; want: nil", res)
+ t.Fatalf("Upload result: got: %v; want: nil", res)
}
if pr.updates != nil {
t.Errorf("progress updates: got %v; want: nil", pr.updates)
@@ -259,20 +270,24 @@
Media: NewMediaBuffer(media, chunkSize),
MediaType: "text/plain",
Callback: pr.ProgressUpdate,
- Backoff: NoPauseStrategy,
}
+
+ oldBackoff := backoff
+ backoff = func() Backoff { return new(NoPauseBackoff) }
+ defer func() { backoff = oldBackoff }()
+
res, err := rx.Upload(ctx)
if err != context.Canceled {
- t.Errorf("Upload err: got: %v; want: context cancelled", err)
+ t.Fatalf("Upload err: got: %v; want: context cancelled", err)
}
if res != nil {
- t.Errorf("Upload result: got: %v; want: nil", res)
+ t.Fatalf("Upload result: got: %v; want: nil", res)
}
if got, want := tr.buf, []byte(strings.Repeat("a", chunkSize*2)); !reflect.DeepEqual(got, want) {
- t.Errorf("transferred contents:\ngot %s\nwant %s", got, want)
+ t.Fatalf("transferred contents:\ngot %s\nwant %s", got, want)
}
if got, want := pr.updates, []int64{chunkSize, chunkSize * 2}; !reflect.DeepEqual(got, want) {
- t.Errorf("progress updates: got %v; want: %v", got, want)
+ t.Fatalf("progress updates: got %v; want: %v", got, want)
}
if len(tr.bodies) > 0 {
t.Errorf("unclosed request bodies: %v", tr.bodies)
diff --git a/gensupport/retry.go b/gensupport/retry.go
deleted file mode 100644
index fdde3f4..0000000
--- a/gensupport/retry.go
+++ /dev/null
@@ -1,84 +0,0 @@
-// Copyright 2017 Google LLC
-//
-// 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 gensupport
-
-import (
- "context"
- "io"
- "net"
- "net/http"
- "time"
-)
-
-// Retry invokes the given function, retrying it multiple times if the connection failed or
-// the HTTP status response indicates the request should be attempted again. ctx may be nil.
-func Retry(ctx context.Context, f func() (*http.Response, error), backoff BackoffStrategy) (*http.Response, error) {
- for {
- resp, err := f()
-
- var status int
- if resp != nil {
- status = resp.StatusCode
- }
-
- // Return if we shouldn't retry.
- pause, retry := backoff.Pause()
- if !shouldRetry(status, err) || !retry {
- return resp, err
- }
-
- // Ensure the response body is closed, if any.
- if resp != nil && resp.Body != nil {
- resp.Body.Close()
- }
-
- // Pause, but still listen to ctx.Done if context is not nil.
- var done <-chan struct{}
- if ctx != nil {
- done = ctx.Done()
- }
- select {
- case <-done:
- return nil, ctx.Err()
- case <-time.After(pause):
- }
- }
-}
-
-// DefaultBackoffStrategy returns a default strategy to use for retrying failed upload requests.
-func DefaultBackoffStrategy() BackoffStrategy {
- return &ExponentialBackoff{
- Base: 250 * time.Millisecond,
- Max: 16 * time.Second,
- }
-}
-
-// shouldRetry returns true if the HTTP response / error indicates that the
-// request should be attempted again.
-func shouldRetry(status int, err error) bool {
- if 500 <= status && status <= 599 {
- return true
- }
- if status == statusTooManyRequests {
- return true
- }
- if err == io.ErrUnexpectedEOF {
- return true
- }
- if err, ok := err.(net.Error); ok {
- return err.Temporary()
- }
- return false
-}
diff --git a/gensupport/retry_test.go b/gensupport/retry_test.go
deleted file mode 100644
index d061345..0000000
--- a/gensupport/retry_test.go
+++ /dev/null
@@ -1,150 +0,0 @@
-// Copyright 2017 Google LLC
-//
-// 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 gensupport
-
-import (
- "context"
- "errors"
- "io"
- "net"
- "net/http"
- "testing"
-)
-
-func TestRetry(t *testing.T) {
- testCases := []struct {
- desc string
- respStatus []int // HTTP status codes returned (length indicates number of calls we expect).
- maxRetry int // Max number of calls allowed by the BackoffStrategy.
- wantStatus int // StatusCode of returned response.
- }{
- {
- desc: "First call successful",
- respStatus: []int{200},
- maxRetry: 3,
- wantStatus: 200,
- },
- {
- desc: "Retry before success",
- respStatus: []int{500, 500, 500, 200},
- maxRetry: 3,
- wantStatus: 200,
- },
- {
- desc: "Backoff strategy abandons after 3 retries",
- respStatus: []int{500, 500, 500, 500},
- maxRetry: 3,
- wantStatus: 500,
- },
- {
- desc: "Backoff strategy abandons after 2 retries",
- respStatus: []int{500, 500, 500},
- maxRetry: 2,
- wantStatus: 500,
- },
- }
- for _, tt := range testCases {
- // Function consumes tt.respStatus
- f := func() (*http.Response, error) {
- if len(tt.respStatus) == 0 {
- return nil, errors.New("too many requests to function")
- }
- resp := &http.Response{StatusCode: tt.respStatus[0]}
- tt.respStatus = tt.respStatus[1:]
- return resp, nil
- }
-
- backoff := &LimitRetryStrategy{
- Max: tt.maxRetry,
- Strategy: NoPauseStrategy,
- }
-
- resp, err := Retry(context.Background(), f, backoff)
- if err != nil {
- t.Errorf("%s: Retry returned err %v", tt.desc, err)
- }
- if got := resp.StatusCode; got != tt.wantStatus {
- t.Errorf("%s: Retry returned response with StatusCode=%d; want %d", tt.desc, got, tt.wantStatus)
- }
- if len(tt.respStatus) != 0 {
- t.Errorf("%s: f was not called enough; status codes remaining: %v", tt.desc, tt.respStatus)
- }
- }
-}
-
-type checkCloseReader struct {
- closed bool
-}
-
-func (c *checkCloseReader) Read(p []byte) (n int, err error) { return 0, io.EOF }
-func (c *checkCloseReader) Close() error {
- c.closed = true
- return nil
-}
-
-func TestRetryClosesBody(t *testing.T) {
- var i int
- responses := []*http.Response{
- {StatusCode: 500, Body: &checkCloseReader{}},
- {StatusCode: 500, Body: &checkCloseReader{}},
- {StatusCode: 200, Body: &checkCloseReader{}},
- }
- f := func() (*http.Response, error) {
- resp := responses[i]
- i++
- return resp, nil
- }
-
- resp, err := Retry(context.Background(), f, NoPauseStrategy)
- if err != nil {
- t.Fatalf("Retry returned error: %v", err)
- }
- if resp != responses[2] {
- t.Errorf("Retry returned %v; want %v", resp, responses[2])
- }
- for i, resp := range responses {
- want := i != 2 // Only the last response should not be closed.
- got := resp.Body.(*checkCloseReader).closed
- if got != want {
- t.Errorf("response[%d].Body closed = %t, want %t", i, got, want)
- }
- }
-}
-
-func TestShouldRetry(t *testing.T) {
- testCases := []struct {
- status int
- err error
- want bool
- }{
- {status: 200, want: false},
- {status: 308, want: false},
- {status: 403, want: false},
- {status: 429, want: true},
- {status: 500, want: true},
- {status: 503, want: true},
- {status: 600, want: false},
- {err: io.EOF, want: false},
- {err: errors.New("random badness"), want: false},
- {err: io.ErrUnexpectedEOF, want: true},
- {err: &net.AddrError{}, want: false}, // Not temporary.
- {err: &net.DNSError{IsTimeout: true}, want: true}, // Temporary.
- }
- for _, tt := range testCases {
- if got := shouldRetry(tt.status, tt.err); got != tt.want {
- t.Errorf("shouldRetry(%d, %v) = %t; want %t", tt.status, tt.err, got, tt.want)
- }
- }
-}
diff --git a/gensupport/util_test.go b/gensupport/util_test.go
index f9f31a1..c29cea1 100644
--- a/gensupport/util_test.go
+++ b/gensupport/util_test.go
@@ -27,31 +27,7 @@
return n, nil
}
-// UniformPauseStrategy implements BackoffStrategy with uniform pause.
-type UniformPauseStrategy time.Duration
+// NoPauseBackoff implements backoff strategy with infinite 0-length pauses.
+type NoPauseBackoff struct{}
-func (p UniformPauseStrategy) Pause() (time.Duration, bool) { return time.Duration(p), true }
-func (p UniformPauseStrategy) Reset() {}
-
-// NoPauseStrategy implements BackoffStrategy with infinite 0-length pauses.
-const NoPauseStrategy = UniformPauseStrategy(0)
-
-// LimitRetryStrategy wraps a BackoffStrategy but limits the number of retries.
-type LimitRetryStrategy struct {
- Max int
- Strategy BackoffStrategy
- n int
-}
-
-func (l *LimitRetryStrategy) Pause() (time.Duration, bool) {
- l.n++
- if l.n > l.Max {
- return 0, false
- }
- return l.Strategy.Pause()
-}
-
-func (l *LimitRetryStrategy) Reset() {
- l.n = 0
- l.Strategy.Reset()
-}
+func (bo *NoPauseBackoff) Pause() time.Duration { return 0 }
diff --git a/go.mod b/go.mod
index df804e9..63abbca 100644
--- a/go.mod
+++ b/go.mod
@@ -6,6 +6,7 @@
cloud.google.com/go v0.38.0 // indirect
github.com/golang/protobuf v1.3.1 // indirect
github.com/google/go-cmp v0.3.0
+ github.com/googleapis/gax-go/v2 v2.0.5
github.com/hashicorp/golang-lru v0.5.1 // indirect
go.opencensus.io v0.21.0
golang.org/x/lint v0.0.0-20190409202823-959b441ac422
diff --git a/go.sum b/go.sum
index a703973..e38b0ee 100644
--- a/go.sum
+++ b/go.sum
@@ -19,6 +19,8 @@
github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
+github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM=
+github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU=
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
