WIP - POC - Read and close response body before seeking to retry

[liggitt]

What type of PR is this?

/kind bug
/kind failing-test

Which issue(s) this PR fixes:

xref #108906

Fixes a bug in our client retry logic that was not draining/closing the body prior to trying to reset the request body for a retry. According to go upstream, this is required to avoid races.

WIP because one place we call IsNextRetry currently depends on being able to read from the response body after trying the Seek(0,0) call. Not sure what to do about that.

NONE

/assign @smarterclayton

[liggitt]

this is the place I don't see a simple way to 1) drain the response body before Seeking to retry, 2) only call fn once, and 3) still retry if the seek fails

[liggitt]

/approve cancel

[k8s-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
To complete the pull request process, please ask for approval from smarterclayton after the PR has been reviewed.

The full list of commands accepted by this bot can be found here.

Needs approval from an approver in each of these files:

• staging/src/k8s.io/client-go/OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[aojea]

don't we have to close it always that we call r.prepareForNextRetry?

[aojea]

maybe we can unconditionally close here

[liggitt]

there are several paths in prepareForNextRetry that don't

[aojea]

are you sure? I see that at this point we don't have to do anything anymore with the response

	// if we are here, we have either a or b:
	//  a: we have a retryable error, for which we already
	//     have an artificial "Retry-After" response.
	//  b: we have a response from the server for which we
	//     need to check if it is retryable
	seconds, wait := checkWait(resp)
	if !wait {
		return false
	}

we are after the wait

[aojea]

I think that we can close it multiple times, we are not doing this in parallel, no?

[aojea]

if we can do that , is a matter of Close() it on the right places, like in

func (r *Request) transformResponse(resp *http.Response, req *http.Request) Result {

that will solve https://github.com/kubernetes/kubernetes/pull/109028/files#r835508597

 func (r *Request) transformResponse(resp *http.Response, req *http.Request) Result {
        var body []byte
        if resp.Body != nil {
+               defer resp.Body.Close()
                data, err := ioutil.ReadAll(resp.Body)
                switch err.(type) {
                case nil:
diff --git a/staging/src/k8s.io/client-go/rest/with_retry.go b/staging/src/k8s.io/client-go/rest/with_retry.go
index 11b9b5225b1..4469f5a1b9b 100644
--- a/staging/src/k8s.io/client-go/rest/with_retry.go
+++ b/staging/src/k8s.io/client-go/rest/with_retry.go
@@ -157,7 +157,7 @@ func (r *withRetry) IsNextRetry(ctx context.Context, restReq *Request, httpReq *
        r.retryAfter.Wait = time.Duration(seconds) * time.Second
        r.retryAfter.Reason = getRetryReason(r.attempts, seconds, resp, err)
 
-       if err := r.prepareForNextRetry(ctx, restReq); err != nil {
+       if err := r.prepareForNextRetry(ctx, restReq, resp); err != nil {
                klog.V(4).Infof("Could not retry request - %v", err)
                return false
        }
@@ -172,7 +172,7 @@ func (r *withRetry) IsNextRetry(ctx context.Context, restReq *Request, httpReq *
 // - 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.
-func (r *withRetry) prepareForNextRetry(ctx context.Context, request *Request) error {
+func (r *withRetry) prepareForNextRetry(ctx context.Context, request *Request, response *http.Response) error {
        if ctx.Err() != nil {
                return ctx.Err()
        }
@@ -180,6 +180,7 @@ func (r *withRetry) prepareForNextRetry(ctx context.Context, request *Request) e
        // Ensure the response body is fully read and closed before
        // we reconnect, so that we reuse the same TCP connection.
        if seeker, ok := request.body.(io.Seeker); ok && request.body != nil {
+               readAndCloseResponseBody(response)
                if _, err := seeker.Seek(0, 0); err != nil {
                        return fmt.Errorf("can't Seek() back to beginning of body for %T", request)
                }

[liggitt]

I actually don't have bandwidth to finish this fix out if someone else does and can pick it up (or come up with a different approach)

[dims]

cc @MadhavJivrajani

[cici37]

/triage accepted