allow service freeze/unfreeze on a setup (#13707)

an active running speedTest will reject all
new S3 requests to the server, until speedTest
is complete.

this is to ensure that speedTest results are
accurate and trusted.

Co-authored-by: Klaus Post <klauspost@gmail.com>

cmd/admin-handler-utils.go

@@ -29,7 +29,7 @@ import (
 	iampolicy "github.com/minio/pkg/iam/policy"
 )
 
-func validateAdminReq(ctx context.Context, w http.ResponseWriter, r *http.Request, action iampolicy.AdminAction) (ObjectLayer, auth.Credentials) {
+func validateAdminReq(ctx context.Context, w http.ResponseWriter, r *http.Request, actions ...iampolicy.AdminAction) (ObjectLayer, auth.Credentials) {
 	// Get current object layer instance.
 	objectAPI := newObjectLayerFn()
 	if objectAPI == nil || globalNotificationSys == nil {
@@ -37,14 +37,17 @@ func validateAdminReq(ctx context.Context, w http.ResponseWriter, r *http.Reques
 		return nil, auth.Credentials{}
 	}
 
-	// Validate request signature.
-	cred, adminAPIErr := checkAdminRequestAuth(ctx, r, action, "")
-	if adminAPIErr != ErrNone {
-		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
-		return nil, cred
+	for _, action := range actions {
+		// Validate request signature.
+		cred, adminAPIErr := checkAdminRequestAuth(ctx, r, action, "")
+		if adminAPIErr != ErrNone {
+			writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(adminAPIErr), r.URL)
+			return nil, cred
+		}
+		return objectAPI, cred
 	}
-
-	return objectAPI, cred
+	writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+	return nil, auth.Credentials{}
 }
 
 // AdminError - is a generic error for all admin APIs.

cmd/admin-handlers.go

@@ -190,7 +190,11 @@ func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Req
 
 // ServiceHandler - POST /minio/admin/v3/service?action={action}
 // ----------
-// restarts/stops minio server gracefully. In a distributed setup,
+// Supports following actions:
+// - restart (restarts all the MinIO instances in a setup)
+// - stop (stops all the MinIO instances in a setup)
+// - freeze (freezes all incoming S3 API calls)
+// - unfreeze (unfreezes previously frozen S3 API calls)
 func (a adminAPIHandlers) ServiceHandler(w http.ResponseWriter, r *http.Request) {
 	ctx := newContext(r, w, "Service")
 
@@ -205,6 +209,10 @@ func (a adminAPIHandlers) ServiceHandler(w http.ResponseWriter, r *http.Request)
 		serviceSig = serviceRestart
 	case madmin.ServiceActionStop:
 		serviceSig = serviceStop
+	case madmin.ServiceActionFreeze:
+		serviceSig = serviceFreeze
+	case madmin.ServiceActionUnfreeze:
+		serviceSig = serviceUnFreeze
 	default:
 		logger.LogIf(ctx, fmt.Errorf("Unrecognized service action %s requested", action), logger.Application)
 		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrMalformedPOSTRequest), r.URL)
@@ -217,6 +225,8 @@ func (a adminAPIHandlers) ServiceHandler(w http.ResponseWriter, r *http.Request)
 		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceRestartAdminAction)
 	case serviceStop:
 		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceStopAdminAction)
+	case serviceFreeze, serviceUnFreeze:
+		objectAPI, _ = validateAdminReq(ctx, w, r, iampolicy.ServiceFreezeAdminAction)
 	}
 	if objectAPI == nil {
 		return
@@ -233,7 +243,14 @@ func (a adminAPIHandlers) ServiceHandler(w http.ResponseWriter, r *http.Request)
 	// Reply to the client before restarting, stopping MinIO server.
 	writeSuccessResponseHeadersOnly(w)
 
-	globalServiceSignalCh <- serviceSig
+	switch serviceSig {
+	case serviceFreeze:
+		freezeServices()
+	case serviceUnFreeze:
+		unfreezeServices()
+	case serviceRestart, serviceStop:
+		globalServiceSignalCh <- serviceSig
+	}
 }
 
 // ServerProperties holds some server information such as, version, region
@@ -917,12 +934,16 @@ func (a adminAPIHandlers) BackgroundHealStatusHandler(w http.ResponseWriter, r *
 	}
 }
 
