migrate bootstrap logic directly to websockets (#18855)

improve performance for startup sequences by 2x for 300+ nodes.
2024-01-24 13:36:44 -08:00
parent c905d3fe21
commit e377bb949a
19 changed files with 568 additions and 269 deletions
--- a/cmd/bootstrap-peer-server.go
+++ b/cmd/bootstrap-peer-server.go
@@ -19,71 +19,47 @@ package cmd

 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
-	"net/http"
-	"net/url"
-	"os"
+	"math/rand"
 	"reflect"
 	"strings"
+	"sync"
 	"time"

 	"github.com/minio/minio-go/v7/pkg/set"
-	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/grid"
 	"github.com/minio/minio/internal/logger"
-	"github.com/minio/minio/internal/rest"
-	"github.com/minio/mux"
 	"github.com/minio/pkg/v2/env"
 )

-const (
-	bootstrapRESTVersion       = "v1"
-	bootstrapRESTVersionPrefix = SlashSeparator + bootstrapRESTVersion
-	bootstrapRESTPrefix        = minioReservedBucketPath + "/bootstrap"
-	bootstrapRESTPath          = bootstrapRESTPrefix + bootstrapRESTVersionPrefix
-)
-
-const (
-	bootstrapRESTMethodHealth = "/health"
-	bootstrapRESTMethodVerify = "/verify"
-)
-
 // To abstract a node over network.
 type bootstrapRESTServer struct{}

+//go:generate msgp -file=$GOFILE
+
 // ServerSystemConfig - captures information about server configuration.
 type ServerSystemConfig struct {
-	MinioEndpoints EndpointServerPools
-	MinioEnv       map[string]string
+	NEndpoints int
+	CmdLines   []string
+	MinioEnv   map[string]string
 }

 // Diff - returns error on first difference found in two configs.
