chore: close http responses in a way to allow the Transport to re-use…

… the TCP connection

cmd/devtool/commands/event.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/urfave/cli/v2"
 )
 
@@ -88,11 +89,11 @@ func EventSend(c *cli.Context) error {
 		if err != nil {
 			return err
 		}
+		defer func() { httputil.CloseResponse(resp) }()
 		b, err := io.ReadAll(resp.Body)
 		if err != nil {
 			return err
 		}
-		defer resp.Body.Close()
 
 		if resp.StatusCode != http.StatusOK {
 			fmt.Printf("%s\n%s\n", resp.Status, b)

config/backend-config/namespace_config.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/rudderlabs/rudder-server/config"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
@@ -138,13 +139,12 @@ func (nc *namespaceConfig) makeHTTPRequest(ctx context.Context, url string) ([]b
 		return nil, err
 	}
 
+	defer func() { httputil.CloseResponse(resp) }()
 	respBody, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
 	}
 
-	defer func() { _ = resp.Body.Close() }()
-
 	if resp.StatusCode >= 300 {
 		return nil, getNotOKError(respBody, resp.StatusCode)
 	}

config/backend-config/single_workspace.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/rudderlabs/rudder-server/config"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
@@ -150,13 +151,12 @@ func (wc *singleWorkspaceConfig) makeHTTPRequest(ctx context.Context, url string
 		return nil, err
 	}
 
+	defer func() { httputil.CloseResponse(resp) }()
 	respBody, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
 	}
 
-	defer func() { _ = resp.Body.Close() }()
-
 	if resp.StatusCode >= 300 {
 		return nil, getNotOKError(respBody, resp.StatusCode)
 	}

enterprise/reporting/reporting.go

