refactor the perf client for TTFB and TotalResponseTime (#17901)

cmd/perf-tests.go

@@ -19,6 +19,7 @@ package cmd
 
 import (
 	"context"
+	"encoding/gob"
 	"errors"
 	"fmt"
 	"io"
@@ -367,29 +368,14 @@ func siteNetperf(ctx context.Context, duration time.Duration) madmin.SiteNetPerf
 		for i := 0; i < connectionsPerPeer; i++ {
 			go func() {
 				defer wg.Done()
-				cli, err := globalSiteReplicationSys.getAdminClient(ctx, info.DeploymentID)
+				ctx, cancel := context.WithTimeout(ctx, duration+10*time.Second)
-				if err != nil {
+				defer cancel()
-					return
+				perfNetRequest(
-				}
+					ctx,
-				rp := cli.GetEndpointURL()
+					info.DeploymentID,
-				reqURL := &url.URL{
+					adminPathPrefix+adminAPIVersionPrefix+adminAPISiteReplicationDevNull,
-					Scheme: rp.Scheme,
+					r,
-					Host:   rp.Host,
+				)
-					Path:   adminPathPrefix + adminAPIVersionPrefix + adminAPISiteReplicationDevNull,
-				}
-				req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL.String(), r)
-				if err != nil {
-					return
-				}
-				client := &http.Client{
-					Timeout:   duration + 10*time.Second,
-					Transport: globalRemoteTargetTransport,
-				}
-				resp, err := client.Do(req)
-				if err != nil {
-					return
-				}
-				defer xhttp.DrainBody(resp.Body)
 			}()
 		}
 	}
@@ -422,3 +408,41 @@ func siteNetperf(ctx context.Context, duration time.Duration) madmin.SiteNetPerf
 		TotalConn:       uint64(connectionsPerPeer),
 	}
 }
+
+// perfNetRequest - reader for http.request.body
+func perfNetRequest(ctx context.Context, deploymentID, reqPath string, reader io.Reader) (result madmin.SiteNetPerfNodeResult) {
+	result = madmin.SiteNetPerfNodeResult{}
+	cli, err := globalSiteReplicationSys.getAdminClient(ctx, deploymentID)
+	if err != nil {
+		result.Error = err.Error()
+		return
+	}
+	rp := cli.GetEndpointURL()
+	reqURL := &url.URL{
+		Scheme: rp.Scheme,
+		Host:   rp.Host,
+		Path:   reqPath,
+	}
+	result.Endpoint = rp.String()
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL.String(), reader)
+	if err != nil {
+		result.Error = err.Error()
+		return
+	}
+	client := &http.Client{
+		Transport: globalRemoteTargetTransport,
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		result.Error = err.Error()
+		return
+	}
+	defer xhttp.DrainBody(resp.Body)
+	err = gob.NewDecoder(resp.Body).Decode(&result)
+	// endpoint have been overwritten
+	result.Endpoint = rp.String()
+	if err != nil {
+		result.Error = err.Error()
+	}
+	return
+}

cmd/site-replication.go

@@ -22,12 +22,10 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/binary"
-	"encoding/gob"
 	"encoding/json"
 	"encoding/xml"
 	"errors"
 	"fmt"
-	"net/http"
 	"net/url"
 	"reflect"
 	"runtime"
@@ -43,7 +41,6 @@ import (
 	"github.com/minio/minio-go/v7/pkg/set"
 	"github.com/minio/minio/internal/auth"
 	sreplication "github.com/minio/minio/internal/bucket/replication"
-	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
 	bktpolicy "github.com/minio/pkg/bucket/policy"
 	iampolicy "github.com/minio/pkg/iam/policy"
@@ -663,64 +660,38 @@ func (c *SiteReplicationSys) Netperf(ctx context.Context, duration time.Duration
 		// will call siteNetperf, means call others's adminAPISiteReplicationDevNull
 		if globalDeploymentID == info.DeploymentID {
 			wg.Add(1)
-			go func() (err error) {
+			go func() {
-				defer wg.Done()
-				result := &madmin.SiteNetPerfNodeResult{}
-				defer func() {
-					if err != nil {
-						result.Error = err.Error()
-					}
-					resultsMu.Lock()
-					results.NodeResults = append(results.NodeResults, *result)
-					resultsMu.Unlock()
-				}()
-				cli, err := globalSiteReplicationSys.getAdminClient(ctx, info.DeploymentID)
-				if err != nil {
-					return err
-				}
-				*result = siteNetperf(ctx, duration)
-				result.Endpoint = cli.GetEndpointURL().String()
-				return nil
-			}()
-			continue
-		}
-		wg.Add(1)
-		go func() (err error) {
 				defer wg.Done()
 				result := madmin.SiteNetPerfNodeResult{}
-			defer func() {
+				cli, err := globalSiteReplicationSys.getAdminClient(ctx, info.DeploymentID)
 				if err != nil {
 					result.Error = err.Error()
+				} else {
+					result = siteNetperf(ctx, duration)
+					result.Endpoint = cli.GetEndpointURL().String()
 				}
 				resultsMu.Lock()
 				results.NodeResults = append(results.NodeResults, result)
 				resultsMu.Unlock()
+				return
 			}()
-			cli, err := globalSiteReplicationSys.getAdminClient(ctx, info.DeploymentID)
+			continue
-			if err != nil {
-				return err
 		}
-			rp := cli.GetEndpointURL()
+		wg.Add(1)
-			reqURL := &url.URL{
+		go func() {
-				Scheme: rp.Scheme,
+			defer wg.Done()
-				Host:   rp.Host,
+			ctx, cancel := context.WithTimeout(ctx, duration+10*time.Second)
-				Path:   adminPathPrefix + adminAPIVersionPrefix + adminAPISiteReplicationNetPerf,
+			defer cancel()
-			}
+			result := perfNetRequest(
-			result.Endpoint = rp.String()
+				ctx,
-			req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL.String(), nil)
+				info.DeploymentID,
-			if err != nil {
+				adminPathPrefix+adminAPIVersionPrefix+adminAPISiteReplicationNetPerf,
-				return err
+				nil,
-			}
+			)
-			client := &http.Client{
+			resultsMu.Lock()
-				Timeout:   duration + 10*time.Second,
+			results.NodeResults = append(results.NodeResults, result)
-				Transport: globalRemoteTargetTransport,
+			resultsMu.Unlock()
-			}
+			return
-			resp, err := client.Do(req)
-			if err != nil {
-				return err
-			}
-			defer xhttp.DrainBody(resp.Body)
-			return gob.NewDecoder(resp.Body).Decode(&result)
 		}()
 	}
 	wg.Wait()