-func (s1 ServerSystemConfig) Diff(s2 ServerSystemConfig) error {
-	if s1.MinioEndpoints.NEndpoints() != s2.MinioEndpoints.NEndpoints() {
-		return fmt.Errorf("Expected number of endpoints %d, seen %d", s1.MinioEndpoints.NEndpoints(),
-			s2.MinioEndpoints.NEndpoints())
+func (s1 *ServerSystemConfig) Diff(s2 *ServerSystemConfig) error {
+	ns1 := s1.NEndpoints
+	ns2 := s2.NEndpoints
+	if ns1 != ns2 {
+		return fmt.Errorf("Expected number of endpoints %d, seen %d", ns1, ns2)
 	}

-	for i, ep := range s1.MinioEndpoints {
-		if ep.CmdLine != s2.MinioEndpoints[i].CmdLine {
-			return fmt.Errorf("Expected command line argument %s, seen %s", ep.CmdLine,
-				s2.MinioEndpoints[i].CmdLine)
-		}
-		if ep.SetCount != s2.MinioEndpoints[i].SetCount {
-			return fmt.Errorf("Expected set count %d, seen %d", ep.SetCount,
-				s2.MinioEndpoints[i].SetCount)
-		}
-		if ep.DrivesPerSet != s2.MinioEndpoints[i].DrivesPerSet {
-			return fmt.Errorf("Expected drives pet set %d, seen %d", ep.DrivesPerSet,
-				s2.MinioEndpoints[i].DrivesPerSet)
-		}
-		if ep.Platform != s2.MinioEndpoints[i].Platform {
-			return fmt.Errorf("Expected platform '%s', found to be on '%s'",
-				ep.Platform, s2.MinioEndpoints[i].Platform)
+	for i, cmdLine := range s1.CmdLines {
+		if cmdLine != s2.CmdLines[i] {
+			return fmt.Errorf("Expected command line argument %s, seen %s", cmdLine,
+				s2.CmdLines[i])
 		}
 	}
+
 	if reflect.DeepEqual(s1.MinioEnv, s2.MinioEnv) {
 		return nil
 	}
@@ -131,7 +107,7 @@ var skipEnvs = map[string]struct{}{
 	"MINIO_SECRET_KEY":    {},
 }

-func getServerSystemCfg() ServerSystemConfig {
+func getServerSystemCfg() *ServerSystemConfig {
 	envs := env.List("MINIO_")
 	envValues := make(map[string]string, len(envs))
 	for _, envK := range envs {
@@ -144,120 +120,102 @@ func getServerSystemCfg() ServerSystemConfig {
 		}
 		envValues[envK] = logger.HashString(env.Get(envK, ""))
 	}
-	return ServerSystemConfig{
-		MinioEndpoints: globalEndpoints,
-		MinioEnv:       envValues,
+	scfg := &ServerSystemConfig{NEndpoints: globalEndpoints.NEndpoints(), MinioEnv: envValues}
+	var cmdLines []string
+	for _, ep := range globalEndpoints {
+		cmdLines = append(cmdLines, ep.CmdLine)
 	}
+	scfg.CmdLines = cmdLines
+	return scfg
 }

-func (b *bootstrapRESTServer) writeErrorResponse(w http.ResponseWriter, err error) {
-	w.WriteHeader(http.StatusForbidden)
-	w.Write([]byte(err.Error()))
+func (s *bootstrapRESTServer) VerifyHandler(params *grid.MSS) (*ServerSystemConfig, *grid.RemoteErr) {
+	return getServerSystemCfg(), nil
 }

-// HealthHandler returns success if request is valid
-func (b *bootstrapRESTServer) HealthHandler(w http.ResponseWriter, r *http.Request) {}
-
-func (b *bootstrapRESTServer) VerifyHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "VerifyHandler")
-
-	if err := storageServerRequestValidate(r); err != nil {
-		b.writeErrorResponse(w, err)
-		return
-	}
-
-	cfg := getServerSystemCfg()
-	logger.LogIf(ctx, json.NewEncoder(w).Encode(&cfg))
-}
+var serverVerifyHandler = grid.NewSingleHandler[*grid.MSS, *ServerSystemConfig](grid.HandlerServerVerify, grid.NewMSS, func() *ServerSystemConfig { return &ServerSystemConfig{} })

 // registerBootstrapRESTHandlers - register bootstrap rest router.
-func registerBootstrapRESTHandlers(router *mux.Router) {
-	h := func(f http.HandlerFunc) http.HandlerFunc {
-		return collectInternodeStats(httpTraceHdrs(f))
-	}
-
+func registerBootstrapRESTHandlers(gm *grid.Manager) {
 	server := &bootstrapRESTServer{}
-	subrouter := router.PathPrefix(bootstrapRESTPrefix).Subrouter()
-
-	subrouter.Methods(http.MethodPost).Path(bootstrapRESTVersionPrefix + bootstrapRESTMethodHealth).HandlerFunc(
-		h(server.HealthHandler))
-
-	subrouter.Methods(http.MethodPost).Path(bootstrapRESTVersionPrefix + bootstrapRESTMethodVerify).HandlerFunc(
-		h(server.VerifyHandler))
+	logger.FatalIf(serverVerifyHandler.Register(gm, server.VerifyHandler), "unable to register handler")
 }

 // client to talk to bootstrap NEndpoints.
 type bootstrapRESTClient struct {
-	endpoint   Endpoint
-	restClient *rest.Client
+	gridConn *grid.Connection
 }

-// Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
-// permanently. The only way to restore the connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
-// after verifying format.json
-func (client *bootstrapRESTClient) callWithContext(ctx context.Context, method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
-	if values == nil {
-		values = make(url.Values)
+// Verify function verifies the server config.
+func (client *bootstrapRESTClient) Verify(ctx context.Context, srcCfg *ServerSystemConfig) (err error) {
+	if newObjectLayerFn() != nil {
+		return nil
 	}

-	respBody, err = client.restClient.Call(ctx, method, values, body, length)
-	if err == nil {
-		return respBody, nil
+	recvCfg, err := serverVerifyHandler.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{}))
+	if err != nil {
+		return err
 	}

-	return nil, err
+	return srcCfg.Diff(recvCfg)
 }

 // Stringer provides a canonicalized representation of node.
 func (client *bootstrapRESTClient) String() string {
-	return client.endpoint.String()
+	return client.gridConn.String()
 }

-// Verify - fetches system server config.
-func (client *bootstrapRESTClient) Verify(ctx context.Context, srcCfg ServerSystemConfig) (err error) {
-	if newObjectLayerFn() != nil {
-		return nil
-	}
-	respBody, err := client.callWithContext(ctx, bootstrapRESTMethodVerify, nil, nil, -1)
-	if err != nil {
-		return
-	}
-	defer xhttp.DrainBody(respBody)
-	recvCfg := ServerSystemConfig{}
-	if err = json.NewDecoder(respBody).Decode(&recvCfg); err != nil {
-		return err
-	}
-	return srcCfg.Diff(recvCfg)
-}
-
-func verifyServerSystemConfig(ctx context.Context, endpointServerPools EndpointServerPools) error {
+func verifyServerSystemConfig(ctx context.Context, endpointServerPools EndpointServerPools, gm *grid.Manager) error {
 	srcCfg := getServerSystemCfg()
-	clnts := newBootstrapRESTClients(endpointServerPools)
+	clnts := newBootstrapRESTClients(endpointServerPools, gm)
 	var onlineServers int
 	var offlineEndpoints []error
 	var incorrectConfigs []error
 	var retries int
+	var mu sync.Mutex
 	for onlineServers < len(clnts)/2 {
+		var wg sync.WaitGroup
+		wg.Add(len(clnts))
+		onlineServers = 0
 		for _, clnt := range clnts {
-			if err := clnt.Verify(ctx, srcCfg); err != nil {
-				bootstrapTraceMsg(fmt.Sprintf("clnt.Verify: %v, endpoint: %v", err, clnt.endpoint))
-				if !isNetworkError(err) {
-					logger.LogOnceIf(ctx, fmt.Errorf("%s has incorrect configuration: %w", clnt.String(), err), clnt.String())
-					incorrectConfigs = append(incorrectConfigs, fmt.Errorf("%s has incorrect configuration: %w", clnt.String(), err))
-				} else {
-					offlineEndpoints = append(offlineEndpoints, fmt.Errorf("%s is unreachable: %w", clnt.String(), err))
+			go func(clnt *bootstrapRESTClient) {
+				defer wg.Done()
+
+				if clnt.gridConn.State() != grid.StateConnected {
+					mu.Lock()
+					offlineEndpoints = append(offlineEndpoints, fmt.Errorf("%s is unreachable: %w", clnt, grid.ErrDisconnected))
+					mu.Unlock()
+					return
 				}
-				continue
-			}
-			onlineServers++
+
+				ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
+				defer cancel()
+
+				err := clnt.Verify(ctx, srcCfg)
+				mu.Lock()
+				if err != nil {
+					bootstrapTraceMsg(fmt.Sprintf("clnt.Verify: %v, endpoint: %s", err, clnt))
+					if !isNetworkError(err) {
+						logger.LogOnceIf(context.Background(), fmt.Errorf("%s has incorrect configuration: %w", clnt, err), "incorrect_"+clnt.String())
+						incorrectConfigs = append(incorrectConfigs, fmt.Errorf("%s has incorrect configuration: %w", clnt, err))
+					} else {
+						offlineEndpoints = append(offlineEndpoints, fmt.Errorf("%s is unreachable: %w", clnt, err))
+					}
+				} else {
+					onlineServers++
+				}
+				mu.Unlock()
+			}(clnt)
 		}
+		wg.Wait()
+
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		default:
-			// Sleep for a while - so that we don't go into
-			// 100% CPU when half the endpoints are offline.
-			time.Sleep(100 * time.Millisecond)
+			// Sleep and stagger to avoid blocked CPU and thundering
+			// herd upon start up sequence.
+			time.Sleep(25*time.Millisecond + time.Duration(rand.Int63n(int64(100*time.Millisecond))))
 			retries++
 			// after 20 retries start logging that servers are not reachable yet
 			if retries >= 20 {
@@ -268,7 +226,7 @@ func verifyServerSystemConfig(ctx context.Context, endpointServerPools EndpointS
 				if len(incorrectConfigs) > 0 {
 					logger.Info(fmt.Sprintf("Following servers have mismatching configuration %s", incorrectConfigs))
 				}
-				retries = 0 // reset to log again after 5 retries.
+				retries = 0 // reset to log again after 20 retries.
 			}
 			offlineEndpoints = nil
 			incorrectConfigs = nil
@@ -277,39 +235,20 @@ func verifyServerSystemConfig(ctx context.Context, endpointServerPools EndpointS
 	return nil
 }

-func newBootstrapRESTClients(endpointServerPools EndpointServerPools) []*bootstrapRESTClient {
-	seenHosts := set.NewStringSet()
+func newBootstrapRESTClients(endpointServerPools EndpointServerPools, gm *grid.Manager) []*bootstrapRESTClient {
+	seenClient := set.NewStringSet()
 	var clnts []*bootstrapRESTClient
 	for _, ep := range endpointServerPools {
 		for _, endpoint := range ep.Endpoints {
-			if seenHosts.Contains(endpoint.Host) {
+			if endpoint.IsLocal {
 				continue
 			}
-			seenHosts.Add(endpoint.Host)
-
-			// Only proceed for remote endpoints.
-			if !endpoint.IsLocal {
-				cl := newBootstrapRESTClient(endpoint)
-				if serverDebugLog {
-					cl.restClient.TraceOutput = os.Stdout
-				}
-				clnts = append(clnts, cl)
+			if seenClient.Contains(endpoint.Host) {
+				continue
 			}
+			seenClient.Add(endpoint.Host)
+			clnts = append(clnts, &bootstrapRESTClient{gm.Connection(endpoint.GridHost())})
 		}
 	}
 	return clnts
 }
-
-// Returns a new bootstrap client.
-func newBootstrapRESTClient(endpoint Endpoint) *bootstrapRESTClient {
-	serverURL := &url.URL{
-		Scheme: endpoint.Scheme,
-		Host:   endpoint.Host,
-		Path:   bootstrapRESTPath,
-	}
-
-	restClient := rest.NewClient(serverURL, globalInternodeTransport, newCachedAuthToken())
-	restClient.HealthCheckFn = nil
-
-	return &bootstrapRESTClient{endpoint: endpoint, restClient: restClient}
-}