Add basic bandwidth monitoring for replication. (#10501)

This change tracks bandwidth for a bucket and object

- [x] Add Admin API
- [x] Add Peer API
- [x] Add BW throttling
- [x] Admin APIs to set replication limit
- [x] Admin APIs for fetch bandwidth

cmd/admin-handlers.go

@@ -35,17 +35,19 @@ import (
 	"time"
 
 	"github.com/gorilla/mux"
-
 	"github.com/minio/minio/cmd/config"
 	"github.com/minio/minio/cmd/crypto"
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/cmd/logger/message/log"
 	"github.com/minio/minio/pkg/auth"
+	bandwidth "github.com/minio/minio/pkg/bandwidth"
+	bucketBandwidth "github.com/minio/minio/pkg/bucket/bandwidth"
 	"github.com/minio/minio/pkg/handlers"
 	iampolicy "github.com/minio/minio/pkg/iam/policy"
 	"github.com/minio/minio/pkg/madmin"
 	xnet "github.com/minio/minio/pkg/net"
+	"github.com/minio/minio/pkg/sync/errgroup"
 	trace "github.com/minio/minio/pkg/trace"
 )
 
@@ -1425,6 +1427,66 @@ func (a adminAPIHandlers) OBDInfoHandler(w http.ResponseWriter, r *http.Request)
 
 }
 
+// BandwidthMonitorHandler - GET /minio/admin/v3/bandwidth
+// ----------
+// Get bandwidth consumption information
+func (a adminAPIHandlers) BandwidthMonitorHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "BandwidthMonitor")
+
+	// Validate request signature.
+	_, adminAPIErr := checkAdminRequestAuthType(ctx, r, iampolicy.BandwidthMonitorAction, "")
+	if adminAPIErr != ErrNone {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
+		return
+	}
+
+	setEventStreamHeaders(w)
+	peers := newPeerRestClients(globalEndpoints)
+	bucketsRequestedString := r.URL.Query().Get("buckets")
+	var bucketsRequested []string
+	reports := make([]*bandwidth.Report, len(peers))
+	selectBuckets := bucketBandwidth.SelectAllBuckets()
+	if bucketsRequestedString != "" {
+		bucketsRequested = strings.Split(bucketsRequestedString, ",")
+		selectBuckets = bucketBandwidth.SelectBuckets(bucketsRequested...)
+	}
+	reports = append(reports, globalBucketMonitor.GetReport(selectBuckets))
+	g := errgroup.WithNErrs(len(peers))
+	for index, peer := range peers {
+		if peer == nil {
+			continue
+		}
+		index := index
+		g.Go(func() error {
+			var err error
+			reports[index], err = peer.MonitorBandwidth(ctx, bucketsRequested)
+			return err
+		}, index)
+	}
+	consolidatedReport := bandwidth.Report{
+		BucketStats: make(map[string]bandwidth.Details),
+	}
+
+	for _, report := range reports {
+		for bucket := range report.BucketStats {
+			d, ok := consolidatedReport.BucketStats[bucket]
+			if !ok {
+				consolidatedReport.BucketStats[bucket] = bandwidth.Details{}
+				d = consolidatedReport.BucketStats[bucket]
+				d.LimitInBytesPerSecond = report.BucketStats[bucket].LimitInBytesPerSecond
+			}
+			d.CurrentBandwidthInBytesPerSecond += report.BucketStats[bucket].CurrentBandwidthInBytesPerSecond
+			consolidatedReport.BucketStats[bucket] = d
+		}
+	}
+	enc := json.NewEncoder(w)
+	err := enc.Encode(consolidatedReport)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInternalError), r.URL)
+	}
+	w.(http.Flusher).Flush()
+}
+
 // ServerInfoHandler - GET /minio/admin/v3/info
 // ----------
 // Get server information

cmd/admin-router.go

@@ -214,6 +214,8 @@ func registerAdminRouter(router *mux.Router, enableConfigOps, enableIAMOps bool)
 			// -- OBD API --
 			adminRouter.Methods(http.MethodGet).Path(adminVersion + "/obdinfo").
 				HandlerFunc(httpTraceHdrs(adminAPI.OBDInfoHandler))