+// SpeedtestHandler - reports maximum speed of a cluster by performing PUT and
+// GET operations on the server, supports auto tuning by default by automatically
+// increasing concurrency and stopping when we have reached the limits on the
+// system.
 func (a adminAPIHandlers) SpeedtestHandler(w http.ResponseWriter, r *http.Request) {
 	ctx := newContext(r, w, "SpeedtestHandler")
 
 	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
 
-	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.HealAdminAction)
+	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.HealthInfoAdminAction)
 	if objectAPI == nil {
 		return
 	}
@@ -935,12 +956,7 @@ func (a adminAPIHandlers) SpeedtestHandler(w http.ResponseWriter, r *http.Reques
 	sizeStr := r.Form.Get(peerRESTSize)
 	durationStr := r.Form.Get(peerRESTDuration)
 	concurrentStr := r.Form.Get(peerRESTConcurrent)
-	autotuneStr := r.Form.Get("autotune")
-
-	var autotune bool
-	if autotuneStr != "" {
-		autotune = true
-	}
+	autotune := r.Form.Get("autotune") == "true"
 
 	size, err := strconv.Atoi(sizeStr)
 	if err != nil {
@@ -965,6 +981,12 @@ func (a adminAPIHandlers) SpeedtestHandler(w http.ResponseWriter, r *http.Reques
 		})
 	}
 
+	// Freeze all incoming S3 API calls before running speedtest.
+	globalNotificationSys.ServiceFreeze(ctx, true)
+
+	// unfreeze all incoming S3 API calls after speedtest.
+	defer globalNotificationSys.ServiceFreeze(ctx, false)
+
 	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
 	defer keepAliveTicker.Stop()
 

cmd/globals.go

