mirror of
https://github.com/ceph/ceph-csi.git
synced 2025-06-13 02:33:34 +00:00
rebase: update kubernetes dep to 1.24.0
As kubernetes 1.24.0 is released, updating kubernetes dependencies to 1.24.0 updates: #3086 Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
This commit is contained in:
committed by
mergify[bot]
parent
fc1529f268
commit
c4f79d455f
30
vendor/k8s.io/client-go/rest/OWNERS
generated
vendored
30
vendor/k8s.io/client-go/rest/OWNERS
generated
vendored
@ -1,22 +1,14 @@
|
||||
# See the OWNERS docs at https://go.k8s.io/owners
|
||||
|
||||
reviewers:
|
||||
- thockin
|
||||
- smarterclayton
|
||||
- caesarxuchao
|
||||
- wojtek-t
|
||||
- deads2k
|
||||
- brendandburns
|
||||
- liggitt
|
||||
- sttts
|
||||
- luxas
|
||||
- dims
|
||||
- errordeveloper
|
||||
- hongchaodeng
|
||||
- krousey
|
||||
- resouer
|
||||
- cjcullen
|
||||
- rmmh
|
||||
- asalkeld
|
||||
- juanvallejo
|
||||
- lojies
|
||||
- thockin
|
||||
- smarterclayton
|
||||
- caesarxuchao
|
||||
- wojtek-t
|
||||
- deads2k
|
||||
- liggitt
|
||||
- sttts
|
||||
- luxas
|
||||
- dims
|
||||
- cjcullen
|
||||
- lojies
|
||||
|
254
vendor/k8s.io/client-go/rest/request.go
generated
vendored
254
vendor/k8s.io/client-go/rest/request.go
generated
vendored
@ -82,6 +82,12 @@ func (r *RequestConstructionError) Error() string {
|
||||
|
||||
var noBackoff = &NoBackoff{}
|
||||
|
||||
type requestRetryFunc func(maxRetries int) WithRetry
|
||||
|
||||
func defaultRequestRetryFn(maxRetries int) WithRetry {
|
||||
return &withRetry{maxRetries: maxRetries}
|
||||
}
|
||||
|
||||
// Request allows for building up a request to a server in a chained fashion.
|
||||
// Any errors are stored until the end of your call, so you only have to
|
||||
// check once.
|
||||
@ -93,6 +99,7 @@ type Request struct {
|
||||
rateLimiter flowcontrol.RateLimiter
|
||||
backoff BackoffManager
|
||||
timeout time.Duration
|
||||
maxRetries int
|
||||
|
||||
// generic components accessible via method setters
|
||||
verb string
|
||||
@ -109,9 +116,10 @@ type Request struct {
|
||||
subresource string
|
||||
|
||||
// output
|
||||
err error
|
||||
body io.Reader
|
||||
retry WithRetry
|
||||
err error
|
||||
body io.Reader
|
||||
|
||||
retryFn requestRetryFunc
|
||||
}
|
||||
|
||||
// NewRequest creates a new request helper object for accessing runtime.Objects on a server.
|
||||
@ -142,7 +150,8 @@ func NewRequest(c *RESTClient) *Request {
|
||||
backoff: backoff,
|
||||
timeout: timeout,
|
||||
pathPrefix: pathPrefix,
|
||||
retry: &withRetry{maxRetries: 10},
|
||||
maxRetries: 10,
|
||||
retryFn: defaultRequestRetryFn,
|
||||
warningHandler: c.warningHandler,
|
||||
}
|
||||
|
||||
@ -408,7 +417,10 @@ func (r *Request) Timeout(d time.Duration) *Request {
|
||||
// function is specifically called with a different value.
|
||||
// A zero maxRetries prevent it from doing retires and return an error immediately.
|
||||
func (r *Request) MaxRetries(maxRetries int) *Request {
|
||||
r.retry.SetMaxRetries(maxRetries)
|
||||
if maxRetries < 0 {
|
||||
maxRetries = 0
|
||||
}
|
||||
r.maxRetries = maxRetries
|
||||
return r
|
||||
}
|
||||
|
||||
@ -496,84 +508,6 @@ func (r *Request) URL() *url.URL {
|
||||
return finalURL
|
||||
}
|
||||
|
||||
// finalURLTemplate is similar to URL(), but will make all specific parameter values equal
|
||||
// - instead of name or namespace, "{name}" and "{namespace}" will be used, and all query
|
||||
// parameters will be reset. This creates a copy of the url so as not to change the
|
||||
// underlying object.
|
||||
func (r Request) finalURLTemplate() url.URL {
|
||||
newParams := url.Values{}
|
||||
v := []string{"{value}"}
|
||||
for k := range r.params {
|
||||
newParams[k] = v
|
||||
}
|
||||
r.params = newParams
|
||||
url := r.URL()
|
||||
|
||||
segments := strings.Split(url.Path, "/")
|
||||
groupIndex := 0
|
||||
index := 0
|
||||
trimmedBasePath := ""
|
||||
if url != nil && r.c.base != nil && strings.Contains(url.Path, r.c.base.Path) {
|
||||
p := strings.TrimPrefix(url.Path, r.c.base.Path)
|
||||
if !strings.HasPrefix(p, "/") {
|
||||
p = "/" + p
|
||||
}
|
||||
// store the base path that we have trimmed so we can append it
|
||||
// before returning the URL
|
||||
trimmedBasePath = r.c.base.Path
|
||||
segments = strings.Split(p, "/")
|
||||
groupIndex = 1
|
||||
}
|
||||
if len(segments) <= 2 {
|
||||
return *url
|
||||
}
|
||||
|
||||
const CoreGroupPrefix = "api"
|
||||
const NamedGroupPrefix = "apis"
|
||||
isCoreGroup := segments[groupIndex] == CoreGroupPrefix
|
||||
isNamedGroup := segments[groupIndex] == NamedGroupPrefix
|
||||
if isCoreGroup {
|
||||
// checking the case of core group with /api/v1/... format
|
||||
index = groupIndex + 2
|
||||
} else if isNamedGroup {
|
||||
// checking the case of named group with /apis/apps/v1/... format
|
||||
index = groupIndex + 3
|
||||
} else {
|
||||
// this should not happen that the only two possibilities are /api... and /apis..., just want to put an
|
||||
// outlet here in case more API groups are added in future if ever possible:
|
||||
// https://kubernetes.io/docs/concepts/overview/kubernetes-api/#api-groups
|
||||
// if a wrong API groups name is encountered, return the {prefix} for url.Path
|
||||
url.Path = "/{prefix}"
|
||||
url.RawQuery = ""
|
||||
return *url
|
||||
}
|
||||
//switch segLength := len(segments) - index; segLength {
|
||||
switch {
|
||||
// case len(segments) - index == 1:
|
||||
// resource (with no name) do nothing
|
||||
case len(segments)-index == 2:
|
||||
// /$RESOURCE/$NAME: replace $NAME with {name}
|
||||
segments[index+1] = "{name}"
|
||||
case len(segments)-index == 3:
|
||||
if segments[index+2] == "finalize" || segments[index+2] == "status" {
|
||||
// /$RESOURCE/$NAME/$SUBRESOURCE: replace $NAME with {name}
|
||||
segments[index+1] = "{name}"
|
||||
} else {
|
||||
// /namespace/$NAMESPACE/$RESOURCE: replace $NAMESPACE with {namespace}
|
||||
segments[index+1] = "{namespace}"
|
||||
}
|
||||
case len(segments)-index >= 4:
|
||||
segments[index+1] = "{namespace}"
|
||||
// /namespace/$NAMESPACE/$RESOURCE/$NAME: replace $NAMESPACE with {namespace}, $NAME with {name}
|
||||
if segments[index+3] != "finalize" && segments[index+3] != "status" {
|
||||
// /$RESOURCE/$NAME/$SUBRESOURCE: replace $NAME with {name}
|
||||
segments[index+3] = "{name}"
|
||||
}
|
||||
}
|
||||
url.Path = path.Join(trimmedBasePath, path.Join(segments...))
|
||||
return *url
|
||||
}
|
||||
|
||||
func (r *Request) tryThrottleWithInfo(ctx context.Context, retryInfo string) error {
|
||||
if r.rateLimiter == nil {
|
||||
return nil
|
||||
@ -582,7 +516,9 @@ func (r *Request) tryThrottleWithInfo(ctx context.Context, retryInfo string) err
|
||||
now := time.Now()
|
||||
|
||||
err := r.rateLimiter.Wait(ctx)
|
||||
|
||||
if err != nil {
|
||||
err = fmt.Errorf("client rate limiter Wait returned an error: %w", err)
|
||||
}
|
||||
latency := time.Since(now)
|
||||
|
||||
var message string
|
||||
@ -601,7 +537,7 @@ func (r *Request) tryThrottleWithInfo(ctx context.Context, retryInfo string) err
|
||||
// but we use a throttled logger to prevent spamming.
|
||||
globalThrottledLogger.Infof("%s", message)
|
||||
}
|
||||
metrics.RateLimiterLatency.Observe(ctx, r.verb, r.finalURLTemplate(), latency)
|
||||
metrics.RateLimiterLatency.Observe(ctx, r.verb, *r.URL(), latency)
|
||||
|
||||
return err
|
||||
}
|
||||
@ -688,34 +624,21 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
|
||||
}
|
||||
return false
|
||||
}
|
||||
var retryAfter *RetryAfter
|
||||
retry := r.retryFn(r.maxRetries)
|
||||
url := r.URL().String()
|
||||
for {
|
||||
if err := retry.Before(ctx, r); err != nil {
|
||||
return nil, retry.WrapPreviousError(err)
|
||||
}
|
||||
|
||||
req, err := r.newHTTPRequest(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
r.backoff.Sleep(r.backoff.CalculateBackoff(r.URL()))
|
||||
if retryAfter != nil {
|
||||
// We are retrying the request that we already send to apiserver
|
||||
// at least once before.
|
||||
// This request should also be throttled with the client-internal rate limiter.
|
||||
if err := r.tryThrottleWithInfo(ctx, retryAfter.Reason); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
retryAfter = nil
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
updateURLMetrics(ctx, r, resp, err)
|
||||
if r.c.base != nil {
|
||||
if err != nil {
|
||||
r.backoff.UpdateBackoff(r.c.base, err, 0)
|
||||
} else {
|
||||
r.backoff.UpdateBackoff(r.c.base, err, resp.StatusCode)
|
||||
}
|
||||
}
|
||||
retry.After(ctx, r, resp, err)
|
||||
if err == nil && resp.StatusCode == http.StatusOK {
|
||||
return r.newStreamWatcher(resp)
|
||||
}
|
||||
@ -723,14 +646,8 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
|
||||
done, transformErr := func() (bool, error) {
|
||||
defer readAndCloseResponseBody(resp)
|
||||
|
||||
var retry bool
|
||||
retryAfter, retry = r.retry.NextRetry(req, resp, err, isErrRetryableFunc)
|
||||
if retry {
|
||||
err := r.retry.BeforeNextRetry(ctx, r.backoff, retryAfter, url, r.body)
|
||||
if err == nil {
|
||||
return false, nil
|
||||
}
|
||||
klog.V(4).Infof("Could not retry request - %v", err)
|
||||
if retry.IsNextRetry(ctx, r, req, resp, err, isErrRetryableFunc) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if resp == nil {
|
||||
@ -751,7 +668,7 @@ func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {
|
||||
// we need to return the error object from that.
|
||||
err = transformErr
|
||||
}
|
||||
return nil, err
|
||||
return nil, retry.WrapPreviousError(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -816,9 +733,13 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
|
||||
var retryAfter *RetryAfter
|
||||
retry := r.retryFn(r.maxRetries)
|
||||
url := r.URL().String()
|
||||
for {
|
||||
if err := retry.Before(ctx, r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
req, err := r.newHTTPRequest(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -826,27 +747,9 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
|
||||
if r.body != nil {
|
||||
req.Body = ioutil.NopCloser(r.body)
|
||||
}
|
||||
|
||||
r.backoff.Sleep(r.backoff.CalculateBackoff(r.URL()))
|
||||
if retryAfter != nil {
|
||||
// We are retrying the request that we already send to apiserver
|
||||
// at least once before.
|
||||
// This request should also be throttled with the client-internal rate limiter.
|
||||
if err := r.tryThrottleWithInfo(ctx, retryAfter.Reason); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
retryAfter = nil
|
||||
}
|
||||
|
||||
resp, err := client.Do(req)
|
||||
updateURLMetrics(ctx, r, resp, err)
|
||||
if r.c.base != nil {
|
||||
if err != nil {
|
||||
r.backoff.UpdateBackoff(r.URL(), err, 0)
|
||||
} else {
|
||||
r.backoff.UpdateBackoff(r.URL(), err, resp.StatusCode)
|
||||
}
|
||||
}
|
||||
retry.After(ctx, r, resp, err)
|
||||
if err != nil {
|
||||
// we only retry on an HTTP response with 'Retry-After' header
|
||||
return nil, err
|
||||
@ -861,14 +764,8 @@ func (r *Request) Stream(ctx context.Context) (io.ReadCloser, error) {
|
||||
done, transformErr := func() (bool, error) {
|
||||
defer resp.Body.Close()
|
||||
|
||||
var retry bool
|
||||
retryAfter, retry = r.retry.NextRetry(req, resp, err, neverRetryError)
|
||||
if retry {
|
||||
err := r.retry.BeforeNextRetry(ctx, r.backoff, retryAfter, url, r.body)
|
||||
if err == nil {
|
||||
return false, nil
|
||||
}
|
||||
klog.V(4).Infof("Could not retry request - %v", err)
|
||||
if retry.IsNextRetry(ctx, r, req, resp, err, neverRetryError) {
|
||||
return false, nil
|
||||
}
|
||||
result := r.transformResponse(resp, req)
|
||||
if err := result.Error(); err != nil {
|
||||
@ -929,7 +826,7 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
|
||||
//Metrics for total request latency
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
metrics.RequestLatency.Observe(ctx, r.verb, r.finalURLTemplate(), time.Since(start))
|
||||
metrics.RequestLatency.Observe(ctx, r.verb, *r.URL(), time.Since(start))
|
||||
}()
|
||||
|
||||
if r.err != nil {
|
||||
@ -959,31 +856,38 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
|
||||
defer cancel()
|
||||
}
|
||||
|
||||
isErrRetryableFunc := func(req *http.Request, err error) bool {
|
||||
// "Connection reset by peer" or "apiserver is shutting down" are usually a transient errors.
|
||||
// Thus in case of "GET" operations, we simply retry it.
|
||||
// We are not automatically retrying "write" operations, as they are not idempotent.
|
||||
if req.Method != "GET" {
|
||||
return false
|
||||
}
|
||||
// For connection errors and apiserver shutdown errors retry.
|
||||
if net.IsConnectionReset(err) || net.IsProbableEOF(err) {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Right now we make about ten retry attempts if we get a Retry-After response.
|
||||
var retryAfter *RetryAfter
|
||||
retry := r.retryFn(r.maxRetries)
|
||||
for {
|
||||
if err := retry.Before(ctx, r); err != nil {
|
||||
return retry.WrapPreviousError(err)
|
||||
}
|
||||
req, err := r.newHTTPRequest(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.backoff.Sleep(r.backoff.CalculateBackoff(r.URL()))
|
||||
if retryAfter != nil {
|
||||
// We are retrying the request that we already send to apiserver
|
||||
// at least once before.
|
||||
// This request should also be throttled with the client-internal rate limiter.
|
||||
if err := r.tryThrottleWithInfo(ctx, retryAfter.Reason); err != nil {
|
||||
return err
|
||||
}
|
||||
retryAfter = nil
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
updateURLMetrics(ctx, r, resp, err)
|
||||
if err != nil {
|
||||
r.backoff.UpdateBackoff(r.URL(), err, 0)
|
||||
} else {
|
||||
r.backoff.UpdateBackoff(r.URL(), err, resp.StatusCode)
|
||||
// The value -1 or a value of 0 with a non-nil Body indicates that the length is unknown.
|
||||
// https://pkg.go.dev/net/http#Request
|
||||
if req.ContentLength >= 0 && !(req.Body != nil && req.ContentLength == 0) {
|
||||
metrics.RequestSize.Observe(ctx, r.verb, r.URL().Host, float64(req.ContentLength))
|
||||
}
|
||||
retry.After(ctx, r, resp, err)
|
||||
|
||||
done := func() bool {
|
||||
defer readAndCloseResponseBody(resp)
|
||||
@ -996,33 +900,15 @@ func (r *Request) request(ctx context.Context, fn func(*http.Request, *http.Resp
|
||||
fn(req, resp)
|
||||
}
|
||||
|
||||
var retry bool
|
||||
retryAfter, retry = r.retry.NextRetry(req, resp, err, func(req *http.Request, err error) bool {
|
||||
// "Connection reset by peer" or "apiserver is shutting down" are usually a transient errors.
|
||||
// Thus in case of "GET" operations, we simply retry it.
|
||||
// We are not automatically retrying "write" operations, as they are not idempotent.
|
||||
if r.verb != "GET" {
|
||||
return false
|
||||
}
|
||||
// For connection errors and apiserver shutdown errors retry.
|
||||
if net.IsConnectionReset(err) || net.IsProbableEOF(err) {
|
||||
return true
|
||||
}
|
||||
if retry.IsNextRetry(ctx, r, req, resp, err, isErrRetryableFunc) {
|
||||
return false
|
||||
})
|
||||
if retry {
|
||||
err := r.retry.BeforeNextRetry(ctx, r.backoff, retryAfter, req.URL.String(), r.body)
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
klog.V(4).Infof("Could not retry request - %v", err)
|
||||
}
|
||||
|
||||
f(req, resp)
|
||||
return true
|
||||
}()
|
||||
if done {
|
||||
return err
|
||||
return retry.WrapPreviousError(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1041,6 +927,9 @@ func (r *Request) Do(ctx context.Context) Result {
|
||||
if err != nil {
|
||||
return Result{err: err}
|
||||
}
|
||||
if result.err == nil || len(result.body) > 0 {
|
||||
metrics.ResponseSize.Observe(ctx, r.verb, r.URL().Host, float64(len(result.body)))
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
@ -1057,6 +946,9 @@ func (r *Request) DoRaw(ctx context.Context) ([]byte, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if result.err == nil || len(result.body) > 0 {
|
||||
metrics.ResponseSize.Observe(ctx, r.verb, r.URL().Host, float64(len(result.body)))
|
||||
}
|
||||
return result.body, result.err
|
||||
}
|
||||
|
||||
@ -1172,13 +1064,13 @@ func truncateBody(body string) string {
|
||||
// allocating a new string for the body output unless necessary. Uses a simple heuristic to determine
|
||||
// whether the body is printable.
|
||||
func glogBody(prefix string, body []byte) {
|
||||
if klog.V(8).Enabled() {
|
||||
if klogV := klog.V(8); klogV.Enabled() {
|
||||
if bytes.IndexFunc(body, func(r rune) bool {
|
||||
return r < 0x0a
|
||||
}) != -1 {
|
||||
klog.Infof("%s:\n%s", prefix, truncateBody(hex.Dump(body)))
|
||||
klogV.Infof("%s:\n%s", prefix, truncateBody(hex.Dump(body)))
|
||||
} else {
|
||||
klog.Infof("%s: %s", prefix, truncateBody(string(body)))
|
||||
klogV.Infof("%s: %s", prefix, truncateBody(string(body)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
222
vendor/k8s.io/client-go/rest/with_retry.go
generated
vendored
222
vendor/k8s.io/client-go/rest/with_retry.go
generated
vendored
@ -22,6 +22,7 @@ import (
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"k8s.io/klog/v2"
|
||||
@ -51,42 +52,52 @@ var neverRetryError = IsRetryableErrorFunc(func(_ *http.Request, _ error) bool {
|
||||
// Note that WithRetry is not safe for concurrent use by multiple
|
||||
// goroutines without additional locking or coordination.
|
||||
type WithRetry interface {
|
||||
// SetMaxRetries makes the request use the specified integer as a ceiling
|
||||
// for retries upon receiving a 429 status code and the "Retry-After" header
|
||||
// in the response.
|
||||
// A zero maxRetries should prevent from doing any retry and return immediately.
|
||||
SetMaxRetries(maxRetries int)
|
||||
|
||||
// NextRetry advances the retry counter appropriately and returns true if the
|
||||
// request should be retried, otherwise it returns false if:
|
||||
// IsNextRetry advances the retry counter appropriately
|
||||
// and returns true if the request should be retried,
|
||||
// otherwise it returns false, if:
|
||||
// - we have already reached the maximum retry threshold.
|
||||
// - the error does not fall into the retryable category.
|
||||
// - the server has not sent us a 429, or 5xx status code and the
|
||||
// 'Retry-After' response header is not set with a value.
|
||||
// - we need to seek to the beginning of the request body before we
|
||||
// initiate the next retry, the function should log an error and
|
||||
// return false if it fails to do so.
|
||||
//
|
||||
// if retry is set to true, retryAfter will contain the information
|
||||
// regarding the next retry.
|
||||
//
|
||||
// request: the original request sent to the server
|
||||
// restReq: the associated rest.Request
|
||||
// httpReq: the HTTP Request sent to the server
|
||||
// resp: the response sent from the server, it is set if err is nil
|
||||
// err: the server sent this error to us, if err is set then resp is nil.
|
||||
// f: a IsRetryableErrorFunc function provided by the client that determines
|
||||
// if the err sent by the server is retryable.
|
||||
NextRetry(req *http.Request, resp *http.Response, err error, f IsRetryableErrorFunc) (*RetryAfter, bool)
|
||||
IsNextRetry(ctx context.Context, restReq *Request, httpReq *http.Request, resp *http.Response, err error, f IsRetryableErrorFunc) bool
|
||||
|
||||
// BeforeNextRetry is responsible for carrying out operations that need
|
||||
// to be completed before the next retry is initiated:
|
||||
// - if the request context is already canceled there is no need to
|
||||
// retry, the function will return ctx.Err().
|
||||
// - we need to seek to the beginning of the request body before we
|
||||
// initiate the next retry, the function should return an error if
|
||||
// it fails to do so.
|
||||
// - we should wait the number of seconds the server has asked us to
|
||||
// in the 'Retry-After' response header.
|
||||
// Before should be invoked prior to each attempt, including
|
||||
// the first one. If an error is returned, the request should
|
||||
// be aborted immediately.
|
||||
//
|
||||
// If BeforeNextRetry returns an error the client should abort the retry,
|
||||
// otherwise it is safe to initiate the next retry.
|
||||
BeforeNextRetry(ctx context.Context, backoff BackoffManager, retryAfter *RetryAfter, url string, body io.Reader) error
|
||||
// Before may also be additionally responsible for preparing
|
||||
// the request for the next retry, namely in terms of resetting
|
||||
// the request body in case it has been read.
|
||||
Before(ctx context.Context, r *Request) error
|
||||
|
||||
// After should be invoked immediately after an attempt is made.
|
||||
After(ctx context.Context, r *Request, resp *http.Response, err error)
|
||||
|
||||
// WrapPreviousError wraps the error from any previous attempt into
|
||||
// the final error specified in 'finalErr', so the user has more
|
||||
// context why the request failed.
|
||||
// For example, if a request times out after multiple retries then
|
||||
// we see a generic context.Canceled or context.DeadlineExceeded
|
||||
// error which is not very useful in debugging. This function can
|
||||
// wrap any error from previous attempt(s) to provide more context to
|
||||
// the user. The error returned in 'err' must satisfy the
|
||||
// following conditions:
|
||||
// a: errors.Unwrap(err) = errors.Unwrap(finalErr) if finalErr
|
||||
// implements Unwrap
|
||||
// b: errors.Unwrap(err) = finalErr if finalErr does not
|
||||
// implements Unwrap
|
||||
// c: errors.Is(err, otherErr) = errors.Is(finalErr, otherErr)
|
||||
WrapPreviousError(finalErr error) (err error)
|
||||
}
|
||||
|
||||
// RetryAfter holds information associated with the next retry.
|
||||
@ -107,37 +118,58 @@ type RetryAfter struct {
|
||||
type withRetry struct {
|
||||
maxRetries int
|
||||
attempts int
|
||||
|
||||
// retry after parameters that pertain to the attempt that is to
|
||||
// be made soon, so as to enable 'Before' and 'After' to refer
|
||||
// to the retry parameters.
|
||||
// - for the first attempt, it will always be nil
|
||||
// - for consecutive attempts, it is non nil and holds the
|
||||
// retry after parameters for the next attempt to be made.
|
||||
retryAfter *RetryAfter
|
||||
|
||||
// we keep track of two most recent errors, if the most
|
||||
// recent attempt is labeled as 'N' then:
|
||||
// - currentErr represents the error returned by attempt N, it
|
||||
// can be nil if attempt N did not return an error.
|
||||
// - previousErr represents an error from an attempt 'M' which
|
||||
// precedes attempt 'N' (N - M >= 1), it is non nil only when:
|
||||
// - for a sequence of attempt(s) 1..n (n>1), there
|
||||
// is an attempt k (k<n) that returned an error.
|
||||
previousErr, currentErr error
|
||||
}
|
||||
|
||||
func (r *withRetry) SetMaxRetries(maxRetries int) {
|
||||
if maxRetries < 0 {
|
||||
maxRetries = 0
|
||||
func (r *withRetry) trackPreviousError(err error) {
|
||||
// keep track of two most recent errors
|
||||
if r.currentErr != nil {
|
||||
r.previousErr = r.currentErr
|
||||
}
|
||||
r.maxRetries = maxRetries
|
||||
r.currentErr = err
|
||||
}
|
||||
|
||||
func (r *withRetry) NextRetry(req *http.Request, resp *http.Response, err error, f IsRetryableErrorFunc) (*RetryAfter, bool) {
|
||||
if req == nil || (resp == nil && err == nil) {
|
||||
func (r *withRetry) IsNextRetry(ctx context.Context, restReq *Request, httpReq *http.Request, resp *http.Response, err error, f IsRetryableErrorFunc) bool {
|
||||
defer r.trackPreviousError(err)
|
||||
|
||||
if httpReq == nil || (resp == nil && err == nil) {
|
||||
// bad input, we do nothing.
|
||||
return nil, false
|
||||
return false
|
||||
}
|
||||
|
||||
r.attempts++
|
||||
retryAfter := &RetryAfter{Attempt: r.attempts}
|
||||
r.retryAfter = &RetryAfter{Attempt: r.attempts}
|
||||
if r.attempts > r.maxRetries {
|
||||
return retryAfter, false
|
||||
return false
|
||||
}
|
||||
|
||||
// if the server returned an error, it takes precedence over the http response.
|
||||
var errIsRetryable bool
|
||||
if f != nil && err != nil && f.IsErrorRetryable(req, err) {
|
||||
if f != nil && err != nil && f.IsErrorRetryable(httpReq, err) {
|
||||
errIsRetryable = true
|
||||
// we have a retryable error, for which we will create an
|
||||
// artificial "Retry-After" response.
|
||||
resp = retryAfterResponse()
|
||||
}
|
||||
if err != nil && !errIsRetryable {
|
||||
return retryAfter, false
|
||||
return false
|
||||
}
|
||||
|
||||
// if we are here, we have either a or b:
|
||||
@ -147,34 +179,126 @@ func (r *withRetry) NextRetry(req *http.Request, resp *http.Response, err error,
|
||||
// need to check if it is retryable
|
||||
seconds, wait := checkWait(resp)
|
||||
if !wait {
|
||||
return retryAfter, false
|
||||
return false
|
||||
}
|
||||
|
||||
retryAfter.Wait = time.Duration(seconds) * time.Second
|
||||
retryAfter.Reason = getRetryReason(r.attempts, seconds, resp, err)
|
||||
return retryAfter, true
|
||||
r.retryAfter.Wait = time.Duration(seconds) * time.Second
|
||||
r.retryAfter.Reason = getRetryReason(r.attempts, seconds, resp, err)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *withRetry) BeforeNextRetry(ctx context.Context, backoff BackoffManager, retryAfter *RetryAfter, url string, body io.Reader) error {
|
||||
// Ensure the response body is fully read and closed before
|
||||
// we reconnect, so that we reuse the same TCP connection.
|
||||
func (r *withRetry) Before(ctx context.Context, request *Request) error {
|
||||
// If the request context is already canceled there
|
||||
// is no need to retry.
|
||||
if ctx.Err() != nil {
|
||||
r.trackPreviousError(ctx.Err())
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
if seeker, ok := body.(io.Seeker); ok && body != nil {
|
||||
if _, err := seeker.Seek(0, 0); err != nil {
|
||||
return fmt.Errorf("can't Seek() back to beginning of body for %T", r)
|
||||
url := request.URL()
|
||||
// r.retryAfter represents the retry after parameters calculated
|
||||
// from the (response, err) tuple from the last attempt, so 'Before'
|
||||
// can apply these retry after parameters prior to the next attempt.
|
||||
// 'r.retryAfter == nil' indicates that this is the very first attempt.
|
||||
if r.retryAfter == nil {
|
||||
// we do a backoff sleep before the first attempt is made,
|
||||
// (preserving current behavior).
|
||||
request.backoff.Sleep(request.backoff.CalculateBackoff(url))
|
||||
return nil
|
||||
}
|
||||
|
||||
// At this point we've made atleast one attempt, post which the response
|
||||
// body should have been fully read and closed in order for it to be safe
|
||||
// to reset the request body before we reconnect, in order for us to reuse
|
||||
// the same TCP connection.
|
||||
if seeker, ok := request.body.(io.Seeker); ok && request.body != nil {
|
||||
if _, err := seeker.Seek(0, io.SeekStart); err != nil {
|
||||
err = fmt.Errorf("failed to reset the request body while retrying a request: %v", err)
|
||||
r.trackPreviousError(err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
klog.V(4).Infof("Got a Retry-After %s response for attempt %d to %v", retryAfter.Wait, retryAfter.Attempt, url)
|
||||
if backoff != nil {
|
||||
backoff.Sleep(retryAfter.Wait)
|
||||
// if we are here, we have made attempt(s) al least once before.
|
||||
if request.backoff != nil {
|
||||
// TODO(tkashem) with default set to use exponential backoff
|
||||
// we can merge these two sleeps:
|
||||
// BackOffManager.Sleep(max(backoffManager.CalculateBackoff(), retryAfter))
|
||||
// see https://github.com/kubernetes/kubernetes/issues/108302
|
||||
request.backoff.Sleep(r.retryAfter.Wait)
|
||||
request.backoff.Sleep(request.backoff.CalculateBackoff(url))
|
||||
}
|
||||
|
||||
// We are retrying the request that we already send to
|
||||
// apiserver at least once before. This request should
|
||||
// also be throttled with the client-internal rate limiter.
|
||||
if err := request.tryThrottleWithInfo(ctx, r.retryAfter.Reason); err != nil {
|
||||
r.trackPreviousError(ctx.Err())
|
||||
return err
|
||||
}
|
||||
|
||||
klog.V(4).Infof("Got a Retry-After %s response for attempt %d to %v", r.retryAfter.Wait, r.retryAfter.Attempt, request.URL().String())
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *withRetry) After(ctx context.Context, request *Request, resp *http.Response, err error) {
|
||||
// 'After' is invoked immediately after an attempt is made, let's label
|
||||
// the attempt we have just made as attempt 'N'.
|
||||
// the current value of r.retryAfter represents the retry after
|
||||
// parameters calculated from the (response, err) tuple from
|
||||
// attempt N-1, so r.retryAfter is outdated and should not be
|
||||
// referred to here.
|
||||
r.retryAfter = nil
|
||||
|
||||
if request.c.base != nil {
|
||||
if err != nil {
|
||||
request.backoff.UpdateBackoff(request.URL(), err, 0)
|
||||
} else {
|
||||
request.backoff.UpdateBackoff(request.URL(), err, resp.StatusCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *withRetry) WrapPreviousError(currentErr error) error {
|
||||
if currentErr == nil || r.previousErr == nil {
|
||||
return currentErr
|
||||
}
|
||||
|
||||
// if both previous and current error objects represent the error,
|
||||
// then there is no need to wrap the previous error.
|
||||
if currentErr.Error() == r.previousErr.Error() {
|
||||
return currentErr
|
||||
}
|
||||
|
||||
previousErr := r.previousErr
|
||||
// net/http wraps the underlying error with an url.Error, if the
|
||||
// previous err object is an instance of url.Error, then we can
|
||||
// unwrap it to get to the inner error object, this is so we can
|
||||
// avoid error message like:
|
||||
// Error: Get "http://foo.bar/api/v1": context deadline exceeded - error \
|
||||
// from a previous attempt: Error: Get "http://foo.bar/api/v1": EOF
|
||||
if urlErr, ok := r.previousErr.(*url.Error); ok && urlErr != nil {
|
||||
if urlErr.Unwrap() != nil {
|
||||
previousErr = urlErr.Unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
return &wrapPreviousError{
|
||||
currentErr: currentErr,
|
||||
previousError: previousErr,
|
||||
}
|
||||
}
|
||||
|
||||
type wrapPreviousError struct {
|
||||
currentErr, previousError error
|
||||
}
|
||||
|
||||
func (w *wrapPreviousError) Unwrap() error { return w.currentErr }
|
||||
func (w *wrapPreviousError) Error() string {
|
||||
return fmt.Sprintf("%s - error from a previous attempt: %s", w.currentErr.Error(), w.previousError.Error())
|
||||
}
|
||||
|
||||
// checkWait returns true along with a number of seconds if
|
||||
// the server instructed us to wait before retrying.
|
||||
func checkWait(resp *http.Response) (int, bool) {
|
||||
|
Reference in New Issue
Block a user