@@ -19,6 +19,7 @@ import (
 	backendconfig "github.com/rudderlabs/rudder-server/config/backend-config"
 	migrator "github.com/rudderlabs/rudder-server/services/sql-migrator"
 	"github.com/rudderlabs/rudder-server/services/stats"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types"
@@ -413,7 +414,7 @@ func (handle *HandleT) sendMetric(ctx context.Context, netClient *http.Client, c
 		httpStatTags["status"] = strconv.Itoa(resp.StatusCode)
 		stats.Default.NewTaggedStat(STAT_REPORTING_HTTP_REQ, stats.CountType, httpStatTags).Count(1)
 
-		defer resp.Body.Close()
+		defer func() { httputil.CloseResponse(resp) }()
 		respBody, err := io.ReadAll(resp.Body)
 		if err != nil {
 			handle.log.Error(err.Error())

enterprise/suppress-user/syncer.go

@@ -15,6 +15,7 @@ import (
 	"github.com/rudderlabs/rudder-server/config"
 	"github.com/rudderlabs/rudder-server/enterprise/suppress-user/model"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
@@ -175,12 +176,7 @@ func (s *Syncer) sync(token []byte) ([]model.Suppression, []byte, error) {
 		if err != nil {
 			return err
 		}
-		defer func() {
-			err := resp.Body.Close()
-			if err != nil {
-				s.log.Error(err)
-			}
-		}()
+		defer func() { httputil.CloseResponse(resp) }()
 
 		// If statusCode is not 2xx, then returning empty regulations
 		if resp.StatusCode < 200 || resp.StatusCode >= 300 {

gateway/gateway.go

@@ -1004,7 +1004,7 @@ func (gateway *HandleT) getWarehousePending(payload []byte) bool {
 		return false
 	}
 
-	defer resp.Body.Close()
+	defer func() { rs_httputil.CloseResponse(resp) }()
 
 	var whPendingResponse warehouseutils.PendingEventsResponseT
 	respData, err := io.ReadAll(resp.Body)

gateway/integration_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/rudderlabs/rudder-server/testhelper/health"
 	"github.com/rudderlabs/rudder-server/testhelper/rand"
 	whUtil "github.com/rudderlabs/rudder-server/testhelper/webhook"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 )
 
 func TestGatewayIntegration(t *testing.T) {
@@ -183,7 +184,7 @@ func testGatewayByAppType(t *testing.T, appType string) {
 	resp, err := http.Get(healthEndpoint)
 	require.ErrorContains(t, err, "connection refused")
 	require.Nil(t, resp)
-	defer resp.Body.Close()
+	defer func() { httputil.CloseResponse(resp) }()
 
 	// Checking now that the configuration has been processed and the server can start
 	t.Log("Checking health endpoint at", healthEndpoint)
@@ -316,7 +317,7 @@ func sendEvent(t *testing.T, httpPort int, payload *strings.Reader, callType, wr
 
 	res, err := httpClient.Do(req)
 	require.NoError(t, err)
-	defer func() { _ = res.Body.Close() }()
+	defer func() { httputil.CloseResponse(res) }()
 
 	body, err := io.ReadAll(res.Body)
 	require.NoError(t, err)

gateway/webhook/webhookTransformer.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	"github.com/rudderlabs/rudder-server/gateway/response"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
@@ -57,7 +58,7 @@ func (bt *batchWebhookTransformerT) transform(events [][]byte, sourceType string
 	}
 
 	respBody, err := io.ReadAll(resp.Body)
-	_ = resp.Body.Close()
+	func() { httputil.CloseResponse(resp) }()
 
 	if err != nil {
 		bt.stats.failedStat.Count(len(events))

integration_test/docker_test/docker_test.go

@@ -40,6 +40,7 @@ import (
 	"github.com/rudderlabs/rudder-server/testhelper/health"
 	"github.com/rudderlabs/rudder-server/testhelper/rand"
 	whUtil "github.com/rudderlabs/rudder-server/testhelper/webhook"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 )
@@ -655,7 +656,7 @@ func getEvent(url, method string) (string, error) {
 	if err != nil {
 		return "", err
 	}
-	defer func() { _ = res.Body.Close() }()
+	defer func() { httputil.CloseResponse(res) }()
 
 	body, err := io.ReadAll(res.Body)
 	if err != nil {
@@ -719,7 +720,7 @@ func sendEvent(t *testing.T, payload *strings.Reader, callType, writeKey string)
 		t.Logf("sendEvent error: %v", err)
 		return
 	}
-	defer func() { _ = res.Body.Close() }()
+	defer func() { httputil.CloseResponse(res) }()
 
 	body, err := io.ReadAll(res.Body)
 	if err != nil {

integration_test/multi_tentant_test/multi_tenant_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/rudderlabs/rudder-server/testhelper/health"
 	"github.com/rudderlabs/rudder-server/testhelper/rand"
 	whUtil "github.com/rudderlabs/rudder-server/testhelper/webhook"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 )
 
@@ -215,7 +216,7 @@ func testMultiTenantByAppType(t *testing.T, appType string) {
 	require.ErrorContains(t, err, "connection refused")
 	require.Nil(t, resp)
 	if err == nil {
-		defer func() { _ = resp.Body.Close() }()
+		defer func() { httputil.CloseResponse(resp) }()
 	}
 
 	// Pushing valid configuration via ETCD
@@ -415,7 +416,7 @@ func sendEvent(t *testing.T, httpPort int, payload *strings.Reader, callType, wr
 
 	res, err := httpClient.Do(req)
 	require.NoError(t, err)
-	defer func() { _ = res.Body.Close() }()
+	defer func() { httputil.CloseResponse(res) }()
 
 	body, err := io.ReadAll(res.Body)
 	require.NoError(t, err)

processor/processor.go

@@ -41,6 +41,7 @@ import (
 	"github.com/rudderlabs/rudder-server/services/stats"
 	"github.com/rudderlabs/rudder-server/services/transientsource"
 	"github.com/rudderlabs/rudder-server/utils/bytesize"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types"
@@ -562,7 +563,7 @@ func (proc *HandleT) makeFeaturesFetchCall() bool {
 		return true
 	}
 
-	defer func() { _ = res.Body.Close() }()
+	defer func() { httputil.CloseResponse(res) }()
 	body, err := io.ReadAll(res.Body)
 	if err != nil {
 		return true

processor/transformer/transformer.go

@@ -20,6 +20,7 @@ import (
 	backendconfig "github.com/rudderlabs/rudder-server/config/backend-config"
 	"github.com/rudderlabs/rudder-server/processor/integrations"
 	"github.com/rudderlabs/rudder-server/services/stats"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
@@ -178,7 +179,7 @@ func GetVersion() (transformerBuildVersion string) {
 		transformerBuildVersion = fmt.Sprintf("No response from transformer. %s", transformerBuildVersion)
 		return
 	}
-	defer func() { _ = resp.Body.Close() }()
+	defer func() { httputil.CloseResponse(resp) }()
 	if resp.StatusCode == http.StatusOK {
 		bodyBytes, err := io.ReadAll(resp.Body)
 		if err != nil {
@@ -380,8 +381,8 @@ func (trans *HandleT) doPost(ctx context.Context, rawJSON []byte, url string, ta
 			if reqErr != nil {
 				return reqErr
 			}
+			defer func() { httputil.CloseResponse(resp) }()
 			respData, reqErr = io.ReadAll(resp.Body)
-			_ = resp.Body.Close()
 			return reqErr
 		},
 		backoff.WithMaxRetries(backoff.NewExponentialBackOff(), uint64(maxRetry)),

regulation-worker/cmd/main_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/filemanager"
 	"github.com/rudderlabs/rudder-server/services/kvstoremanager"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 )
 
 var (
@@ -127,7 +128,7 @@ func TestFlow(t *testing.T) {
 		if err != nil {
 			return err
 		}
-		defer resp.Body.Close()
+		defer func() { httputil.CloseResponse(resp) }()
 		if resp.StatusCode != http.StatusOK {
 			return fmt.Errorf("status code not OK")
 		}

regulation-worker/internal/client/client.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/stats"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 )
 
@@ -51,12 +52,7 @@ func (j *JobAPI) Get(ctx context.Context) (model.Job, error) {
 	if err != nil {
 		return model.Job{}, err
 	}
-	defer func() {
-		err := resp.Body.Close()
-		if err != nil {
-			pkgLogger.Errorf("error while closing response body: %v", err)
-		}
-	}()
+	defer func() { httputil.CloseResponse(resp) }()
 	pkgLogger.Debugf("obtained response code: %v", resp.StatusCode, "response body: ", resp.Body)
 
 	// if successful
@@ -140,7 +136,7 @@ func (j *JobAPI) UpdateStatus(ctx context.Context, status model.JobStatus, jobID
 	if err != nil {
 		return err
 	}
-	defer func() { _ = resp.Body.Close() }()
+	defer func() { httputil.CloseResponse(resp) }()
 
 	pkgLogger.Debugf("response code: %v", resp.StatusCode, "response body: %v", resp.Body)
 	if resp.StatusCode >= 200 && resp.StatusCode < 300 {

regulation-worker/internal/delete/api/api.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/rudderlabs/rudder-server/regulation-worker/internal/model"
 	"github.com/rudderlabs/rudder-server/services/stats"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 )
 
@@ -65,7 +66,7 @@ func (api *APIManager) Delete(ctx context.Context, job model.Job, destConfig map
 		}
 		return model.JobStatusFailed
 	}
-	defer resp.Body.Close()
+	defer func() { httputil.CloseResponse(resp) }()
 	bodyBytes, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return model.JobStatusFailed

router/batchrouter/batchrouter.go

@@ -47,6 +47,7 @@ import (
 	"github.com/rudderlabs/rudder-server/services/filemanager"
 	"github.com/rudderlabs/rudder-server/services/stats"
 	"github.com/rudderlabs/rudder-server/utils/bytesize"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types"
@@ -1132,10 +1133,9 @@ func (brt *HandleT) postToWarehouse(batchJobs *BatchJobsT, output StorageUploadO
 		brt.logger.Errorf("BRT: Failed to route staging file URL to warehouse service@%v, error:%v", uri, err)
 		return
 	}
-	defer func() { _ = resp.Body.Close() }()
+	defer func() { httputil.CloseResponse(resp) }()
 
 	if resp.StatusCode == http.StatusOK {
-		_, err = io.Copy(io.Discard, resp.Body)
 		brt.logger.Infof("BRT: Routed successfully staging file URL to warehouse service@%v", uri)
 	} else {
 		body, _ := io.ReadAll(resp.Body)

router/eventorder_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/rudderlabs/rudder-server/testhelper/destination"
 	trand "github.com/rudderlabs/rudder-server/testhelper/rand"
 	"github.com/rudderlabs/rudder-server/testhelper/workspaceConfig"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 	"github.com/stretchr/testify/require"
 	"github.com/tidwall/gjson"
@@ -210,7 +211,7 @@ func TestEventOrderGuarantee(t *testing.T) {
 					resp, err := client.Do(req)
 					require.NoError(t, err, "should be able to send the request to gateway")
 					require.Equal(t, http.StatusOK, resp.StatusCode, "should be able to send the request to gateway successfully", payload)
-					resp.Body.Close()
+					func() { httputil.CloseResponse(resp) }()
 				}
 			}()
 

router/network.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/rudderlabs/rudder-server/processor/integrations"
 	"github.com/rudderlabs/rudder-server/router/utils"
+	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/logger"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/sysUtils"
@@ -168,7 +169,7 @@ func (network *NetHandleT) SendPost(ctx context.Context, structData integrations
 			}
 		}
 
-		defer func() { _ = resp.Body.Close() }()
+		defer func() { httputil.CloseResponse(resp) }()
 
 		respBody, err := io.ReadAll(resp.Body)
 		if err != nil {