+			adminRouter.Methods(http.MethodGet).Path(adminVersion + "/bandwidth").
+				HandlerFunc(httpTraceHdrs(adminAPI.BandwidthMonitorHandler))
 		}
 	}
 

cmd/bucket-metadata-sys.go

@@ -373,6 +373,20 @@ func (sys *BucketMetadataSys) GetBucketTargetsConfig(bucket string) (*madmin.Buc
 	return meta.bucketTargetConfig, nil
 }
 
+// GetBucketTarget returns the target for the bucket and arn.
+func (sys *BucketMetadataSys) GetBucketTarget(bucket string, arn string) (madmin.BucketTarget, error) {
+	targets, err := sys.GetBucketTargetsConfig(bucket)
+	if err != nil {
+		return madmin.BucketTarget{}, err
+	}
+	for _, t := range targets.Targets {
+		if t.Arn == arn {
+			return t, nil
+		}
+	}
+	return madmin.BucketTarget{}, errConfigNotFound
+}
+
 // GetConfig returns a specific configuration from the bucket metadata.
 // The returned object may not be modified.
 func (sys *BucketMetadataSys) GetConfig(bucket string) (BucketMetadata, error) {

cmd/bucket-replication.go

@@ -30,6 +30,7 @@ import (
 	"github.com/minio/minio/cmd/crypto"
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
+	"github.com/minio/minio/pkg/bucket/bandwidth"
 	"github.com/minio/minio/pkg/bucket/replication"
 	"github.com/minio/minio/pkg/event"
 	iampolicy "github.com/minio/minio/pkg/iam/policy"
@@ -119,7 +120,7 @@ func mustReplicater(ctx context.Context, r *http.Request, bucket, object string,
 	return cfg.Replicate(opts)
 }
 
-func putReplicationOpts(dest replication.Destination, objInfo ObjectInfo) (putOpts miniogo.PutObjectOptions) {
+func putReplicationOpts(ctx context.Context, dest replication.Destination, objInfo ObjectInfo) (putOpts miniogo.PutObjectOptions) {
 	meta := make(map[string]string)
 	for k, v := range objInfo.UserDefined {
 		if k == xhttp.AmzBucketReplicationStatus {
@@ -168,6 +169,7 @@ func putReplicationOpts(dest replication.Destination, objInfo ObjectInfo) (putOp
 	if crypto.S3.IsEncrypted(objInfo.UserDefined) {
 		putOpts.ServerSideEncryption = encrypt.NewSSE()
 	}
+
 	return
 }
 
@@ -184,16 +186,15 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 	}
 	tgt := globalBucketTargetSys.GetRemoteTargetClient(ctx, cfg.RoleArn)
 	if tgt == nil {
+		logger.LogIf(ctx, fmt.Errorf("failed to get target for bucket:%s arn:%s", bucket, cfg.RoleArn))
 		return
 	}
-
 	gr, err := objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{
 		VersionID: objInfo.VersionID,
 	})
 	if err != nil {
 		return
 	}
-
 	objInfo = gr.ObjInfo
 	size, err := objInfo.GetActualSize()
 	if err != nil {
@@ -224,11 +225,26 @@ func replicateObject(ctx context.Context, objInfo ObjectInfo, objectAPI ObjectLa
 			return
 		}
 	}
-	putOpts := putReplicationOpts(dest, objInfo)
 
+	target, err := globalBucketMetadataSys.GetBucketTarget(bucket, cfg.RoleArn)
+	if err != nil {
+		logger.LogIf(ctx, fmt.Errorf("failed to get target for replication bucket:%s cfg:%s err:%s", bucket, cfg.RoleArn, err))
+		return
+	}
+	putOpts := putReplicationOpts(ctx, dest, objInfo)
 	replicationStatus := replication.Complete
-	_, err = tgt.PutObject(ctx, dest.Bucket, object, gr, size, "", "", putOpts)
-	gr.Close()
+
+	// Setup bandwidth throttling
+	peerCount := len(globalEndpoints)
+	b := target.BandwidthLimit / int64(peerCount)
+	var headerSize int
+	for k, v := range putOpts.Header() {
+		headerSize += len(k) + len(v)
+	}
+	r := bandwidth.NewMonitoredReader(ctx, globalBucketMonitor, objInfo.Bucket, objInfo.Name, gr, headerSize, b)
+
+	_, err = tgt.PutObject(ctx, dest.Bucket, object, r, size, "", "", putOpts)
+	r.Close()
 	if err != nil {
 		replicationStatus = replication.Failed
 	}

cmd/bucket-targets.go

@@ -207,14 +207,14 @@ func (sys *BucketTargetSys) Init(ctx context.Context, buckets []BucketInfo, objA
 	return nil
 }
 
-// UpdateTarget updates target to reflect metadata updates
-func (sys *BucketTargetSys) UpdateTarget(bucket string, cfg *madmin.BucketTargets) {
+// UpdateAllTargets updates target to reflect metadata updates
+func (sys *BucketTargetSys) UpdateAllTargets(bucket string, tgts *madmin.BucketTargets) {
 	if sys == nil {
 		return
 	}
 	sys.Lock()
 	defer sys.Unlock()
-	if cfg == nil || cfg.Empty() {
+	if tgts == nil || tgts.Empty() {
 		// remove target and arn association
 		if tgts, ok := sys.targetsMap[bucket]; ok {
 			for _, t := range tgts {
@@ -225,10 +225,10 @@ func (sys *BucketTargetSys) UpdateTarget(bucket string, cfg *madmin.BucketTarget
 		return
 	}
 
-	if len(cfg.Targets) > 0 {
-		sys.targetsMap[bucket] = cfg.Targets
+	if len(tgts.Targets) > 0 {
+		sys.targetsMap[bucket] = tgts.Targets
 	}
-	for _, tgt := range cfg.Targets {
+	for _, tgt := range tgts.Targets {
 		tgtClient, err := sys.getRemoteTargetClient(&tgt)
 		if err != nil {
 			continue
@@ -238,7 +238,7 @@ func (sys *BucketTargetSys) UpdateTarget(bucket string, cfg *madmin.BucketTarget
 			sys.clientsCache[tgtClient.EndpointURL().String()] = tgtClient
 		}
 	}
-	sys.targetsMap[bucket] = cfg.Targets
+	sys.targetsMap[bucket] = tgts.Targets
 }
 
 // create minio-go clients for buckets having remote targets

cmd/globals.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/minio/minio-go/v7/pkg/set"
+	"github.com/minio/minio/pkg/bucket/bandwidth"
 
 	humanize "github.com/dustin/go-humanize"
 	"github.com/minio/minio/cmd/config/cache"
@@ -149,6 +150,7 @@ var (
 	globalEnvTargetList *event.TargetList
 
 	globalBucketMetadataSys *BucketMetadataSys
+	globalBucketMonitor     *bandwidth.Monitor
 	globalPolicySys         *PolicySys
 	globalIAMSys            *IAMSys
 

cmd/peer-rest-client.go

@@ -26,6 +26,7 @@ import (
 	"math"
 	"net/url"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -35,6 +36,7 @@ import (
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/cmd/rest"
+	"github.com/minio/minio/pkg/bandwidth"
 	"github.com/minio/minio/pkg/event"
 	"github.com/minio/minio/pkg/madmin"
 	xnet "github.com/minio/minio/pkg/net"
@@ -884,3 +886,20 @@ func newPeerRESTClient(peer *xnet.Host) *peerRESTClient {
 
 	return &peerRESTClient{host: peer, restClient: restClient}
 }
+
+// MonitorBandwidth - send http trace request to peer nodes
+func (client *peerRESTClient) MonitorBandwidth(ctx context.Context, buckets []string) (*bandwidth.Report, error) {
+	values := make(url.Values)
+	values.Set(peerRESTBuckets, strings.Join(buckets, ","))
+
+	respBody, err := client.callWithContext(ctx, peerRESTMethodGetBandwidth, values, nil, -1)
+	if err != nil {
+		return nil, err
+	}
+	defer http.DrainBody(respBody)
+
+	dec := gob.NewDecoder(respBody)
+	var bandwidthReport bandwidth.Report
+	err = dec.Decode(&bandwidthReport)
+	return &bandwidthReport, err
+}

cmd/peer-rest-common.go

@@ -57,10 +57,12 @@ const (
 	peerRESTMethodListen                = "/listen"
 	peerRESTMethodLog                   = "/log"
 	peerRESTMethodGetLocalDiskIDs       = "/getlocaldiskids"
+	peerRESTMethodGetBandwidth          = "/bandwidth"
 )
 
 const (
 	peerRESTBucket      = "bucket"
+	peerRESTBuckets     = "buckets"
 	peerRESTUser        = "user"
 	peerRESTGroup       = "group"
 	peerRESTUserTemp    = "user-temp"

cmd/peer-rest-server.go

@@ -30,6 +30,8 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/minio/minio/cmd/logger"
+	"github.com/minio/minio/pkg/bandwidth"
+	b "github.com/minio/minio/pkg/bucket/bandwidth"
 	"github.com/minio/minio/pkg/event"
 	"github.com/minio/minio/pkg/madmin"
 	trace "github.com/minio/minio/pkg/trace"
@@ -628,7 +630,7 @@ func (s *peerRESTServer) LoadBucketMetadataHandler(w http.ResponseWriter, r *htt
 	}
 
 	if meta.bucketTargetConfig != nil {
-		globalBucketTargetSys.UpdateTarget(bucketName, meta.bucketTargetConfig)
+		globalBucketTargetSys.UpdateAllTargets(bucketName, meta.bucketTargetConfig)
 	}
 }
 
@@ -1047,6 +1049,34 @@ func (s *peerRESTServer) IsValid(w http.ResponseWriter, r *http.Request) bool {
 	return true
 }
 
+// GetBandwidth gets the bandwidth for the buckets requested.
+func (s *peerRESTServer) GetBandwidth(w http.ResponseWriter, r *http.Request) {
+	if !s.IsValid(w, r) {
+		s.writeErrorResponse(w, errors.New("Invalid request"))
+		return
+	}
+	bucketsString := r.URL.Query().Get("buckets")
+	w.WriteHeader(http.StatusOK)
+	w.(http.Flusher).Flush()
+
+	doneCh := make(chan struct{})
+	defer close(doneCh)
+
+	var report *bandwidth.Report
+	selectBuckets := b.SelectAllBuckets()
+	if bucketsString != "" {
+		selectBuckets = b.SelectBuckets(strings.Split(bucketsString, ",")...)
+	}
+	report = globalBucketMonitor.GetReport(selectBuckets)
+
+	enc := gob.NewEncoder(w)
+	if err := enc.Encode(report); err != nil {
+		s.writeErrorResponse(w, errors.New("Encoding report failed: "+err.Error()))
+		return
+	}
+	w.(http.Flusher).Flush()
+}
+
 // registerPeerRESTHandlers - register peer rest router.
 func registerPeerRESTHandlers(router *mux.Router) {
 	server := &peerRESTServer{}
@@ -1085,4 +1115,5 @@ func registerPeerRESTHandlers(router *mux.Router) {
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodBackgroundHealStatus).HandlerFunc(server.BackgroundHealStatusHandler)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLog).HandlerFunc(server.ConsoleLogHandler)
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodGetLocalDiskIDs).HandlerFunc(httpTraceHdrs(server.GetLocalDiskIDs))
+	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodGetBandwidth).HandlerFunc(httpTraceHdrs(server.GetBandwidth))
 }

cmd/server-main.go

@@ -32,6 +32,7 @@ import (
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/auth"
+	"github.com/minio/minio/pkg/bucket/bandwidth"
 	"github.com/minio/minio/pkg/certs"
 	"github.com/minio/minio/pkg/color"
 	"github.com/minio/minio/pkg/env"
@@ -158,6 +159,9 @@ func newAllSubsystems() {
 	// Create new bucket metadata system.
 	globalBucketMetadataSys = NewBucketMetadataSys()
 
+	// Create the bucket bandwidth monitor
+	globalBucketMonitor = bandwidth.NewMonitor(GlobalServiceDoneCh)
+
 	// Create a new config system.
 	globalConfigSys = NewConfigSys()