scanner: Add a config to disable short sleep between objects scan (#18734)

Add a hidden configuration under the scanner sub section to configure if
the scanner should sleep between two objects scan. The configuration has
only effect when there is no drive activity related to s3 requests or
healing.

By default, the code will keep the current behavior which is doing
sleep between objects.

To forcefully enable the full scan speed in idle mode, you can do this:

   `mc admin config set myminio scanner idle_speed=full`

cmd/config-current.go

@@ -600,6 +600,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 			return fmt.Errorf("Unable to apply scanner config: %w", err)
 		}
 		// update dynamic scanner values.
+		scannerIdleMode.Store(scannerCfg.IdleMode)
 		scannerCycle.Store(scannerCfg.Cycle)
 		logger.LogIf(ctx, scannerSleeper.Update(scannerCfg.Delay, scannerCfg.MaxWait))
 	case config.LoggerWebhookSubSys:

cmd/data-scanner.go

@@ -67,6 +67,7 @@ var (
 	// Sleeper values are updated when config is loaded.
 	scannerSleeper  = newDynamicSleeper(2, time.Second, true) // Keep defaults same as config defaults
 	scannerCycle    = uatomic.NewDuration(dataScannerStartDelay)
+	scannerIdleMode = uatomic.NewInt32(0) // default is throttled when idle
 )
 
 // initDataScanner will start the scanner in the background.

cmd/xl-storage-disk-id-check.go

@@ -286,8 +286,7 @@ func (p *xlStorageDiskIDCheck) NSScanner(ctx context.Context, cache dataUsageCac
 	}
 
 	weSleep := func() bool {
-		// Entire queue is full, so we sleep.
-		return cap(p.health.tokens) == len(p.health.tokens)
+		return scannerIdleMode.Load() == 0
 	}
 
 	return p.storage.NSScanner(ctx, cache, updates, scanMode, weSleep)

internal/config/api/api.go

@@ -132,7 +132,7 @@ var (
 		config.KV{
 			Key:           apiDisableODirect,
 			Value:         "",
-			Deprecated: true,
+			HiddenIfEmpty: true,
 		},
 		config.KV{
 			Key:   apiODirect,

internal/config/config.go

@@ -274,7 +274,7 @@ type KV struct {
 	Key   string `json:"key"`
 	Value string `json:"value"`
 
-	Deprecated bool `json:"-"`
+	HiddenIfEmpty bool `json:"-"`
 }
 
 func (kv KV) String() string {
@@ -1447,7 +1447,7 @@ func (cs *SubsysInfo) WriteTo(b *strings.Builder, off bool) {
 			continue
 		}
 		// Ignore empty and deprecated values
-		if dkv.Deprecated && kv.Value == "" {
+		if dkv.HiddenIfEmpty && kv.Value == "" {
 			continue
 		}
 		// Do not need to print if state is on

internal/config/scanner/scanner.go

@@ -31,6 +31,9 @@ const (
 	Speed    = "speed"
 	EnvSpeed = "MINIO_SCANNER_SPEED"
 
+	IdleSpeed    = "idle_speed"
+	EnvIdleSpeed = "MINIO_SCANNER_IDLE_SPEED"
+
 	// All below are deprecated in October 2022 and
 	// replaced them with a single speed parameter
 	Delay            = "delay"
@@ -47,6 +50,8 @@ const (
 type Config struct {
 	// Delay is the sleep multiplier.
 	Delay float64 `json:"delay"`
+	// Behavior of the scanner when there is no other parallel S3 requests
+	IdleMode int32 // 0 => throttling, 1 => full speed
 	// MaxWait is maximum wait time between operations
 	MaxWait time.Duration
 	// Cycle is the time.Duration between each scanner cycles
@@ -59,23 +64,28 @@ var DefaultKVS = config.KVS{
 		Key:   Speed,
 		Value: "default",
 	},
+	config.KV{
+		Key:           IdleSpeed,
+		Value:         "",
+		HiddenIfEmpty: true,
+	},
 	// Deprecated Oct 2022
 	config.KV{
 		Key:           Delay,
 		Value:         "",
-		Deprecated: true,
+		HiddenIfEmpty: true,
 	},
 	// Deprecated Oct 2022
 	config.KV{
 		Key:           MaxWait,
 		Value:         "",
-		Deprecated: true,
+		HiddenIfEmpty: true,
 	},
 	// Deprecated Oct 2022
 	config.KV{
 		Key:           Cycle,
 		Value:         "",
-		Deprecated: true,
+		HiddenIfEmpty: true,
 	},
 }
 
@@ -107,6 +117,16 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 	default:
 		return cfg, fmt.Errorf("unknown '%s' value", speed)
 	}
+
+	switch idleSpeed := env.Get(EnvIdleSpeed, kvs.GetWithDefault(IdleSpeed, DefaultKVS)); idleSpeed {
+	case "", "throttled": // Empty is the default mode;
+		cfg.IdleMode = 0
+	case "full":
+		cfg.IdleMode = 1
+	default:
+		return cfg, fmt.Errorf("unknown value: '%s'", idleSpeed)
+	}
+
 	return
 }