move Make,Delete,Head,Heal bucket calls to websockets (#18951)

cmd/peer-s3-client.go

@@ -21,14 +21,18 @@ import (
 	"context"
 	"encoding/gob"
 	"errors"
+	"fmt"
 	"io"
 	"net/url"
 	"sort"
 	"strconv"
+	"sync/atomic"
 	"time"
 
 	"github.com/minio/madmin-go/v3"
+	grid "github.com/minio/minio/internal/grid"
 	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/logger"
 	"github.com/minio/minio/internal/rest"
 	"github.com/minio/pkg/v2/sync/errgroup"
 	"golang.org/x/exp/slices"
@@ -49,12 +53,12 @@ type peerS3Client interface {
 }
 
 type localPeerS3Client struct {
-	host  string
+	node  Node
 	pools []int
 }
 
 func (l *localPeerS3Client) GetHost() string {
-	return l.host
+	return l.node.Host
 }
 
 func (l *localPeerS3Client) SetPools(p []int) {
@@ -88,9 +92,13 @@ func (l localPeerS3Client) DeleteBucket(ctx context.Context, bucket string, opts
 
 // client to talk to peer Nodes.
 type remotePeerS3Client struct {
-	host       string
+	node       Node
 	pools      []int
 	restClient *rest.Client
+
+	// Function that returns the grid connection for this peer when initialized.
+	// Will return nil if the grid connection is not initialized yet.
+	gridConn func() *grid.Connection
 }
 
 // Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
@@ -126,7 +134,7 @@ type S3PeerSys struct {
 // NewS3PeerSys - creates new S3 peer calls.
 func NewS3PeerSys(endpoints EndpointServerPools) *S3PeerSys {
 	return &S3PeerSys{
-		peerClients: newPeerS3Clients(endpoints.GetNodes()),
+		peerClients: newPeerS3Clients(endpoints),
 		poolsCount:  len(endpoints),
 	}
 }
@@ -358,37 +366,47 @@ func (client *remotePeerS3Client) ListBuckets(ctx context.Context, opts BucketOp
 }
 
 func (client *remotePeerS3Client) HealBucket(ctx context.Context, bucket string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
-	v := url.Values{}
-	v.Set(peerS3Bucket, bucket)
-	v.Set(peerS3BucketDeleted, strconv.FormatBool(opts.Remove))
-
-	respBody, err := client.call(peerS3MethodHealBucket, v, nil, -1)
-	if err != nil {
-		return madmin.HealResultItem{}, err
+	conn := client.gridConn()
+	if conn == nil {
+		return madmin.HealResultItem{}, nil
 	}
-	defer xhttp.DrainBody(respBody)
 
-	var res madmin.HealResultItem
-	err = gob.NewDecoder(respBody).Decode(&res)
+	mss := grid.NewMSSWith(map[string]string{
+		peerS3Bucket:        bucket,
+		peerS3BucketDeleted: strconv.FormatBool(opts.Remove),
+	})
 
-	return res, err
+	_, err := healBucketHandler.Call(ctx, conn, mss)
+
+	// Initialize heal result info
+	return madmin.HealResultItem{
+		Type:     madmin.HealItemBucket,
+		Bucket:   bucket,
+		SetCount: -1, // explicitly set an invalid value -1, for bucket heal scenario
+	}, toStorageErr(err)
 }
 
 // GetBucketInfo returns bucket stat info from a peer
 func (client *remotePeerS3Client) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
-	v := url.Values{}
-	v.Set(peerS3Bucket, bucket)
-	v.Set(peerS3BucketDeleted, strconv.FormatBool(opts.Deleted))
+	conn := client.gridConn()
+	if conn == nil {
+		return BucketInfo{}, nil
+	}
 
-	respBody, err := client.call(peerS3MethodGetBucketInfo, v, nil, -1)
+	mss := grid.NewMSSWith(map[string]string{
+		peerS3Bucket:        bucket,
+		peerS3BucketDeleted: strconv.FormatBool(opts.Deleted),
+	})
+
+	volInfo, err := headBucketHandler.Call(ctx, conn, mss)
 	if err != nil {
 		return BucketInfo{}, err
 	}
-	defer xhttp.DrainBody(respBody)
 
-	var bucketInfo BucketInfo
-	err = gob.NewDecoder(respBody).Decode(&bucketInfo)
-	return bucketInfo, err
+	return BucketInfo{
+		Name:    volInfo.Name,
+		Created: volInfo.Created,
+	}, nil
 }
 
 // MakeBucket creates bucket across all peers
@@ -421,17 +439,18 @@ func (sys *S3PeerSys) MakeBucket(ctx context.Context, bucket string, opts MakeBu
 
 // MakeBucket creates a bucket on a peer
 func (client *remotePeerS3Client) MakeBucket(ctx context.Context, bucket string, opts MakeBucketOptions) error {
-	v := url.Values{}
-	v.Set(peerS3Bucket, bucket)
-	v.Set(peerS3BucketForceCreate, strconv.FormatBool(opts.ForceCreate))
-
-	respBody, err := client.call(peerS3MethodMakeBucket, v, nil, -1)
-	if err != nil {
-		return err
+	conn := client.gridConn()
+	if conn == nil {
+		return nil
 	}
-	defer xhttp.DrainBody(respBody)
 
-	return nil
+	mss := grid.NewMSSWith(map[string]string{
+		peerS3Bucket:            bucket,
+		peerS3BucketForceCreate: strconv.FormatBool(opts.ForceCreate),
+	})
+
+	_, err := makeBucketHandler.Call(ctx, conn, mss)
+	return toStorageErr(err)
 }
 
 // DeleteBucket deletes bucket across all peers
@@ -466,21 +485,22 @@ func (sys *S3PeerSys) DeleteBucket(ctx context.Context, bucket string, opts Dele
 
 // DeleteBucket deletes bucket on a peer
 func (client *remotePeerS3Client) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
-	v := url.Values{}
-	v.Set(peerS3Bucket, bucket)
-	v.Set(peerS3BucketForceDelete, strconv.FormatBool(opts.Force))
-
-	respBody, err := client.call(peerS3MethodDeleteBucket, v, nil, -1)
-	if err != nil {
-		return err
+	conn := client.gridConn()
+	if conn == nil {
+		return nil
 	}
-	defer xhttp.DrainBody(respBody)
 
-	return nil
+	mss := grid.NewMSSWith(map[string]string{
+		peerS3Bucket:            bucket,
+		peerS3BucketForceDelete: strconv.FormatBool(opts.Force),
+	})
+
+	_, err := deleteBucketHandler.Call(ctx, conn, mss)
+	return toStorageErr(err)
 }
 
 func (client remotePeerS3Client) GetHost() string {
-	return client.host
+	return client.node.Host
 }
 
 func (client remotePeerS3Client) GetPools() []int {
@@ -493,21 +513,23 @@ func (client *remotePeerS3Client) SetPools(p []int) {
 }
 
 // newPeerS3Clients creates new peer clients.
-func newPeerS3Clients(nodes []Node) (peers []peerS3Client) {
+func newPeerS3Clients(endpoints EndpointServerPools) (peers []peerS3Client) {
+	nodes := endpoints.GetNodes()
 	peers = make([]peerS3Client, len(nodes))
 	for i, node := range nodes {
 		if node.IsLocal {
-			peers[i] = &localPeerS3Client{host: node.Host}
+			peers[i] = &localPeerS3Client{node: node}
 		} else {
-			peers[i] = newPeerS3Client(node.Host)
+			peers[i] = newPeerS3Client(node)
 		}
 		peers[i].SetPools(node.Pools)
 	}
-	return
+
+	return peers
 }
 
 // Returns a peer S3 client.
-func newPeerS3Client(peer string) peerS3Client {
+func newPeerS3Client(node Node) peerS3Client {
 	scheme := "http"
 	if globalIsTLS {
 		scheme = "https"
@@ -515,7 +537,7 @@ func newPeerS3Client(peer string) peerS3Client {
 
 	serverURL := &url.URL{
 		Scheme: scheme,
-		Host:   peer,
+		Host:   node.Host,
 		Path:   peerS3Path,
 	}
 
@@ -533,5 +555,32 @@ func newPeerS3Client(peer string) peerS3Client {
 		return !isNetworkError(err)
 	}
 
-	return &remotePeerS3Client{host: peer, restClient: restClient}
+	var gridConn atomic.Pointer[grid.Connection]
+
+	return &remotePeerS3Client{
+		node: node, restClient: restClient,
+		gridConn: func() *grid.Connection {
+			// Lazy initialization of grid connection.
+			// When we create this peer client, the grid connection is likely not yet initialized.
+			if node.GridHost == "" {
+				logger.LogOnceIf(context.Background(), fmt.Errorf("gridHost is empty for peer %s", node.Host), node.Host+":gridHost")
+				return nil
+			}
+			gc := gridConn.Load()
+			if gc != nil {
+				return gc
+			}
+			gm := globalGrid.Load()
+			if gm == nil {
+				return nil
+			}
+			gc = gm.Connection(node.GridHost)
+			if gc == nil {
+				logger.LogOnceIf(context.Background(), fmt.Errorf("gridHost %s not found for peer %s", node.GridHost, node.Host), node.Host+":gridHost")
+				return nil
+			}
+			gridConn.Store(gc)
+			return gc
+		},
+	}
 }