RHINENG-30154: close HTTP response bodies after retries

base/utils/http.go

@@ -3,6 +3,7 @@ package utils
 import (
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httputil"
 	"time"
@@ -24,29 +25,44 @@ func HTTPCallRetry(httpCallFun func() (outputDataPtr interface{}, resp *http.Res
 		attempt++
 		outDataPtr, resp, callErr := httpCallFun()
 		if statusCodeFound(resp, codesToRetry) {
+			closeHTTPResponse(resp)
 			LogWarn("attempt", attempt, "status_code", TryGetStatusCode(resp),
 				"HTTP call ended with wrong status code")
 			continue
 		}
 
 		if callErr == nil {
+			closeHTTPResponse(resp)
 			return outDataPtr, nil
 		}
 
 		if len(codesToRetry) == 0 { // no "retry" codes specified, continue always
+			closeHTTPResponse(resp)
 			LogWarn("attempt", attempt, "err", callErr, "HTTP call failed, trying again")
 			continue
 		}
 
 		responseDetails := tryGetResponseDetails(resp)
-		if resp != nil {
-			resp.Body.Close()
-		}
+		closeHTTPResponse(resp)
 		return nil, errors.Wrap(callErr, "HTTP call failed"+responseDetails)
 	}
 	return nil, errors.Errorf("HTTP retry call failed, attempts: %d", attempt)
 }
 
+// maxHTTPResponseDrain matches net/http maxPostHandlerReadBytes: enough leftover
+// body to allow connection reuse without reading arbitrarily large payloads.
+const maxHTTPResponseDrain = 256 << 10
+
+// closeHTTPResponse drains up to maxHTTPResponseDrain and closes the body so the
+// Transport can release the TCP connection.
+func closeHTTPResponse(resp *http.Response) {
+	if resp == nil || resp.Body == nil {
+		return
+	}
+	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxHTTPResponseDrain))
+	_ = resp.Body.Close()
+}
+
 func CallAPI(client *http.Client, request *http.Request, debugEnabled bool) (*http.Response, error) {
 	if debugEnabled {
 		dump, err := httputil.DumpRequestOut(request, true)

base/utils/http_test.go

@@ -2,12 +2,56 @@ package utils
 
 import (
 	"errors"
+	"io"
 	"net/http"
+	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
+type closeTrackingBody struct {
+	rc        io.ReadCloser
+	readBytes atomic.Int64
+	closed    atomic.Bool
+}
+
+func newCloseTrackingBody(content string) *closeTrackingBody {
+	return &closeTrackingBody{rc: io.NopCloser(strings.NewReader(content))}
+}
+
+func (b *closeTrackingBody) Read(p []byte) (int, error) {
+	n, err := b.rc.Read(p)
+	if n > 0 {
+		b.readBytes.Add(int64(n))
+	}
+	return n, err
+}
+
+func (b *closeTrackingBody) Close() error {
+	b.closed.Store(true)
+	return b.rc.Close()
+}
+
+func trackingResponse(status int, body string) (*http.Response, *closeTrackingBody) {
+	tb := newCloseTrackingBody(body)
+	return &http.Response{
+		StatusCode: status,
+		Body:       tb,
+	}, tb
+}
+
+func assertBodyDrainedAndClosed(t *testing.T, body *closeTrackingBody, content string) {
+	t.Helper()
+	assert.True(t, body.closed.Load())
+	expected := int64(len(content))
+	if expected > maxHTTPResponseDrain {
+		expected = maxHTTPResponseDrain
+	}
+	assert.Equal(t, expected, body.readBytes.Load())
+}
+
 func createFirstlyFailingCallFun() func() (interface{}, *http.Response, error) {
 	i := 0
 	httpCallFun := func() (interface{}, *http.Response, error) {
@@ -36,3 +80,62 @@ func TestHTTPCallRetryFail(t *testing.T) {
 	assert.Equal(t, "HTTP retry call failed, attempts: 2", err.Error())
 	assert.Nil(t, data)
 }
+
+func TestHTTPCallRetryClosesBodyOnSuccess(t *testing.T) {
+	const content = "ok"
+	resp, body := trackingResponse(http.StatusOK, content)
+	data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) {
+		return "data", resp, nil
+	}, false, 1)
+
+	assert.NoError(t, err)
+	assert.Equal(t, "data", data)
+	assertBodyDrainedAndClosed(t, body, content)
+}
+
+func TestHTTPCallRetryClosesBodyOnRetryableStatus(t *testing.T) {
+	const firstContent = "retry"
+	const secondContent = "ok"
+	first, firstBody := trackingResponse(http.StatusServiceUnavailable, firstContent)
+	second, secondBody := trackingResponse(http.StatusOK, secondContent)
+	attempt := 0
+	data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) {
+		attempt++
+		if attempt == 1 {
+			return nil, first, nil
+		}
+		return "data", second, nil
+	}, false, 3, http.StatusServiceUnavailable)
+
+	assert.NoError(t, err)
+	assert.Equal(t, "data", data)
+	assertBodyDrainedAndClosed(t, firstBody, firstContent)
+	assertBodyDrainedAndClosed(t, secondBody, secondContent)
+}
+
+func TestHTTPCallRetryClosesBodyOnNonRetryableError(t *testing.T) {
+	const content = "nope"
+	resp, body := trackingResponse(http.StatusBadRequest, content)
+	data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) {
+		return nil, resp, errors.New("bad request")
+	}, false, 3, http.StatusServiceUnavailable)
+
+	assert.Error(t, err)
+	assert.Nil(t, data)
+	assertBodyDrainedAndClosed(t, body, content)
+}
+
+func TestCloseHTTPResponseDrainsBody(t *testing.T) {
+	const content = "some body data"
+	body := newCloseTrackingBody(content)
+	closeHTTPResponse(&http.Response{StatusCode: http.StatusOK, Body: body})
+	assertBodyDrainedAndClosed(t, body, content)
+}
+
+func TestCloseHTTPResponseLimitsDrain(t *testing.T) {
+	content := strings.Repeat("x", maxHTTPResponseDrain+1024)
+	body := newCloseTrackingBody(content)
+	closeHTTPResponse(&http.Response{StatusCode: http.StatusOK, Body: body})
+	assert.True(t, body.closed.Load())
+	assert.Equal(t, int64(maxHTTPResponseDrain), body.readBytes.Load())
+}

listener/upload.go

@@ -838,12 +838,16 @@ func getYumUpdates(event HostEvent, client *api.Client) (*YumUpdates, error) {
 
 	if yumUpdatesURL != nil && *yumUpdatesURL != "" {
 		resp, err := client.Request(&base.Context, http.MethodGet, *yumUpdatesURL, nil, &parsed)
+		if resp != nil && resp.Body != nil {
+			defer func() {
+				if closeErr := resp.Body.Close(); closeErr != nil {
+					utils.LogWarn("err", closeErr, "closing yum updates response body")
+				}
+			}()
+		}
 		if err != nil {
 			return nil, errors.Wrap(err, "unable to get yum updates from S3")
 		}
-		if err := resp.Body.Close(); err != nil {
-			return nil, errors.Wrap(err, "response error for yum updates from S3")
-		}
 	}
 
 	if (parsed == vmaas.UpdatesV3Response{}) {