@@ -23,6 +23,7 @@ import (
 	"net/http"
 	"os"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/minio/console/restapi"
@@ -325,6 +326,13 @@ var (
 
 	globalConsoleSrv *restapi.Server
 
+	// handles service freeze or un-freeze S3 API calls.
+	globalServiceFreeze atomic.Value
+
+	// Only needed for tracking
+	globalServiceFreezeCnt int32
+	globalServiceFreezeMu  sync.Mutex // Updates.
+
 	// Add new variable global values here.
 )
 

cmd/handler-api.go

@@ -25,7 +25,7 @@ import (
 	"sync"
 	"time"
 
-	mem "github.com/shirou/gopsutil/v3/mem"
+	"github.com/shirou/gopsutil/v3/mem"
 
 	"github.com/minio/minio/internal/config/api"
 	xioutil "github.com/minio/minio/internal/ioutil"
@@ -227,6 +227,13 @@ func (t *apiConfig) getRequestsPool() (chan struct{}, time.Duration) {
 // maxClients throttles the S3 API calls
 func maxClients(f http.HandlerFunc) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
+		if val := globalServiceFreeze.Load(); val != nil {
+			if unlock, ok := val.(chan struct{}); ok {
+				// Wait until unfrozen.
+				<-unlock
+			}
+		}
+
 		pool, deadline := globalAPIConfig.getRequestsPool()
 		if pool == nil {
 			f.ServeHTTP(w, r)

cmd/notification.go

@@ -1501,6 +1501,37 @@ func (sys *NotificationSys) GetClusterMetrics(ctx context.Context) chan Metric {
 	return ch
 }
 
+// ServiceFreeze freezes all S3 API calls when 'freeze' is true,
+// 'freeze' is 'false' would resume all S3 API calls again.
+// NOTE: once a tenant is frozen either two things needs to
+// happen before resuming normal operations.
+// - Server needs to be restarted 'mc admin service restart'
+// - 'freeze' should be set to 'false' for this call
+//   to resume normal operations.
+func (sys *NotificationSys) ServiceFreeze(ctx context.Context, freeze bool) []NotificationPeerErr {
+	serviceSig := serviceUnFreeze
+	if freeze {
+		serviceSig = serviceFreeze
+	}
+	ng := WithNPeers(len(sys.peerClients))
+	for idx, client := range sys.peerClients {
+		if client == nil {
+			continue
+		}
+		client := client
+		ng.Go(GlobalContext, func() error {
+			return client.SignalService(serviceSig)
+		}, idx, *client.host)
+	}
+	nerrs := ng.Wait()
+	if freeze {
+		freezeServices()
+	} else {
+		unfreezeServices()
+	}
+	return nerrs
+}
+
 // Speedtest run GET/PUT tests at input concurrency for requested object size,
 // optionally you can extend the tests longer with time.Duration.
 func (sys *NotificationSys) Speedtest(ctx context.Context, size int, concurrent int, duration time.Duration) []SpeedtestResult {

cmd/peer-rest-common.go

@@ -18,7 +18,7 @@
 package cmd
 
 const (
-	peerRESTVersion       = "v15" // Add LoadTransitionTierConfig
+	peerRESTVersion       = "v16" // Add new ServiceSignals.
 	peerRESTVersionPrefix = SlashSeparator + peerRESTVersion
 	peerRESTPrefix        = minioReservedBucketPath + "/peer"
 	peerRESTPath          = peerRESTPrefix + peerRESTVersionPrefix

cmd/peer-rest-server.go

@@ -800,7 +800,7 @@ func (s *peerRESTServer) ServerUpdateHandler(w http.ResponseWriter, r *http.Requ
 	}
 }
 
-var errUnsupportedSignal = fmt.Errorf("unsupported signal: only restart and stop signals are supported")
+var errUnsupportedSignal = fmt.Errorf("unsupported signal")
 
 // SignalServiceHandler - signal service handler.
 func (s *peerRESTServer) SignalServiceHandler(w http.ResponseWriter, r *http.Request) {
@@ -826,6 +826,10 @@ func (s *peerRESTServer) SignalServiceHandler(w http.ResponseWriter, r *http.Req
 		globalServiceSignalCh <- signal
 	case serviceStop:
 		globalServiceSignalCh <- signal
+	case serviceFreeze:
+		freezeServices()
+	case serviceUnFreeze:
+		unfreezeServices()
 	case serviceReloadDynamic:
 		objAPI := newObjectLayerFn()
 		if objAPI == nil {
@@ -1241,6 +1245,7 @@ func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Dura
 func (s *peerRESTServer) SpeedtestHandler(w http.ResponseWriter, r *http.Request) {
 	if !s.IsValid(w, r) {
 		s.writeErrorResponse(w, errors.New("invalid request"))
+		return
 	}
 
 	objAPI := newObjectLayerFn()

cmd/service.go

@@ -31,6 +31,8 @@ const (
 	serviceRestart       serviceSignal = iota // Restarts the server.
 	serviceStop                               // Stops the server.
 	serviceReloadDynamic                      // Reload dynamic config values.
+	serviceFreeze                             // Freeze all S3 API calls.
+	serviceUnFreeze                           // Un-Freeze previously frozen S3 API calls.
 	// Add new service requests here.
 )
 
@@ -65,3 +67,57 @@ func restartProcess() error {
 	// Re-uses the same pid. This preserves the pid over multiple server-respawns.
 	return syscall.Exec(argv0, os.Args, os.Environ())
 }
+
+// Keep track of number of freeze/unfreeze calls.
+const trackFreezeCount = true
+
+// freezeServices will freeze all incoming S3 API calls.
+// For each call, unfreezeServices must be called once.
+func freezeServices() {
+	// Use atomics for globalServiceFreeze, so we can read without locking.
+	if trackFreezeCount {
+		// We need a lock since we are need the 2 atomic values to remain in sync.
+		globalServiceFreezeMu.Lock()
+		// If multiple calls, first one creates channel.
+		globalServiceFreezeCnt++
+		if globalServiceFreezeCnt == 1 {
+			globalServiceFreeze.Store(make(chan struct{}))
+		}
+		globalServiceFreezeMu.Unlock()
+	} else {
+		// If multiple calls, first one creates channel.
+		globalServiceFreeze.CompareAndSwap(nil, make(chan struct{}))
+	}
+}
+
+// unfreezeServices will unfreeze all incoming S3 API calls.
+// For each call, unfreezeServices must be called once.
+func unfreezeServices() {
+	if trackFreezeCount {
+		// We need a lock since we need the 2 atomic values to remain in sync.
+		globalServiceFreezeMu.Lock()
+		// Close when we reach 0
+		globalServiceFreezeCnt--
+		if globalServiceFreezeCnt <= 0 {
+			// Ensure we only close once.
+			if val := globalServiceFreeze.Load(); val != nil {
+				if ch, ok := val.(chan struct{}); ok {
+					globalServiceFreeze.Store(nil)
+					close(ch)
+				}
+			}
+			globalServiceFreezeCnt = 0 // Don't risk going negative.
+		}
+		globalServiceFreezeMu.Unlock()
+	} else {
+		// If multiple calls, first one closes channel.
+		if val := globalServiceFreeze.Load(); val != nil {
+			if ch, ok := val.(chan struct{}); ok {
+				// Ensure we only close once.
+				if globalServiceFreeze.CompareAndSwap(val, nil) {
+					close(ch)
+				}
+			}
+		}
+	}
+}