Use S3 client for uplooads/downloads during perf test (#14570)

cmd/admin-handlers.go

@@ -1004,6 +1004,8 @@ func (a adminAPIHandlers) ObjectSpeedtestHandler(w http.ResponseWriter, r *http.
 		return
 	}
 
+	var bucketExists bool
+
 	sizeStr := r.Form.Get(peerRESTSize)
 	durationStr := r.Form.Get(peerRESTDuration)
 	concurrentStr := r.Form.Get(peerRESTConcurrent)
@@ -1054,13 +1056,25 @@ func (a adminAPIHandlers) ObjectSpeedtestHandler(w http.ResponseWriter, r *http.
 		autotune = false
 	}
 
+	err = objectAPI.MakeBucketWithLocation(ctx, globalObjectPerfBucket, BucketOptions{})
+	if err != nil {
+		if _, ok := err.(BucketExists); !ok {
+			// Only BucketExists error can be ignored.
+			writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
+			return
+		}
+		bucketExists = true
+	}
+
 	deleteBucket := func() {
-		objectAPI.DeleteBucket(context.Background(), pathJoin(minioMetaBucket, "speedtest"), DeleteBucketOptions{
+		objectAPI.DeleteBucket(context.Background(), globalObjectPerfBucket, DeleteBucketOptions{
 			Force:      true,
 			NoRecreate: true,
 		})
 	}
+	if !bucketExists {
 		defer deleteBucket()
+	}
 
 	// Freeze all incoming S3 API calls before running speedtest.
 	globalNotificationSys.ServiceFreeze(ctx, true)

cmd/globals.go

@@ -349,6 +349,8 @@ var (
 	// Used for collecting stats for netperf
 	globalNetPerfMinDuration     = time.Second * 10
 	globalNetPerfRX              netPerfRX
+	globalObjectPerfBucket       = "minio-perf-test-tmp-bucket"
+	globalObjectPerfUserMetadata = "X-Amz-Meta-Minio-Object-Perf" // Clients can set this to bypass S3 API service freeze. Used by object pref tests.
 
 	// Add new variable global values here.
 )

cmd/handler-api.go

@@ -246,12 +246,14 @@ func maxClients(f http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		globalHTTPStats.incS3RequestsIncoming()
 
+		if r.Header.Get(globalObjectPerfUserMetadata) == "" {
 			if val := globalServiceFreeze.Load(); val != nil {
 				if unlock, ok := val.(chan struct{}); ok && unlock != nil {
 					// Wait until unfrozen.
 					<-unlock
 				}
 			}
+		}
 
 		pool, deadline := globalAPIConfig.getRequestsPool()
 		if pool == nil {

cmd/perf-tests.go

@@ -24,6 +24,7 @@ import (
 	"io"
 	"io/ioutil"
 	"math/rand"
+	"net/http"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -31,8 +32,8 @@ import (
 	"github.com/dustin/go-humanize"
 	"github.com/google/uuid"
 	"github.com/minio/madmin-go"
-	"github.com/minio/minio/internal/hash"
+	"github.com/minio/minio-go/v7"
-	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio-go/v7/pkg/credentials"
 	"github.com/minio/pkg/randreader"
 )
 
@@ -61,6 +62,21 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
 	var totalBytesWritten uint64
 	var totalBytesRead uint64
 
+	region := globalSite.Region
+	if region == "" {
+		region = "us-east-1"
+	}
+
+	client, err := minio.New(globalLocalNodeName, &minio.Options{
+		Creds:     credentials.NewStaticV4(globalActiveCred.AccessKey, globalActiveCred.SecretKey, ""),
+		Secure:    globalIsTLS,
+		Transport: globalProxyTransport,
+		Region:    region,
+	})
+	if err != nil {
+		return SpeedtestResult{}, err
+	}
+
 	objCountPerThread := make([]uint64, concurrent)
 
 	uploadsCtx, uploadsCancel := context.WithCancel(context.Background())
@@ -71,15 +87,18 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
 		uploadsCancel()
 	}()
 
-	objNamePrefix := "speedtest/objects/" + uuid.New().String()
+	objNamePrefix := uuid.New().String() + "/"
+
+	userMetadata := make(map[string]string)
+	userMetadata[globalObjectPerfUserMetadata] = "true"
 
 	wg.Add(concurrent)
 	for i := 0; i < concurrent; i++ {
 		go func(i int) {
 			defer wg.Done()
 			for {
-				hashReader, err := hash.NewReader(newRandomReader(size),
+				reader := newRandomReader(size)
-					int64(size), "", "", int64(size))
+				info, err := client.PutObject(uploadsCtx, globalObjectPerfBucket, fmt.Sprintf("%s%d.%d", objNamePrefix, i, objCountPerThread[i]), reader, int64(size), minio.PutObjectOptions{UserMetadata: userMetadata, DisableMultipart: true}) // Bypass S3 API freeze
 				if err != nil {
 					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
 						errOnce.Do(func() {
@@ -89,25 +108,7 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
 					uploadsCancel()
 					return
 				}
-				reader := NewPutObjReader(hashReader)
+				atomic.AddUint64(&totalBytesWritten, uint64(info.Size))
-				objInfo, err := objAPI.PutObject(uploadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
-					objNamePrefix, i, objCountPerThread[i]), reader, ObjectOptions{
-					UserDefined: map[string]string{
-						xhttp.AmzStorageClass: storageClass,
-					},
-					Speedtest: true,
-				})
-				if err != nil {
-					objCountPerThread[i]--
-					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
-						errOnce.Do(func() {
-							retError = err.Error()
-						})
-					}
-					uploadsCancel()
-					return
-				}
-				atomic.AddUint64(&totalBytesWritten, uint64(objInfo.Size))
 				objCountPerThread[i]++
 			}
 		}(i)
@@ -138,10 +139,12 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
 				if objCountPerThread[i] == j {
 					j = 0
 				}
-				r, err := objAPI.GetObjectNInfo(downloadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
+				opts := minio.GetObjectOptions{}
-					objNamePrefix, i, j), nil, nil, noLock, ObjectOptions{})
+				opts.Set(globalObjectPerfUserMetadata, "true") // Bypass S3 API freeze
+				r, err := client.GetObject(downloadsCtx, globalObjectPerfBucket, fmt.Sprintf("%s%d.%d", objNamePrefix, i, j), opts)
 				if err != nil {
-					if isErrObjectNotFound(err) {
+					errResp, ok := err.(minio.ErrorResponse)
+					if ok && errResp.StatusCode == http.StatusNotFound {
 						continue
 					}
 					if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {