fix: move list quorum ENV to config (#10804)

cmd/config/api/api.go

@@ -34,12 +34,15 @@ const (
 	apiClusterDeadline         = "cluster_deadline"
 	apiCorsAllowOrigin         = "cors_allow_origin"
 	apiRemoteTransportDeadline = "remote_transport_deadline"
+	apiListQuorum              = "list_quorum"
 
 	EnvAPIRequestsMax             = "MINIO_API_REQUESTS_MAX"
 	EnvAPIRequestsDeadline        = "MINIO_API_REQUESTS_DEADLINE"
 	EnvAPIClusterDeadline         = "MINIO_API_CLUSTER_DEADLINE"
 	EnvAPICorsAllowOrigin         = "MINIO_API_CORS_ALLOW_ORIGIN"
 	EnvAPIRemoteTransportDeadline = "MINIO_API_REMOTE_TRANSPORT_DEADLINE"
+	EnvAPIListQuorum              = "MINIO_API_LIST_QUORUM"
+	EnvAPISecureCiphers           = "MINIO_API_SECURE_CIPHERS"
 )
 
 // Deprecated key and ENVs
@@ -71,6 +74,10 @@ var (
 			Key:   apiRemoteTransportDeadline,
 			Value: "2h",
 		},
+		config.KV{
+			Key:   apiListQuorum,
+			Value: "optimal",
+		},
 	}
 )
 
@@ -81,6 +88,7 @@ type Config struct {
 	ClusterDeadline         time.Duration `json:"cluster_deadline"`
 	CorsAllowOrigin         []string      `json:"cors_allow_origin"`
 	RemoteTransportDeadline time.Duration `json:"remote_transport_deadline"`
+	ListQuorum              string        `json:"list_strict_quorum"`
 }
 
 // UnmarshalJSON - Validate SS and RRS parity when unmarshalling JSON.
@@ -94,6 +102,24 @@ func (sCfg *Config) UnmarshalJSON(data []byte) error {
 	return json.Unmarshal(data, &aux)
 }
 
+// GetListQuorum interprets list quorum values and returns appropriate
+// acceptable quorum expected for list operations
+func (sCfg Config) GetListQuorum() int {
+	switch sCfg.ListQuorum {
+	case "optimal":
+		return 3
+	case "reduced":
+		return 2
+	case "disk":
+		// smallest possible value, generally meant for testing.
+		return 1
+	case "strict":
+		return -1
+	}
+	// Defaults to 3 drives per set.
+	return 3
+}
+
 // LookupConfig - lookup api config and override with valid environment settings if any.
 func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 	// remove this since we have removed this already.
@@ -130,11 +156,19 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 		return cfg, err
 	}
 
+	listQuorum := env.Get(EnvAPIListQuorum, kvs.Get(apiListQuorum))
+	switch listQuorum {
+	case "strict", "optimal", "reduced", "disk":
+	default:
+		return cfg, errors.New("invalid value for list strict quorum")
+	}
+
 	return Config{
 		RequestsMax:             requestsMax,
 		RequestsDeadline:        requestsDeadline,
 		ClusterDeadline:         clusterDeadline,
 		CorsAllowOrigin:         corsAllowOrigin,
 		RemoteTransportDeadline: remoteTransportDeadline,
+		ListQuorum:              listQuorum,
 	}, nil
 }

cmd/config/storageclass/help.go

@@ -33,6 +33,12 @@ var (
 			Optional:    true,
 			Type:        "string",
 		},
+		config.HelpKV{
+			Key:         ClassDMA,
+			Description: `enable O_DIRECT for both read and write, defaults to "write" e.g. "read+write"`,
+			Optional:    true,
+			Type:        "string",
+		},
 		config.HelpKV{
 			Key:         config.Comment,
 			Description: config.DefaultComment,

cmd/config/storageclass/storage-class.go

@@ -36,9 +36,9 @@ const (
 	// DMA storage class
 	DMA = "DMA"
 
-	// Valid values are "write" and "read-write"
+	// Valid values are "write" and "read+write"
 	DMAWrite     = "write"
-	DMAReadWrite = "read-write"
+	DMAReadWrite = "read+write"
 )
 
 // Standard constats for config info storage class

cmd/erasure-server-sets.go

@@ -1340,10 +1340,12 @@ func (z *erasureServerSets) Walk(ctx context.Context, bucket, prefix string, res
 
 	serverSetsListTolerancePerSet := make([]int, 0, len(z.serverSets))
 	for _, zone := range z.serverSets {
-		if zone.listTolerancePerSet == -1 {
+		quorum := globalAPIConfig.getListQuorum()
+		switch quorum {
+		case -1:
 			serverSetsListTolerancePerSet = append(serverSetsListTolerancePerSet, zone.setDriveCount/2)
-		} else {
+		default:
-			serverSetsListTolerancePerSet = append(serverSetsListTolerancePerSet, zone.listTolerancePerSet-2)
+			serverSetsListTolerancePerSet = append(serverSetsListTolerancePerSet, quorum)
 		}
 	}
 

cmd/erasure-sets.go

@@ -31,11 +31,9 @@ import (
 	"github.com/dchest/siphash"
 	"github.com/google/uuid"
 	"github.com/minio/minio-go/v7/pkg/tags"
-	"github.com/minio/minio/cmd/config"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/bpool"
 	"github.com/minio/minio/pkg/dsync"
-	"github.com/minio/minio/pkg/env"
 	"github.com/minio/minio/pkg/madmin"
 	"github.com/minio/minio/pkg/sync/errgroup"
 )
@@ -81,7 +79,6 @@ type erasureSets struct {
 
 	// Total number of sets and the number of disks per set.
 	setCount, setDriveCount int
-	listTolerancePerSet     int
 
 	disksConnectEvent chan diskConnectInfo
 
@@ -345,31 +342,24 @@ func newErasureSets(ctx context.Context, endpoints Endpoints, storageDisks []Sto
 
 	endpointStrings := make([]string, len(endpoints))
 
-	listTolerancePerSet := 3
-	// By default this is off
-	if env.Get("MINIO_API_LIST_STRICT_QUORUM", config.EnableOn) == config.EnableOn {
-		listTolerancePerSet = -1
-	}
-
 	// Initialize the erasure sets instance.
 	s := &erasureSets{
-		sets:                make([]*erasureObjects, setCount),
+		sets:              make([]*erasureObjects, setCount),
-		erasureDisks:        make([][]StorageAPI, setCount),
+		erasureDisks:      make([][]StorageAPI, setCount),
-		erasureLockers:      make([][]dsync.NetLocker, setCount),
+		erasureLockers:    make([][]dsync.NetLocker, setCount),
-		erasureLockOwner:    GetLocalPeer(globalEndpoints),
+		erasureLockOwner:  GetLocalPeer(globalEndpoints),
-		endpoints:           endpoints,
+		endpoints:         endpoints,
-		endpointStrings:     endpointStrings,
+		endpointStrings:   endpointStrings,
-		setCount:            setCount,
+		setCount:          setCount,
-		setDriveCount:       setDriveCount,
+		setDriveCount:     setDriveCount,
-		listTolerancePerSet: listTolerancePerSet,
+		format:            format,
-		format:              format,
+		disksConnectEvent: make(chan diskConnectInfo),
-		disksConnectEvent:   make(chan diskConnectInfo),
+		distributionAlgo:  format.Erasure.DistributionAlgo,
-		distributionAlgo:    format.Erasure.DistributionAlgo,
+		deploymentID:      uuid.MustParse(format.ID),
-		deploymentID:        uuid.MustParse(format.ID),
+		pool:              NewMergeWalkPool(globalMergeLookupTimeout),
-		pool:                NewMergeWalkPool(globalMergeLookupTimeout),
+		poolSplunk:        NewMergeWalkPool(globalMergeLookupTimeout),
-		poolSplunk:          NewMergeWalkPool(globalMergeLookupTimeout),
+		poolVersions:      NewMergeWalkVersionsPool(globalMergeLookupTimeout),
-		poolVersions:        NewMergeWalkVersionsPool(globalMergeLookupTimeout),
+		mrfOperations:     make(map[healSource]int),
-		mrfOperations:       make(map[healSource]int),
 	}
 
 	mutex := newNSLock(globalIsDistErasure)

cmd/globals.go

@@ -160,7 +160,7 @@ var (
 	globalBucketTargetSys    *BucketTargetSys
 	// globalAPIConfig controls S3 API requests throttling,
 	// healthcheck readiness deadlines and cors settings.
-	globalAPIConfig apiConfig
+	globalAPIConfig = apiConfig{listQuorum: 3}
 
 	globalStorageClass storageclass.Config
 	globalLDAPConfig   xldap.Config

cmd/handler-api.go

@@ -32,6 +32,7 @@ type apiConfig struct {
 	requestsDeadline time.Duration
 	requestsPool     chan struct{}
 	clusterDeadline  time.Duration
+	listQuorum       int
 	corsAllowOrigins []string
 }
 
@@ -63,6 +64,14 @@ func (t *apiConfig) init(cfg api.Config, setDriveCount int) {
 
 	t.requestsPool = make(chan struct{}, apiRequestsMaxPerNode)
 	t.requestsDeadline = cfg.RequestsDeadline
+	t.listQuorum = cfg.GetListQuorum()
+}
+
+func (t *apiConfig) getListQuorum() int {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	return t.listQuorum
 }
 
 func (t *apiConfig) getCorsAllowOrigins() []string {

cmd/http/server.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/minio/minio-go/v7/pkg/set"
 	"github.com/minio/minio/cmd/config"
+	"github.com/minio/minio/cmd/config/api"
 	"github.com/minio/minio/pkg/certs"
 	"github.com/minio/minio/pkg/env"
 )
@@ -180,13 +181,9 @@ var secureCipherSuites = []uint16{
 // Go only provides constant-time implementations of Curve25519 and NIST P-256 curve.
 var secureCurves = []tls.CurveID{tls.X25519, tls.CurveP256}
 
-const (
-	enableSecureCiphersEnv = "MINIO_API_SECURE_CIPHERS"
-)
-
 // NewServer - creates new HTTP server using given arguments.
 func NewServer(addrs []string, handler http.Handler, getCert certs.GetCertificateFunc) *Server {
-	secureCiphers := env.Get(enableSecureCiphersEnv, config.EnableOn) == config.EnableOn
+	secureCiphers := env.Get(api.EnvAPISecureCiphers, config.EnableOn) == config.EnableOn
 
 	var tlsConfig *tls.Config
 	if getCert != nil {

cmd/metacache-server-sets.go

@@ -23,9 +23,7 @@ import (
 	"path"
 	"sync"
 
-	"github.com/minio/minio/cmd/config"
 	"github.com/minio/minio/cmd/logger"
-	"github.com/minio/minio/pkg/env"
 )
 
 // listPath will return the requested entries.
@@ -119,20 +117,7 @@ func (z *erasureServerSets) listPath(ctx context.Context, o listPathOptions) (en
 	}
 
 	if o.AskDisks == 0 {
-		switch env.Get("MINIO_API_LIST_STRICT_QUORUM", config.EnableOff) {
+		o.AskDisks = globalAPIConfig.getListQuorum()
-		case config.EnableOn:
-			// If strict, ask at least 50%.
-			o.AskDisks = -1
-		case "reduced":
-			// Reduced safety.
-			o.AskDisks = 2
-		case "disk":
-			// Ask single disk.
-			o.AskDisks = 1
-		default:
-			// By default asks at max 3 disks.
-			o.AskDisks = 3
-		}
 	}
 
 	var mu sync.Mutex