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

This commit is contained in:
Krishna Srinivas 2022-03-16 16:58:46 -07:00 committed by GitHub
parent 0bf80b3c89
commit ff811f594b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 28 additions and 28 deletions

View File

@ -1054,8 +1054,15 @@ func (a adminAPIHandlers) ObjectSpeedtestHandler(w http.ResponseWriter, r *http.
autotune = false autotune = false
} }
err = objectAPI.MakeBucketWithLocation(ctx, globalObjectPerfBucket, BucketOptions{})
if _, ok := err.(BucketExists); !ok {
// Only BucketExists error can be ignored.
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
return
}
deleteBucket := func() { deleteBucket := func() {
objectAPI.DeleteBucket(context.Background(), pathJoin(minioMetaBucket, "speedtest"), DeleteBucketOptions{ objectAPI.DeleteBucket(context.Background(), globalObjectPerfBucket, DeleteBucketOptions{
Force: true, Force: true,
NoRecreate: true, NoRecreate: true,
}) })

View File

@ -349,6 +349,7 @@ var (
// Used for collecting stats for netperf // Used for collecting stats for netperf
globalNetPerfMinDuration = time.Second * 10 globalNetPerfMinDuration = time.Second * 10
globalNetPerfRX netPerfRX globalNetPerfRX netPerfRX
globalObjectPerfBucket = "minio-perf-test-tmp-bucket"
// Add new variable global values here. // Add new variable global values here.
) )

View File

@ -24,6 +24,7 @@ import (
"io" "io"
"io/ioutil" "io/ioutil"
"math/rand" "math/rand"
"net/http"
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -31,8 +32,8 @@ import (
"github.com/dustin/go-humanize" "github.com/dustin/go-humanize"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/minio/madmin-go" "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" "github.com/minio/pkg/randreader"
) )
@ -61,6 +62,15 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
var totalBytesWritten uint64 var totalBytesWritten uint64
var totalBytesRead uint64 var totalBytesRead uint64
client, err := minio.NewCore(globalLocalNodeName, &minio.Options{
Creds: credentials.NewStaticV4(globalActiveCred.AccessKey, globalActiveCred.SecretKey, ""),
Secure: globalIsTLS,
Transport: globalProxyTransport,
})
if err != nil {
return SpeedtestResult{}, err
}
objCountPerThread := make([]uint64, concurrent) objCountPerThread := make([]uint64, concurrent)
uploadsCtx, uploadsCancel := context.WithCancel(context.Background()) uploadsCtx, uploadsCancel := context.WithCancel(context.Background())
@ -71,15 +81,15 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
uploadsCancel() uploadsCancel()
}() }()
objNamePrefix := "speedtest/objects/" + uuid.New().String() objNamePrefix := uuid.New().String() + "/"
wg.Add(concurrent) wg.Add(concurrent)
for i := 0; i < concurrent; i++ { for i := 0; i < concurrent; i++ {
go func(i int) { go func(i int) {
defer wg.Done() defer wg.Done()
for { for {
hashReader, err := hash.NewReader(newRandomReader(size), reader := newRandomReader(size)
int64(size), "", "", int64(size)) _, err := client.PutObject(uploadsCtx, globalObjectPerfBucket, fmt.Sprintf("%s/%d.%d", objNamePrefix, i, objCountPerThread[i]), reader, int64(size), "", "", minio.PutObjectOptions{})
if err != nil { if err != nil {
if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) { if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
errOnce.Do(func() { errOnce.Do(func() {
@ -89,25 +99,7 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
uploadsCancel() uploadsCancel()
return return
} }
reader := NewPutObjReader(hashReader) atomic.AddUint64(&totalBytesWritten, uint64(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]++ objCountPerThread[i]++
} }
}(i) }(i)
@ -138,10 +130,10 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
if objCountPerThread[i] == j { if objCountPerThread[i] == j {
j = 0 j = 0
} }
r, err := objAPI.GetObjectNInfo(downloadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d", r, _, _, err := client.GetObject(downloadsCtx, globalObjectPerfBucket, fmt.Sprintf("%s/%d.%d", objNamePrefix, i, j), minio.GetObjectOptions{})
objNamePrefix, i, j), nil, nil, noLock, ObjectOptions{})
if err != nil { if err != nil {
if isErrObjectNotFound(err) { errResp, ok := err.(minio.ErrorResponse)
if ok && errResp.StatusCode == http.StatusNotFound {
continue continue
} }
if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) { if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {