allow dynamically changing max_object_versions per object (#19265)

cmd/handler-api.go

@@ -55,6 +55,7 @@ type apiConfig struct {
 	gzipObjects                 bool
 	rootAccess                  bool
 	syncEvents                  bool
+	objectMaxVersions           int
 }
 
 const (
@@ -186,6 +187,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
 	t.gzipObjects = cfg.GzipObjects
 	t.rootAccess = cfg.RootAccess
 	t.syncEvents = cfg.SyncEvents
+	t.objectMaxVersions = cfg.ObjectMaxVersions
 }
 
 func (t *apiConfig) odirectEnabled() bool {
@@ -386,3 +388,15 @@ func (t *apiConfig) isSyncEventsEnabled() bool {
 
 	return t.syncEvents
 }
+
+func (t *apiConfig) getObjectMaxVersions() int {
+	t.mu.RLock()
+	defer t.mu.RUnlock()
+
+	if t.objectMaxVersions <= 0 {
+		// defaults to 'maxObjectVersions' when unset.
+		return maxObjectVersions
+	}
+
+	return t.objectMaxVersions
+}

cmd/xl-storage-format-v2.go

@@ -25,7 +25,6 @@ import (
 	"fmt"
 	"io"
 	"sort"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -38,31 +37,12 @@ import (
 	"github.com/minio/minio/internal/config/storageclass"
 	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
-	"github.com/minio/pkg/v2/env"
 	"github.com/tinylib/msgp/msgp"
 )
 
 // Reject creating new versions when a single object is cross maxObjectVersions
 var maxObjectVersions = 10000
 
-func init() {
-	v := env.Get("_MINIO_OBJECT_MAX_VERSIONS", "")
-	if v != "" {
-		maxv, err := strconv.Atoi(v)
-		if err != nil {
-			logger.Info("invalid _MINIO_OBJECT_MAX_VERSIONS value: %s, defaulting to '10000'", v)
-			maxObjectVersions = 10000
-		} else {
-			if maxv < 10 {
-				logger.Info("invalid _MINIO_OBJECT_MAX_VERSIONS value: %s, minimum allowed is '10' defaulting to '10000'", v)
-				maxObjectVersions = 10000
-			} else {
-				maxObjectVersions = maxv
-			}
-		}
-	}
-}
-
 var (
 	// XL header specifies the format
 	xlHeader = [4]byte{'X', 'L', '2', ' '}
@@ -1112,7 +1092,7 @@ func (x *xlMetaV2) addVersion(ver xlMetaV2Version) error {
 	}
 
 	// returns error if we have exceeded maxObjectVersions
-	if len(x.versions)+1 > maxObjectVersions {
+	if len(x.versions)+1 > globalAPIConfig.getObjectMaxVersions() {
 		return errMaxVersionsExceeded
 	}
 

internal/config/api/api.go

@@ -49,6 +49,7 @@ const (
 	apiGzipObjects                 = "gzip_objects"
 	apiRootAccess                  = "root_access"
 	apiSyncEvents                  = "sync_events"
+	apiObjectMaxVersions           = "object_max_versions"
 
 	EnvAPIRequestsMax                 = "MINIO_API_REQUESTS_MAX"
 	EnvAPIRequestsDeadline            = "MINIO_API_REQUESTS_DEADLINE"
@@ -69,6 +70,8 @@ const (
 	EnvAPIGzipObjects                 = "MINIO_API_GZIP_OBJECTS"
 	EnvAPIRootAccess                  = "MINIO_API_ROOT_ACCESS" // default config.EnableOn
 	EnvAPISyncEvents                  = "MINIO_API_SYNC_EVENTS" // default "off"
+	EnvAPIObjectMaxVersions           = "MINIO_API_OBJECT_MAX_VERSIONS"
+	EnvAPIObjectMaxVersionsLegacy     = "_MINIO_OBJECT_MAX_VERSIONS"
 )
 
 // Deprecated key and ENVs
@@ -150,6 +153,10 @@ var (
 			Key:   apiSyncEvents,
 			Value: config.EnableOff,
 		},
+		config.KV{
+			Key:   apiObjectMaxVersions,
+			Value: "10000",
+		},
 	}
 )
 
@@ -172,6 +179,7 @@ type Config struct {
 	GzipObjects                 bool          `json:"gzip_objects"`
 	RootAccess                  bool          `json:"root_access"`
 	SyncEvents                  bool          `json:"sync_events"`
+	ObjectMaxVersions           int           `json:"object_max_versions"`
 }
 
 // UnmarshalJSON - Validate SS and RRS parity when unmarshalling JSON.
@@ -307,5 +315,18 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 
 	cfg.SyncEvents = env.Get(EnvAPISyncEvents, kvs.Get(apiSyncEvents)) == config.EnableOn
 
+	maxVerStr := env.Get(EnvAPIObjectMaxVersions, "")
+	if maxVerStr == "" {
+		maxVerStr = env.Get(EnvAPIObjectMaxVersionsLegacy, kvs.GetWithDefault(apiObjectMaxVersions, DefaultKVS))
+	}
+	maxVersions, err := strconv.Atoi(maxVerStr)
+	if err != nil {
+		return cfg, err
+	}
+	if maxVersions <= 0 {
+		return cfg, fmt.Errorf("invalid object max versions value: %v", maxVersions)
+	}
+	cfg.ObjectMaxVersions = maxVersions
+
 	return cfg, nil
 }

internal/config/api/help.go

@@ -116,5 +116,11 @@ var (
 			Optional:    true,
 			Type:        "boolean",
 		},
+		config.HelpKV{
+			Key:         apiObjectMaxVersions,
+			Description: "set max allowed number of versions per object" + defaultHelpPostfix(apiObjectMaxVersions),
+			Optional:    true,
+			Type:        "number",
+		},
 	}
 )