Fix OOM lakectl FS upload bug

cmd/lakectl/cmd/retry_client.go

@@ -18,7 +18,7 @@ var (
 	notTrustedErrorRe = regexp.MustCompile(`certificate is not trusted`)
 )
 
-func NewRetryClient(retriesCfg RetriesCfg, transport *http.Transport) *http.Client {
+func NewRetryClient(retriesCfg RetriesCfg, transport *http.Transport) *retryablehttp.Client {
 	retryClient := retryablehttp.NewClient()
 	if transport != nil {
 		retryClient.HTTPClient.Transport = transport
@@ -28,7 +28,7 @@ func NewRetryClient(retriesCfg RetriesCfg, transport *http.Transport) *http.Clie
 	retryClient.RetryWaitMin = retriesCfg.MinWaitInterval
 	retryClient.RetryWaitMax = retriesCfg.MaxWaitInterval
 	retryClient.CheckRetry = lakectlRetryPolicy
-	return retryClient.StandardClient()
+	return retryClient
 }
 
 // lakectl retry policy - we retry in the following cases:

cmd/lakectl/cmd/root.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/deepmap/oapi-codegen/pkg/securityprovider"
 	"github.com/go-openapi/swag"
+	"github.com/hashicorp/go-retryablehttp"
 	"github.com/mitchellh/go-homedir"
 	"github.com/mitchellh/mapstructure"
 	"github.com/spf13/cobra"
@@ -467,15 +468,15 @@ func sendStats(cmd *cobra.Command, cmdSuffix string) {
 	}
 }
 
-func getHTTPClient() *http.Client {
+func getHTTPClient() *retryablehttp.Client {
 	// Override MaxIdleConnsPerHost to allow highly concurrent access to our API client.
 	// This is done to avoid accumulating many sockets in `TIME_WAIT` status that were closed
 	// only to be immediately reopened.
 	// see: https://stackoverflow.com/a/39834253
 	transport := http.DefaultTransport.(*http.Transport).Clone()
 	transport.MaxIdleConnsPerHost = DefaultMaxIdleConnsPerHost
 	if !cfg.Server.Retries.Enabled {
-		return &http.Client{Transport: transport}
+		return NewRetryClient(RetriesCfg{MaxAttempts: 1}, transport)
 	}
 	return NewRetryClient(cfg.Server.Retries, transport)
 }
@@ -498,7 +499,7 @@ func getClient() *apigen.ClientWithResponses {
 	oss := osinfo.GetOSInfo()
 	client, err := apigen.NewClientWithResponses(
 		serverEndpoint,
-		apigen.WithHTTPClient(httpClient),
+		apigen.WithHTTPClient(httpClient.StandardClient()),
 		apigen.WithRequestEditorFn(basicAuthProvider.Intercept),
 		apigen.WithRequestEditorFn(func(ctx context.Context, req *http.Request) error {
 			// This UA string structure is agreed upon

pkg/api/helpers/upload.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 
 	"github.com/go-openapi/swag"
+	"github.com/hashicorp/go-retryablehttp"
 	"github.com/treeverse/lakefs/pkg/api/apigen"
 	"github.com/treeverse/lakefs/pkg/api/apiutil"
 	"github.com/treeverse/lakefs/pkg/block/azure"
@@ -104,7 +105,7 @@ func ClientUpload(ctx context.Context, client apigen.ClientWithResponsesInterfac
 // It supports both multipart and single part uploads.
 type PreSignUploader struct {
 	Concurrency      int
-	HTTPClient       *http.Client
+	HTTPClient       *retryablehttp.Client
 	Client           apigen.ClientWithResponsesInterface
 	MultipartSupport bool
 }
@@ -124,7 +125,7 @@ type presignUpload struct {
 	numParts    int
 }
 
-func NewPreSignUploader(client apigen.ClientWithResponsesInterface, httpClient *http.Client, multipartSupport bool) *PreSignUploader {
+func NewPreSignUploader(client apigen.ClientWithResponsesInterface, httpClient *retryablehttp.Client, multipartSupport bool) *PreSignUploader {
 	return &PreSignUploader{
 		Concurrency:      DefaultUploadConcurrency,
 		HTTPClient:       httpClient,
@@ -300,7 +301,7 @@ func (u *presignUpload) initMultipart(ctx context.Context) (*apigen.PresignMulti
 }
 
 func (u *presignUpload) uploadPart(ctx context.Context, partReader *io.SectionReader, partURL string) (string, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodPut, partURL, partReader)
+	req, err := retryablehttp.NewRequestWithContext(ctx, http.MethodPut, partURL, partReader)
 	if err != nil {
 		return "", err
 	}
@@ -340,8 +341,7 @@ func (u *presignUpload) uploadObject(ctx context.Context) (*apigen.ObjectStats,
 		// Passing Reader with content length == 0 results in 501 Not Implemented
 		body = u.reader
 	}
-
-	req, err := http.NewRequestWithContext(ctx, http.MethodPut, preSignURL, body)
+	req, err := retryablehttp.NewRequestWithContext(ctx, http.MethodPut, preSignURL, body)
 	if err != nil {
 		return nil, err
 	}
@@ -403,7 +403,7 @@ func (u *presignUpload) Upload(ctx context.Context) (*apigen.ObjectStats, error)
 	return u.uploadObject(ctx)
 }
 
-func ClientUploadPreSign(ctx context.Context, client apigen.ClientWithResponsesInterface, httpClient *http.Client, repoID, branchID, objPath string, metadata map[string]string, contentType string, contents io.ReadSeeker, presignMultipartSupport bool) (*apigen.ObjectStats, error) {
+func ClientUploadPreSign(ctx context.Context, client apigen.ClientWithResponsesInterface, httpClient *retryablehttp.Client, repoID, branchID, objPath string, metadata map[string]string, contentType string, contents io.ReadSeeker, presignMultipartSupport bool) (*apigen.ObjectStats, error) {
 	// upload loop, retry on conflict
 	uploader := NewPreSignUploader(client, httpClient, presignMultipartSupport)
 	for {

pkg/local/sync.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/go-openapi/swag"
+	"github.com/hashicorp/go-retryablehttp"
 	"github.com/jedib0t/go-pretty/v6/progress"
 	"github.com/treeverse/lakefs/pkg/api/apigen"
 	"github.com/treeverse/lakefs/pkg/api/helpers"
@@ -47,13 +48,13 @@ type Tasks struct {
 type SyncManager struct {
 	ctx         context.Context
 	client      *apigen.ClientWithResponses
-	httpClient  *http.Client
+	httpClient  *retryablehttp.Client
 	progressBar *ProgressPool
 	tasks       Tasks
 	cfg         Config
 }
 
-func NewSyncManager(ctx context.Context, client *apigen.ClientWithResponses, httpClient *http.Client, cfg Config) *SyncManager {
+func NewSyncManager(ctx context.Context, client *apigen.ClientWithResponses, httpClient *retryablehttp.Client, cfg Config) *SyncManager {
 	sm := &SyncManager{
 		ctx:         ctx,
 		client:      client,