Increase context timeout for bandwidth throttled reader (#12820)

increase default timeout up to one hour for toy setups.

fixes #12812

cmd/bucket-replication.go

@@ -43,6 +43,8 @@ import (
 	iampolicy "github.com/minio/pkg/iam/policy"
 )
 
+const throttleDeadline = 1 * time.Hour
+
 // gets replication config associated to a given bucket name.
 func getReplicationConfig(ctx context.Context, bucketName string) (rc *replication.Config, err error) {
 	if globalIsGateway {
@@ -792,15 +794,21 @@ func replicateObject(ctx context.Context, ri ReplicateObjectInfo, objectAPI Obje
 			Bucket:     objInfo.Bucket,
 			HeaderSize: headerSize,
 		}
-		newCtx, cancel := context.WithTimeout(ctx, globalOperationTimeout.Timeout())
-		defer cancel()
+		newCtx := ctx
+		if globalBucketMonitor.IsThrottled(bucket) {
+			var cancel context.CancelFunc
+			newCtx, cancel = context.WithTimeout(ctx, throttleDeadline)
+			defer cancel()
+		}
 		r := bandwidth.NewMonitoredReader(newCtx, globalBucketMonitor, gr, opts)
 		if len(objInfo.Parts) > 1 {
 			if uploadID, err := replicateObjectWithMultipart(ctx, c, dest.Bucket, object, r, objInfo, putOpts); err != nil {
 				replicationStatus = replication.Failed
 				logger.LogIf(ctx, fmt.Errorf("Unable to replicate for object %s/%s(%s): %s", bucket, objInfo.Name, objInfo.VersionID, err))
 				// block and abort remote upload upon failure.
-				c.AbortMultipartUpload(ctx, dest.Bucket, object, uploadID)
+				if err = c.AbortMultipartUpload(ctx, dest.Bucket, object, uploadID); err != nil {
+					logger.LogIf(ctx, fmt.Errorf("Unable to clean up failed upload %s on remote %s/%s: %w", uploadID, dest.Bucket, object, err))
+				}
 			}
 		} else {
 			if _, err = c.PutObject(ctx, dest.Bucket, object, r, size, "", "", putOpts); err != nil {

cmd/bucket-targets.go

@@ -346,6 +346,7 @@ func (sys *BucketTargetSys) getRemoteTargetClient(tcfg *madmin.BucketTarget) (*T
 	getRemoteTargetInstanceTransportOnce.Do(func() {
 		getRemoteTargetInstanceTransport = NewRemoteTargetHTTPTransport()
 	})
+
 	api, err := minio.New(tcfg.Endpoint, &miniogo.Options{
 		Creds:     creds,
 		Secure:    tcfg.Secure,

cmd/common-main.go

@@ -44,6 +44,7 @@ import (
 	"github.com/minio/console/restapi"
 	"github.com/minio/console/restapi/operations"
 	"github.com/minio/kes"
+	minio "github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/minio-go/v7/pkg/set"
 	"github.com/minio/minio/internal/auth"
@@ -104,6 +105,9 @@ func init() {
 			},
 		},
 	}
+	// Set number of max retries to 1 for minio-go clients
+	minio.MaxRetry = 1
+
 }
 
 const consolePrefix = "CONSOLE_"

internal/bucket/bandwidth/monitor.go

@@ -179,3 +179,11 @@ func (m *Monitor) SetBandwidthLimit(bucket string, limit int64) {
 	t.Limiter = rate.NewLimiter(newlimit, int(t.NodeBandwidthPerSec))
 	m.bucketThrottle[bucket] = t
 }
+
+// IsThrottled returns true if a bucket has bandwidth throttling enabled.
+func (m *Monitor) IsThrottled(bucket string) bool {
+	m.tlock.RLock()
+	defer m.tlock.RUnlock()
+	_, ok := m.bucketThrottle[bucket]
+	return ok
+}