diff --git a/base/utils/http.go b/base/utils/http.go index a979f1ef8..c4eafff11 100644 --- a/base/utils/http.go +++ b/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) diff --git a/base/utils/http_test.go b/base/utils/http_test.go index e0b5de794..a659ef0f5 100644 --- a/base/utils/http_test.go +++ b/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()) +} diff --git a/listener/upload.go b/listener/upload.go index eee481220..a626b3e6f 100644 --- a/listener/upload.go +++ b/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{}) {