mirror of
https://github.com/minio/minio.git
synced 2024-12-24 06:05:55 -05:00
simplify further bucket configuration properly (#9650)
This PR is a continuation from #9586, now the entire parsing logic is fully merged into bucket metadata sub-system, simplify the quota API further by reducing the remove quota handler implementation.
This commit is contained in:
parent
0cc2ed04f5
commit
6656fa3066
@ -68,21 +68,9 @@ func prepareAdminXLTestBed(ctx context.Context) (*adminXLTestBed, error) {
|
||||
|
||||
globalEndpoints = mustGetZoneEndpoints(xlDirs...)
|
||||
|
||||
globalConfigSys = NewConfigSys()
|
||||
newAllSubsystems()
|
||||
|
||||
globalIAMSys = NewIAMSys()
|
||||
globalIAMSys.Init(ctx, objLayer)
|
||||
|
||||
buckets, err := objLayer.ListBuckets(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
globalPolicySys = NewPolicySys()
|
||||
globalPolicySys.Init(buckets, objLayer)
|
||||
|
||||
globalNotificationSys = NewNotificationSys(globalEndpoints)
|
||||
globalNotificationSys.Init(buckets, objLayer)
|
||||
initAllSubsystems(objLayer)
|
||||
|
||||
// Setup admin mgmt REST API handlers.
|
||||
adminRouter := mux.NewRouter()
|
||||
|
@ -17,7 +17,7 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"encoding/json"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
|
||||
@ -101,46 +101,18 @@ func (a adminAPIHandlers) GetBucketQuotaConfigHandler(w http.ResponseWriter, r *
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketQuotaConfigFile)
|
||||
config, err := globalBucketMetadataSys.GetQuotaConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, BucketQuotaConfigNotFound{Bucket: bucket}), r.URL)
|
||||
} else {
|
||||
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
||||
}
|
||||
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
||||
return
|
||||
}
|
||||
|
||||
// Write success response.
|
||||
writeSuccessResponseJSON(w, configData)
|
||||
}
|
||||
|
||||
// RemoveBucketQuotaConfigHandler - removes Bucket quota configuration.
|
||||
// ----------
|
||||
// Removes quota configuration on the specified bucket.
|
||||
func (a adminAPIHandlers) RemoveBucketQuotaConfigHandler(w http.ResponseWriter, r *http.Request) {
|
||||
ctx := newContext(r, w, "RemoveBucketQuotaConfig")
|
||||
|
||||
defer logger.AuditLog(w, r, "RemoveBucketQuotaConfig", mustGetClaimsFromToken(r))
|
||||
|
||||
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.SetBucketQuotaAdminAction)
|
||||
if objectAPI == nil {
|
||||
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
||||
return
|
||||
}
|
||||
vars := mux.Vars(r)
|
||||
bucket := vars["bucket"]
|
||||
|
||||
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
||||
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
||||
return
|
||||
}
|
||||
|
||||
if err := globalBucketMetadataSys.Update(bucket, bucketQuotaConfigFile, nil); err != nil {
|
||||
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
||||
return
|
||||
}
|
||||
|
||||
// Write success response.
|
||||
writeSuccessNoContent(w)
|
||||
}
|
||||
|
@ -177,9 +177,6 @@ func registerAdminRouter(router *mux.Router, enableConfigOps, enableIAMOps, enab
|
||||
// PutBucketQuotaConfig
|
||||
adminRouter.Methods(http.MethodPut).Path(adminVersion+"/set-bucket-quota").HandlerFunc(
|
||||
httpTraceHdrs(adminAPI.PutBucketQuotaConfigHandler)).Queries("bucket", "{bucket:.*}")
|
||||
// RemoveBucketQuotaConfig
|
||||
adminRouter.Methods(http.MethodDelete).Path(adminVersion+"/remove-bucket-quota").HandlerFunc(
|
||||
httpTraceHdrs(adminAPI.RemoveBucketQuotaConfigHandler)).Queries("bucket", "{bucket:.*}")
|
||||
}
|
||||
|
||||
// -- Top APIs --
|
||||
|
@ -18,7 +18,6 @@ package cmd
|
||||
|
||||
import (
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@ -128,11 +127,14 @@ func (api objectAPIHandlers) GetBucketEncryptionHandler(w http.ResponseWriter, r
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketSSEConfig)
|
||||
config, err := globalBucketMetadataSys.GetSSEConfig(bucket)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
err = BucketSSEConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
@ -17,7 +17,6 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"io"
|
||||
|
||||
@ -25,56 +24,15 @@ import (
|
||||
)
|
||||
|
||||
// BucketSSEConfigSys - in-memory cache of bucket encryption config
|
||||
type BucketSSEConfigSys struct {
|
||||
bucketSSEConfigMap map[string]*bucketsse.BucketSSEConfig
|
||||
}
|
||||
type BucketSSEConfigSys struct{}
|
||||
|
||||
// NewBucketSSEConfigSys - Creates an empty in-memory bucket encryption configuration cache
|
||||
func NewBucketSSEConfigSys() *BucketSSEConfigSys {
|
||||
return &BucketSSEConfigSys{
|
||||
bucketSSEConfigMap: make(map[string]*bucketsse.BucketSSEConfig),
|
||||
}
|
||||
}
|
||||
|
||||
// load - Loads the bucket encryption configuration for the given list of buckets
|
||||
func (sys *BucketSSEConfigSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketSSEConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
config, err := bucketsse.ParseBucketSSEConfig(bytes.NewReader(configData))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
sys.bucketSSEConfigMap[bucket.Name] = config
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Init - Initializes in-memory bucket encryption config cache for the given list of buckets
|
||||
func (sys *BucketSSEConfigSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
if objAPI == nil {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
|
||||
// We don't cache bucket encryption config in gateway mode, nothing to do.
|
||||
if globalIsGateway {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Load bucket encryption config cache once during boot.
|
||||
return sys.load(buckets, objAPI)
|
||||
return &BucketSSEConfigSys{}
|
||||
}
|
||||
|
||||
// Get - gets bucket encryption config for the given bucket.
|
||||
func (sys *BucketSSEConfigSys) Get(bucket string) (config *bucketsse.BucketSSEConfig, err error) {
|
||||
func (sys *BucketSSEConfigSys) Get(bucket string) (*bucketsse.BucketSSEConfig, error) {
|
||||
if globalIsGateway {
|
||||
objAPI := newObjectLayerWithoutSafeModeFn()
|
||||
if objAPI == nil {
|
||||
@ -84,18 +42,7 @@ func (sys *BucketSSEConfigSys) Get(bucket string) (config *bucketsse.BucketSSECo
|
||||
return nil, BucketSSEConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
|
||||
config, ok := sys.bucketSSEConfigMap[bucket]
|
||||
if !ok {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketSSEConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketSSEConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return bucketsse.ParseBucketSSEConfig(bytes.NewReader(configData))
|
||||
}
|
||||
return config, nil
|
||||
return globalBucketMetadataSys.GetSSEConfig(bucket)
|
||||
}
|
||||
|
||||
// validateBucketSSEConfig parses bucket encryption configuration and validates if it is supported by MinIO.
|
||||
|
@ -19,7 +19,6 @@ package cmd
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@ -1069,11 +1068,14 @@ func (api objectAPIHandlers) GetBucketObjectLockConfigHandler(w http.ResponseWri
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, objectLockConfig)
|
||||
config, err := globalBucketMetadataSys.GetObjectLockConfig(bucket)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
err = BucketObjectLockConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
@ -1148,11 +1150,14 @@ func (api objectAPIHandlers) GetBucketTaggingHandler(w http.ResponseWriter, r *h
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketTaggingConfigFile)
|
||||
config, err := globalBucketMetadataSys.GetTaggingConfig(bucket)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
err = BucketTaggingNotFound{Bucket: bucket}
|
||||
}
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
@ -18,7 +18,6 @@ package cmd
|
||||
|
||||
import (
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
@ -120,11 +119,14 @@ func (api objectAPIHandlers) GetBucketLifecycleHandler(w http.ResponseWriter, r
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketLifecycleConfig)
|
||||
config, err := globalBucketMetadataSys.GetLifecycleConfig(bucket)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
err = BucketLifecycleNotFound{Bucket: bucket}
|
||||
}
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
@ -17,10 +17,6 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
|
||||
"github.com/minio/minio/cmd/logger"
|
||||
"github.com/minio/minio/pkg/bucket/lifecycle"
|
||||
)
|
||||
|
||||
@ -31,9 +27,7 @@ const (
|
||||
)
|
||||
|
||||
// LifecycleSys - Bucket lifecycle subsystem.
|
||||
type LifecycleSys struct {
|
||||
bucketLifecycleMap map[string]*lifecycle.Lifecycle
|
||||
}
|
||||
type LifecycleSys struct{}
|
||||
|
||||
// Get - gets lifecycle config associated to a given bucket name.
|
||||
func (sys *LifecycleSys) Get(bucketName string) (lc *lifecycle.Lifecycle, err error) {
|
||||
@ -46,61 +40,10 @@ func (sys *LifecycleSys) Get(bucketName string) (lc *lifecycle.Lifecycle, err er
|
||||
return nil, BucketLifecycleNotFound{Bucket: bucketName}
|
||||
}
|
||||
|
||||
lc, ok := sys.bucketLifecycleMap[bucketName]
|
||||
if !ok {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucketName, bucketLifecycleConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketLifecycleNotFound{Bucket: bucketName}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return lifecycle.ParseLifecycleConfig(bytes.NewReader(configData))
|
||||
}
|
||||
return lc, nil
|
||||
return globalBucketMetadataSys.GetLifecycleConfig(bucketName)
|
||||
}
|
||||
|
||||
// NewLifecycleSys - creates new lifecycle system.
|
||||
func NewLifecycleSys() *LifecycleSys {
|
||||
return &LifecycleSys{
|
||||
bucketLifecycleMap: make(map[string]*lifecycle.Lifecycle),
|
||||
}
|
||||
}
|
||||
|
||||
// Init - initializes lifecycle system from lifecycle.xml of all buckets.
|
||||
func (sys *LifecycleSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
if objAPI == nil {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
|
||||
// In gateway mode, we always fetch the bucket lifecycle configuration from the gateway backend.
|
||||
// So, this is a no-op for gateway servers.
|
||||
if globalIsGateway {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Load LifecycleSys once during boot.
|
||||
return sys.load(buckets, objAPI)
|
||||
}
|
||||
|
||||
// Loads lifecycle policies for all buckets into LifecycleSys.
|
||||
func (sys *LifecycleSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketLifecycleConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
config, err := lifecycle.ParseLifecycleConfig(bytes.NewReader(configData))
|
||||
if err != nil {
|
||||
logger.LogIf(GlobalContext, err)
|
||||
continue
|
||||
}
|
||||
|
||||
sys.bucketLifecycleMap[bucket.Name] = config
|
||||
}
|
||||
return nil
|
||||
return &LifecycleSys{}
|
||||
}
|
||||
|
@ -19,19 +19,23 @@ package cmd
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/minio/minio-go/v6/pkg/tags"
|
||||
bucketsse "github.com/minio/minio/pkg/bucket/encryption"
|
||||
"github.com/minio/minio/pkg/bucket/lifecycle"
|
||||
objectlock "github.com/minio/minio/pkg/bucket/object/lock"
|
||||
"github.com/minio/minio/pkg/bucket/policy"
|
||||
"github.com/minio/minio/pkg/event"
|
||||
"github.com/minio/minio/pkg/madmin"
|
||||
)
|
||||
|
||||
// BucketMetadataSys captures all bucket metadata for a given cluster.
|
||||
type BucketMetadataSys struct {
|
||||
sync.RWMutex
|
||||
metadataMap map[string]BucketMetadata
|
||||
|
||||
// Do fallback to old config when unable to find a bucket config
|
||||
fallback bool
|
||||
}
|
||||
|
||||
// Remove bucket metadata from memory.
|
||||
@ -84,13 +88,9 @@ func (sys *BucketMetadataSys) Update(bucket string, configFile string, configDat
|
||||
return errInvalidArgument
|
||||
}
|
||||
|
||||
defer globalNotificationSys.LoadBucketMetadata(GlobalContext, bucket)
|
||||
|
||||
sys.Lock()
|
||||
meta, ok := sys.metadataMap[bucket]
|
||||
sys.Unlock()
|
||||
if !ok {
|
||||
return BucketNotFound{Bucket: bucket}
|
||||
meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch configFile {
|
||||
@ -116,9 +116,8 @@ func (sys *BucketMetadataSys) Update(bucket string, configFile string, configDat
|
||||
return err
|
||||
}
|
||||
|
||||
sys.Lock()
|
||||
sys.metadataMap[bucket] = meta
|
||||
sys.Unlock()
|
||||
sys.Set(bucket, meta)
|
||||
globalNotificationSys.LoadBucketMetadata(GlobalContext, bucket)
|
||||
|
||||
return nil
|
||||
}
|
||||
@ -143,62 +142,134 @@ func (sys *BucketMetadataSys) Get(bucket string) (BucketMetadata, error) {
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// GetTaggingConfig returns configured tagging config
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetTaggingConfig(bucket string) (*tags.Tags, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketTaggingNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if meta.taggingConfig == nil {
|
||||
return nil, BucketTaggingNotFound{Bucket: bucket}
|
||||
}
|
||||
return meta.taggingConfig, nil
|
||||
}
|
||||
|
||||
// GetObjectLockConfig returns configured object lock config
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetObjectLockConfig(bucket string) (*objectlock.Config, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketObjectLockConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if meta.objectLockConfig == nil {
|
||||
return nil, BucketObjectLockConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
return meta.objectLockConfig, nil
|
||||
}
|
||||
|
||||
// GetLifecycleConfig returns configured lifecycle config
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetLifecycleConfig(bucket string) (*lifecycle.Lifecycle, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketLifecycleNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if meta.lifecycleConfig == nil {
|
||||
return nil, BucketLifecycleNotFound{Bucket: bucket}
|
||||
}
|
||||
return meta.lifecycleConfig, nil
|
||||
}
|
||||
|
||||
// GetNotificationConfig returns configured notification config
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetNotificationConfig(bucket string) (*event.Config, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return meta.notificationConfig, nil
|
||||
}
|
||||
|
||||
// GetSSEConfig returns configured SSE config
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetSSEConfig(bucket string) (*bucketsse.BucketSSEConfig, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketSSEConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if meta.sseConfig == nil {
|
||||
return nil, BucketSSEConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
return meta.sseConfig, nil
|
||||
}
|
||||
|
||||
// GetPolicyConfig returns configured bucket policy
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetPolicyConfig(bucket string) (*policy.Policy, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return nil, BucketPolicyNotFound{Bucket: bucket}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if meta.policyConfig == nil {
|
||||
return nil, BucketPolicyNotFound{Bucket: bucket}
|
||||
}
|
||||
return meta.policyConfig, nil
|
||||
}
|
||||
|
||||
// GetQuotaConfig returns configured bucket quota
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetQuotaConfig(bucket string) (*madmin.BucketQuota, error) {
|
||||
meta, err := sys.GetConfig(bucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return meta.quotaConfig, nil
|
||||
}
|
||||
|
||||
// GetConfig returns a specific configuration from the bucket metadata.
|
||||
// The returned data should be copied before being modified.
|
||||
func (sys *BucketMetadataSys) GetConfig(bucket string, configFile string) ([]byte, error) {
|
||||
// The returned object may not be modified.
|
||||
func (sys *BucketMetadataSys) GetConfig(bucket string) (BucketMetadata, error) {
|
||||
objAPI := newObjectLayerWithoutSafeModeFn()
|
||||
if objAPI == nil {
|
||||
return newBucketMetadata(bucket), errServerNotInitialized
|
||||
}
|
||||
|
||||
if globalIsGateway {
|
||||
return nil, NotImplemented{}
|
||||
return newBucketMetadata(bucket), NotImplemented{}
|
||||
}
|
||||
|
||||
if bucket == minioMetaBucket {
|
||||
return nil, errInvalidArgument
|
||||
return newBucketMetadata(bucket), errInvalidArgument
|
||||
}
|
||||
|
||||
sys.RLock()
|
||||
defer sys.RUnlock()
|
||||
|
||||
sys.Lock()
|
||||
defer sys.Unlock()
|
||||
meta, ok := sys.metadataMap[bucket]
|
||||
if !ok {
|
||||
if !sys.fallback {
|
||||
return nil, errConfigNotFound
|
||||
}
|
||||
objAPI := newObjectLayerWithoutSafeModeFn()
|
||||
if objAPI == nil {
|
||||
return nil, errServerNotInitialized
|
||||
}
|
||||
var err error
|
||||
meta, err = loadBucketMetadata(GlobalContext, objAPI, bucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sys.metadataMap[bucket] = meta
|
||||
if ok {
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
var configData []byte
|
||||
switch configFile {
|
||||
case bucketPolicyConfig:
|
||||
configData = meta.PolicyConfigJSON
|
||||
case bucketNotificationConfig:
|
||||
configData = meta.NotificationXML
|
||||
case bucketLifecycleConfig:
|
||||
configData = meta.LifecycleConfigXML
|
||||
case bucketSSEConfig:
|
||||
configData = meta.EncryptionConfigXML
|
||||
case bucketTaggingConfigFile:
|
||||
configData = meta.TaggingConfigXML
|
||||
case objectLockConfig:
|
||||
configData = meta.ObjectLockConfigurationXML
|
||||
case bucketQuotaConfigFile:
|
||||
configData = meta.QuotaConfigJSON
|
||||
default:
|
||||
return nil, fmt.Errorf("Unknown bucket %s metadata update requested %s", bucket, configFile)
|
||||
meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
|
||||
if err != nil {
|
||||
return meta, err
|
||||
}
|
||||
|
||||
if len(configData) == 0 {
|
||||
return nil, errConfigNotFound
|
||||
}
|
||||
|
||||
return configData, nil
|
||||
sys.metadataMap[bucket] = meta
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// Init - initializes bucket metadata system for all buckets.
|
||||
@ -218,7 +289,6 @@ func (sys *BucketMetadataSys) Init(ctx context.Context, buckets []BucketInfo, ob
|
||||
}
|
||||
|
||||
// Loads bucket metadata for all buckets into BucketMetadataSys.
|
||||
// When if done successfully fallback will be disabled.
|
||||
func (sys *BucketMetadataSys) load(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
sys.Lock()
|
||||
defer sys.Unlock()
|
||||
@ -230,7 +300,6 @@ func (sys *BucketMetadataSys) load(ctx context.Context, buckets []BucketInfo, ob
|
||||
}
|
||||
sys.metadataMap[bucket.Name] = meta
|
||||
}
|
||||
sys.fallback = false
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -238,8 +307,5 @@ func (sys *BucketMetadataSys) load(ctx context.Context, buckets []BucketInfo, ob
|
||||
func NewBucketMetadataSys() *BucketMetadataSys {
|
||||
return &BucketMetadataSys{
|
||||
metadataMap: make(map[string]BucketMetadata),
|
||||
|
||||
// Do fallback until all buckets have been loaded.
|
||||
fallback: true,
|
||||
}
|
||||
}
|
||||
|
@ -17,14 +17,23 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"time"
|
||||
|
||||
"github.com/minio/minio-go/v6/pkg/tags"
|
||||
"github.com/minio/minio/cmd/logger"
|
||||
bucketsse "github.com/minio/minio/pkg/bucket/encryption"
|
||||
"github.com/minio/minio/pkg/bucket/lifecycle"
|
||||
objectlock "github.com/minio/minio/pkg/bucket/object/lock"
|
||||
"github.com/minio/minio/pkg/bucket/policy"
|
||||
"github.com/minio/minio/pkg/event"
|
||||
"github.com/minio/minio/pkg/madmin"
|
||||
)
|
||||
|
||||
const (
|
||||
@ -58,6 +67,15 @@ type BucketMetadata struct {
|
||||
EncryptionConfigXML []byte
|
||||
TaggingConfigXML []byte
|
||||
QuotaConfigJSON []byte
|
||||
|
||||
// Unexported fields. Must be updated atomically.
|
||||
policyConfig *policy.Policy
|
||||
notificationConfig *event.Config
|
||||
lifecycleConfig *lifecycle.Lifecycle
|
||||
objectLockConfig *objectlock.Config
|
||||
sseConfig *bucketsse.BucketSSEConfig
|
||||
taggingConfig *tags.Tags
|
||||
quotaConfig *madmin.BucketQuota
|
||||
}
|
||||
|
||||
// newBucketMetadata creates BucketMetadata with the supplied name and Created to Now.
|
||||
@ -112,6 +130,76 @@ func loadBucketMetadata(ctx context.Context, objectAPI ObjectLayer, bucket strin
|
||||
return b, b.convertLegacyConfigs(ctx, objectAPI)
|
||||
}
|
||||
|
||||
// parseAllConfigs will parse all configs and populate the private fields.
|
||||
// The first error encountered is returned.
|
||||
func (b *BucketMetadata) parseAllConfigs(ctx context.Context, objectAPI ObjectLayer) (err error) {
|
||||
if len(b.PolicyConfigJSON) != 0 {
|
||||
b.policyConfig, err = policy.ParseConfig(bytes.NewReader(b.PolicyConfigJSON), b.Name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.policyConfig = nil
|
||||
}
|
||||
|
||||
if len(b.NotificationXML) != 0 {
|
||||
if err = xml.Unmarshal(b.NotificationXML, b.notificationConfig); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.notificationConfig = &event.Config{
|
||||
XMLNS: "http://s3.amazonaws.com/doc/2006-03-01/",
|
||||
}
|
||||
b.notificationConfig.SetRegion(globalServerRegion)
|
||||
}
|
||||
|
||||
if len(b.LifecycleConfigXML) != 0 {
|
||||
b.lifecycleConfig, err = lifecycle.ParseLifecycleConfig(bytes.NewReader(b.LifecycleConfigXML))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.lifecycleConfig = nil
|
||||
}
|
||||
|
||||
if len(b.EncryptionConfigXML) != 0 {
|
||||
b.sseConfig, err = bucketsse.ParseBucketSSEConfig(bytes.NewReader(b.EncryptionConfigXML))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.sseConfig = nil
|
||||
}
|
||||
|
||||
if len(b.TaggingConfigXML) != 0 {
|
||||
b.taggingConfig, err = tags.ParseBucketXML(bytes.NewReader(b.TaggingConfigXML))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.taggingConfig = nil
|
||||
}
|
||||
|
||||
if len(b.ObjectLockConfigurationXML) != 0 {
|
||||
b.objectLockConfig, err = objectlock.ParseObjectLockConfig(bytes.NewReader(b.ObjectLockConfigurationXML))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
b.objectLockConfig = nil
|
||||
}
|
||||
if len(b.QuotaConfigJSON) != 0 {
|
||||
qcfg, err := parseBucketQuota(b.Name, b.QuotaConfigJSON)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
b.quotaConfig = qcfg
|
||||
} else {
|
||||
b.quotaConfig = &madmin.BucketQuota{}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *BucketMetadata) convertLegacyConfigs(ctx context.Context, objectAPI ObjectLayer) error {
|
||||
legacyConfigs := []string{
|
||||
legacyBucketObjectLockEnabledConfigFile,
|
||||
@ -150,7 +238,7 @@ func (b *BucketMetadata) convertLegacyConfigs(ctx context.Context, objectAPI Obj
|
||||
|
||||
if len(configs) == 0 {
|
||||
// nothing to update, return right away.
|
||||
return nil
|
||||
return b.parseAllConfigs(ctx, objectAPI)
|
||||
}
|
||||
|
||||
for legacyFile, configData := range configs {
|
||||
@ -194,6 +282,10 @@ func (b *BucketMetadata) convertLegacyConfigs(ctx context.Context, objectAPI Obj
|
||||
|
||||
// Save config to supplied ObjectLayer api.
|
||||
func (b *BucketMetadata) Save(ctx context.Context, api ObjectLayer) error {
|
||||
if err := b.parseAllConfigs(ctx, api); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
data := make([]byte, 4, b.Msgsize()+4)
|
||||
|
||||
// Initialize the header.
|
||||
|
@ -69,56 +69,43 @@ func (api objectAPIHandlers) GetBucketNotificationHandler(w http.ResponseWriter,
|
||||
return
|
||||
}
|
||||
|
||||
var config = event.Config{
|
||||
XMLNS: "http://s3.amazonaws.com/doc/2006-03-01/",
|
||||
}
|
||||
config.SetRegion(globalServerRegion)
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucketName, bucketNotificationConfig)
|
||||
config, err := globalBucketMetadataSys.GetNotificationConfig(bucketName)
|
||||
if err != nil {
|
||||
if err != errConfigNotFound {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
} else {
|
||||
if err = xml.Unmarshal(configData, &config); err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
if err = config.Validate(globalServerRegion, globalNotificationSys.targetList); err != nil {
|
||||
arnErr, ok := err.(*event.ErrARNNotFound)
|
||||
if ok {
|
||||
for i, queue := range config.QueueList {
|
||||
// Remove ARN not found queues, because we previously allowed
|
||||
// adding unexpected entries into the config.
|
||||
//
|
||||
// With newer config disallowing changing / turning off
|
||||
// notification targets without removing ARN in notification
|
||||
// configuration we won't see this problem anymore.
|
||||
if reflect.DeepEqual(queue.ARN, arnErr.ARN) && i < len(config.QueueList) {
|
||||
config.QueueList = append(config.QueueList[:i],
|
||||
config.QueueList[i+1:]...)
|
||||
}
|
||||
// This is a one time activity we shall do this
|
||||
// here and allow stale ARN to be removed. We shall
|
||||
// never reach a stage where we will have stale
|
||||
// notification configs.
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
if err = config.Validate(globalServerRegion, globalNotificationSys.targetList); err != nil {
|
||||
arnErr, ok := err.(*event.ErrARNNotFound)
|
||||
if ok {
|
||||
for i, queue := range config.QueueList {
|
||||
// Remove ARN not found queues, because we previously allowed
|
||||
// adding unexpected entries into the config.
|
||||
//
|
||||
// With newer config disallowing changing / turning off
|
||||
// notification targets without removing ARN in notification
|
||||
// configuration we won't see this problem anymore.
|
||||
if reflect.DeepEqual(queue.ARN, arnErr.ARN) && i < len(config.QueueList) {
|
||||
config.QueueList = append(config.QueueList[:i],
|
||||
config.QueueList[i+1:]...)
|
||||
}
|
||||
} else {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
// This is a one time activity we shall do this
|
||||
// here and allow stale ARN to be removed. We shall
|
||||
// never reach a stage where we will have stale
|
||||
// notification configs.
|
||||
}
|
||||
} else {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
notificationBytes, err := xml.Marshal(config)
|
||||
configData, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
writeSuccessResponseXML(w, notificationBytes)
|
||||
writeSuccessResponseXML(w, configData)
|
||||
}
|
||||
|
||||
// PutBucketNotificationHandler - This HTTP handler stores given notification configuration as per
|
||||
|
@ -17,9 +17,7 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"math"
|
||||
"net/http"
|
||||
|
||||
@ -30,9 +28,7 @@ import (
|
||||
)
|
||||
|
||||
// BucketObjectLockSys - map of bucket and retention configuration.
|
||||
type BucketObjectLockSys struct {
|
||||
retentionMap map[string]objectlock.Retention
|
||||
}
|
||||
type BucketObjectLockSys struct{}
|
||||
|
||||
// Get - Get retention configuration.
|
||||
func (sys *BucketObjectLockSys) Get(bucketName string) (r objectlock.Retention, err error) {
|
||||
@ -45,20 +41,13 @@ func (sys *BucketObjectLockSys) Get(bucketName string) (r objectlock.Retention,
|
||||
return r, nil
|
||||
}
|
||||
|
||||
r, ok := sys.retentionMap[bucketName]
|
||||
if ok {
|
||||
return r, nil
|
||||
}
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucketName, objectLockConfig)
|
||||
config, err := globalBucketMetadataSys.GetObjectLockConfig(bucketName)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
if _, ok := err.(BucketObjectLockConfigNotFound); ok {
|
||||
return r, nil
|
||||
}
|
||||
return r, err
|
||||
}
|
||||
config, err := objectlock.ParseObjectLockConfig(bytes.NewReader(configData))
|
||||
if err != nil {
|
||||
return r, err
|
||||
|
||||
}
|
||||
return config.ToRetention(), nil
|
||||
}
|
||||
@ -418,46 +407,5 @@ func checkPutObjectLockAllowed(ctx context.Context, r *http.Request, bucket, obj
|
||||
|
||||
// NewBucketObjectLockSys returns initialized BucketObjectLockSys
|
||||
func NewBucketObjectLockSys() *BucketObjectLockSys {
|
||||
return &BucketObjectLockSys{
|
||||
retentionMap: make(map[string]objectlock.Retention),
|
||||
}
|
||||
}
|
||||
|
||||
// Init - initializes bucket object lock config system for all buckets
|
||||
func (sys *BucketObjectLockSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
if objAPI == nil {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
|
||||
// In gateway mode, we always fetch the bucket object lock configuration from the gateway backend.
|
||||
// So, this is a no-op for gateway servers.
|
||||
if globalIsGateway {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Load BucketObjectLockSys once during boot.
|
||||
return sys.load(buckets, objAPI)
|
||||
}
|
||||
|
||||
func (sys *BucketObjectLockSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
ctx := logger.SetReqInfo(GlobalContext, &logger.ReqInfo{BucketName: bucket.Name})
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, objectLockConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
retention := objectlock.Retention{}
|
||||
config, err := objectlock.ParseObjectLockConfig(bytes.NewReader(configData))
|
||||
if err != nil {
|
||||
logger.LogIf(ctx, err)
|
||||
sys.retentionMap[bucket.Name] = retention
|
||||
continue
|
||||
}
|
||||
sys.retentionMap[bucket.Name] = config.ToRetention()
|
||||
}
|
||||
return nil
|
||||
return &BucketObjectLockSys{}
|
||||
}
|
||||
|
@ -18,7 +18,6 @@ package cmd
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
|
||||
@ -165,11 +164,14 @@ func (api objectAPIHandlers) GetBucketPolicyHandler(w http.ResponseWriter, r *ht
|
||||
}
|
||||
|
||||
// Read bucket access policy.
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketPolicyConfig)
|
||||
config, err := globalBucketMetadataSys.GetPolicyConfig(bucket)
|
||||
if err != nil {
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
||||
configData, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
err = BucketPolicyNotFound{Bucket: bucket}
|
||||
}
|
||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
|
||||
return
|
||||
}
|
||||
|
@ -17,9 +17,7 @@
|
||||
package cmd
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
@ -35,9 +33,7 @@ import (
|
||||
)
|
||||
|
||||
// PolicySys - policy subsystem.
|
||||
type PolicySys struct {
|
||||
bucketPolicyMap map[string]*policy.Policy
|
||||
}
|
||||
type PolicySys struct{}
|
||||
|
||||
// Get returns stored bucket policy
|
||||
func (sys *PolicySys) Get(bucket string) (*policy.Policy, error) {
|
||||
@ -48,48 +44,19 @@ func (sys *PolicySys) Get(bucket string) (*policy.Policy, error) {
|
||||
}
|
||||
return objAPI.GetBucketPolicy(GlobalContext, bucket)
|
||||
}
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket, bucketPolicyConfig)
|
||||
if err != nil {
|
||||
if !errors.Is(err, errConfigNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
return nil, BucketPolicyNotFound{Bucket: bucket}
|
||||
}
|
||||
return policy.ParseConfig(bytes.NewReader(configData), bucket)
|
||||
return globalBucketMetadataSys.GetPolicyConfig(bucket)
|
||||
}
|
||||
|
||||
// IsAllowed - checks given policy args is allowed to continue the Rest API.
|
||||
func (sys *PolicySys) IsAllowed(args policy.Args) bool {
|
||||
if globalIsGateway {
|
||||
objAPI := newObjectLayerFn()
|
||||
if objAPI == nil {
|
||||
return false
|
||||
}
|
||||
p, err := objAPI.GetBucketPolicy(GlobalContext, args.BucketName)
|
||||
if err == nil {
|
||||
return p.IsAllowed(args)
|
||||
}
|
||||
return args.IsOwner
|
||||
}
|
||||
|
||||
// If policy is available for given bucket, check the policy.
|
||||
p, found := sys.bucketPolicyMap[args.BucketName]
|
||||
if found {
|
||||
p, err := sys.Get(args.BucketName)
|
||||
if err == nil {
|
||||
return p.IsAllowed(args)
|
||||
}
|
||||
|
||||
configData, err := globalBucketMetadataSys.GetConfig(args.BucketName, bucketPolicyConfig)
|
||||
if err != nil {
|
||||
if !errors.Is(err, errConfigNotFound) {
|
||||
logger.LogIf(GlobalContext, err)
|
||||
}
|
||||
} else {
|
||||
p, err = policy.ParseConfig(bytes.NewReader(configData), args.BucketName)
|
||||
if err != nil {
|
||||
logger.LogIf(GlobalContext, err)
|
||||
} else {
|
||||
return p.IsAllowed(args)
|
||||
}
|
||||
// Log unhandled errors.
|
||||
if _, ok := err.(BucketPolicyNotFound); !ok {
|
||||
logger.LogIf(GlobalContext, err)
|
||||
}
|
||||
|
||||
// As policy is not available for given bucket name, returns IsOwner i.e.
|
||||
@ -97,47 +64,9 @@ func (sys *PolicySys) IsAllowed(args policy.Args) bool {
|
||||
return args.IsOwner
|
||||
}
|
||||
|
||||
// Loads policies for all buckets into PolicySys.
|
||||
func (sys *PolicySys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketPolicyConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
config, err := policy.ParseConfig(bytes.NewReader(configData), bucket.Name)
|
||||
if err != nil {
|
||||
logger.LogIf(GlobalContext, err)
|
||||
continue
|
||||
}
|
||||
sys.bucketPolicyMap[bucket.Name] = config
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Init - initializes policy system from policy.json of all buckets.
|
||||
func (sys *PolicySys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
if objAPI == nil {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
|
||||
// In gateway mode, we don't need to load the policies
|
||||
// from the backend.
|
||||
if globalIsGateway {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Load PolicySys once during boot.
|
||||
return sys.load(buckets, objAPI)
|
||||
}
|
||||
|
||||
// NewPolicySys - creates new policy system.
|
||||
func NewPolicySys() *PolicySys {
|
||||
return &PolicySys{
|
||||
bucketPolicyMap: make(map[string]*policy.Policy),
|
||||
}
|
||||
return &PolicySys{}
|
||||
}
|
||||
|
||||
func getConditionValues(r *http.Request, lc string, username string, claims map[string]interface{}) map[string][]string {
|
||||
|
@ -19,7 +19,6 @@ package cmd
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
@ -32,95 +31,34 @@ import (
|
||||
)
|
||||
|
||||
// BucketQuotaSys - map of bucket and quota configuration.
|
||||
type BucketQuotaSys struct {
|
||||
quotaMap map[string]madmin.BucketQuota
|
||||
}
|
||||
type BucketQuotaSys struct{}
|
||||
|
||||
// Get - Get quota configuration.
|
||||
func (sys *BucketQuotaSys) Get(bucketName string) (q madmin.BucketQuota, err error) {
|
||||
func (sys *BucketQuotaSys) Get(bucketName string) (*madmin.BucketQuota, error) {
|
||||
if globalIsGateway {
|
||||
objAPI := newObjectLayerFn()
|
||||
if objAPI == nil {
|
||||
return q, errServerNotInitialized
|
||||
return nil, errServerNotInitialized
|
||||
}
|
||||
return q, BucketQuotaConfigNotFound{Bucket: bucketName}
|
||||
return &madmin.BucketQuota{}, nil
|
||||
}
|
||||
|
||||
q, ok := sys.quotaMap[bucketName]
|
||||
if !ok {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucketName, bucketQuotaConfigFile)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
return q, BucketQuotaConfigNotFound{Bucket: bucketName}
|
||||
}
|
||||
return q, err
|
||||
}
|
||||
return parseBucketQuota(bucketName, configData)
|
||||
}
|
||||
return q, nil
|
||||
}
|
||||
|
||||
// Buckets - list of buckets with quota configuration
|
||||
func (sys *BucketQuotaSys) Buckets() []string {
|
||||
var buckets []string
|
||||
for k := range sys.quotaMap {
|
||||
buckets = append(buckets, k)
|
||||
}
|
||||
return buckets
|
||||
}
|
||||
|
||||
// Init initialize bucket quota sys configuration with all buckets.
|
||||
func (sys *BucketQuotaSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
if objAPI == nil {
|
||||
return errServerNotInitialized
|
||||
}
|
||||
|
||||
// In gateway mode, we do not support bucket quota.
|
||||
// So, this is a no-op for gateway servers.
|
||||
if globalIsGateway {
|
||||
return nil
|
||||
}
|
||||
|
||||
return sys.load(buckets, objAPI)
|
||||
}
|
||||
|
||||
func (sys *BucketQuotaSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
ctx := logger.SetReqInfo(GlobalContext, &logger.ReqInfo{BucketName: bucket.Name})
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketQuotaConfigFile)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
quotaCfg, err := parseBucketQuota(bucket.Name, configData)
|
||||
if err != nil {
|
||||
if _, ok := err.(BucketQuotaConfigNotFound); !ok {
|
||||
logger.LogIf(ctx, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
sys.quotaMap[bucket.Name] = quotaCfg
|
||||
}
|
||||
return nil
|
||||
return globalBucketMetadataSys.GetQuotaConfig(bucketName)
|
||||
}
|
||||
|
||||
// NewBucketQuotaSys returns initialized BucketQuotaSys
|
||||
func NewBucketQuotaSys() *BucketQuotaSys {
|
||||
return &BucketQuotaSys{quotaMap: map[string]madmin.BucketQuota{}}
|
||||
return &BucketQuotaSys{}
|
||||
}
|
||||
|
||||
// parseBucketQuota parses BucketQuota from json
|
||||
func parseBucketQuota(bucket string, data []byte) (quotaCfg madmin.BucketQuota, err error) {
|
||||
if len(data) == 0 {
|
||||
return quotaCfg, BucketQuotaConfigNotFound{Bucket: bucket}
|
||||
}
|
||||
if err = json.Unmarshal(data, "aCfg); err != nil {
|
||||
func parseBucketQuota(bucket string, data []byte) (quotaCfg *madmin.BucketQuota, err error) {
|
||||
quotaCfg = &madmin.BucketQuota{}
|
||||
if err = json.Unmarshal(data, quotaCfg); err != nil {
|
||||
return quotaCfg, err
|
||||
}
|
||||
if !quotaCfg.Type.IsValid() {
|
||||
return quotaCfg, fmt.Errorf("Invalid quota type %s", quotaCfg.Type)
|
||||
if !quotaCfg.IsValid() {
|
||||
return quotaCfg, fmt.Errorf("Invalid quota config %#v", quotaCfg)
|
||||
}
|
||||
return
|
||||
}
|
||||
@ -131,7 +69,12 @@ type bucketStorageCache struct {
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func (b *bucketStorageCache) check(ctx context.Context, q madmin.BucketQuota, bucket string, size int64) error {
|
||||
func (b *bucketStorageCache) check(ctx context.Context, q *madmin.BucketQuota, bucket string, size int64) error {
|
||||
if q.Quota == 0 {
|
||||
// No quota set return quickly.
|
||||
return nil
|
||||
}
|
||||
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if time.Since(b.lastUpdate) > 10*time.Second {
|
||||
@ -143,7 +86,7 @@ func (b *bucketStorageCache) check(ctx context.Context, q madmin.BucketQuota, bu
|
||||
b.bucketsSizes = dui.BucketsSizes
|
||||
}
|
||||
currUsage := b.bucketsSizes[bucket]
|
||||
if (currUsage+uint64(size)) > q.Quota && q.Quota > 0 {
|
||||
if (currUsage + uint64(size)) > q.Quota {
|
||||
return BucketQuotaExceeded{Bucket: bucket}
|
||||
}
|
||||
return nil
|
||||
@ -155,15 +98,13 @@ func enforceBucketQuota(ctx context.Context, bucket string, size int64) error {
|
||||
|
||||
q, err := globalBucketQuotaSys.Get(bucket)
|
||||
if err != nil {
|
||||
if _, ok := err.(BucketQuotaConfigNotFound); !ok {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
if q.Type == madmin.FIFOQuota {
|
||||
return nil
|
||||
}
|
||||
|
||||
return globalBucketStorageCache.check(ctx, q, bucket, size)
|
||||
}
|
||||
|
||||
@ -198,7 +139,12 @@ func enforceFIFOQuota(ctx context.Context, objectAPI ObjectLayer) error {
|
||||
if env.Get(envDataUsageCrawlConf, config.EnableOn) == config.EnableOff {
|
||||
return nil
|
||||
}
|
||||
for _, bucket := range globalBucketQuotaSys.Buckets() {
|
||||
buckets, err := objectAPI.ListBuckets(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, binfo := range buckets {
|
||||
bucket := binfo.Name
|
||||
// Check if the current bucket has quota restrictions, if not skip it
|
||||
cfg, err := globalBucketQuotaSys.Get(bucket)
|
||||
if err != nil {
|
||||
|
@ -410,7 +410,6 @@ func (fs *FSObjects) ListBuckets(ctx context.Context) ([]BucketInfo, error) {
|
||||
if err == nil {
|
||||
created = meta.Created
|
||||
}
|
||||
globalBucketMetadataSys.Set(fi.Name(), meta)
|
||||
|
||||
bucketInfos = append(bucketInfos, BucketInfo{
|
||||
Name: fi.Name(),
|
||||
|
@ -20,7 +20,6 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/url"
|
||||
@ -613,15 +612,11 @@ func (sys *NotificationSys) AddRemoteTarget(bucketName string, target event.Targ
|
||||
func (sys *NotificationSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
ctx := logger.SetReqInfo(GlobalContext, &logger.ReqInfo{BucketName: bucket.Name})
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketNotificationConfig)
|
||||
config, err := globalBucketMetadataSys.GetNotificationConfig(bucket.Name)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
config, err := event.ParseConfig(bytes.NewReader(configData), globalServerRegion, globalNotificationSys.targetList)
|
||||
if err != nil {
|
||||
if err = config.Validate(globalServerRegion, globalNotificationSys.targetList); err != nil {
|
||||
if _, ok := err.(*event.ErrARNNotFound); !ok {
|
||||
logger.LogIf(ctx, err)
|
||||
}
|
||||
|
@ -601,10 +601,8 @@ func (s *peerRESTServer) LoadBucketMetadataHandler(w http.ResponseWriter, r *htt
|
||||
|
||||
meta, err := loadBucketMetadata(r.Context(), objAPI, bucketName)
|
||||
if err != nil {
|
||||
if !errors.Is(err, errConfigNotFound) {
|
||||
s.writeErrorResponse(w, err)
|
||||
return
|
||||
}
|
||||
s.writeErrorResponse(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
globalBucketMetadataSys.Set(bucketName, meta)
|
||||
|
@ -27,8 +27,7 @@ import (
|
||||
)
|
||||
|
||||
func TestPolicySysIsAllowed(t *testing.T) {
|
||||
policySys := NewPolicySys()
|
||||
policySys.bucketPolicyMap["mybucket"] = &policy.Policy{
|
||||
p := &policy.Policy{
|
||||
Version: policy.DefaultVersion,
|
||||
Statements: []policy.Statement{
|
||||
policy.NewStatement(
|
||||
@ -121,22 +120,21 @@ func TestPolicySysIsAllowed(t *testing.T) {
|
||||
}
|
||||
|
||||
testCases := []struct {
|
||||
policySys *PolicySys
|
||||
args policy.Args
|
||||
expectedResult bool
|
||||
}{
|
||||
{policySys, anonGetBucketLocationArgs, true},
|
||||
{policySys, anonPutObjectActionArgs, true},
|
||||
{policySys, anonGetObjectActionArgs, false},
|
||||
{policySys, getBucketLocationArgs, true},
|
||||
{policySys, putObjectActionArgs, true},
|
||||
{policySys, getObjectActionArgs, true},
|
||||
{policySys, yourbucketAnonGetObjectActionArgs, false},
|
||||
{policySys, yourbucketGetObjectActionArgs, true},
|
||||
{anonGetBucketLocationArgs, true},
|
||||
{anonPutObjectActionArgs, true},
|
||||
{anonGetObjectActionArgs, false},
|
||||
{getBucketLocationArgs, true},
|
||||
{putObjectActionArgs, true},
|
||||
{getObjectActionArgs, true},
|
||||
{yourbucketAnonGetObjectActionArgs, false},
|
||||
{yourbucketGetObjectActionArgs, true},
|
||||
}
|
||||
|
||||
for i, testCase := range testCases {
|
||||
result := testCase.policySys.IsAllowed(testCase.args)
|
||||
result := p.IsAllowed(testCase.args)
|
||||
|
||||
if result != testCase.expectedResult {
|
||||
t.Fatalf("case %v: expected: %v, got: %v\n", i+1, testCase.expectedResult, result)
|
||||
|
@ -335,31 +335,6 @@ func initAllSubsystems(newObject ObjectLayer) (err error) {
|
||||
return fmt.Errorf("Unable to initialize notification system: %w", err)
|
||||
}
|
||||
|
||||
// Initialize policy system.
|
||||
if err = globalPolicySys.Init(buckets, newObject); err != nil {
|
||||
return fmt.Errorf("Unable to initialize policy system: %w", err)
|
||||
}
|
||||
|
||||
// Initialize lifecycle system.
|
||||
if err = globalLifecycleSys.Init(buckets, newObject); err != nil {
|
||||
return fmt.Errorf("Unable to initialize lifecycle system: %w", err)
|
||||
}
|
||||
|
||||
// Initialize bucket encryption subsystem.
|
||||
if err = globalBucketSSEConfigSys.Init(buckets, newObject); err != nil {
|
||||
return fmt.Errorf("Unable to initialize bucket encryption subsystem: %w", err)
|
||||
}
|
||||
|
||||
// Initialize bucket object lock.
|
||||
if err = globalBucketObjectLockSys.Init(buckets, newObject); err != nil {
|
||||
return fmt.Errorf("Unable to initialize object lock system: %w", err)
|
||||
}
|
||||
|
||||
// Initialize bucket quota system.
|
||||
if err = globalBucketQuotaSys.Init(buckets, newObject); err != nil {
|
||||
return fmt.Errorf("Unable to initialize bucket quota system: %w", err)
|
||||
}
|
||||
|
||||
// Populate existing buckets to the etcd backend
|
||||
if globalDNSConfig != nil {
|
||||
initFederatorBackend(buckets, newObject)
|
||||
|
@ -323,8 +323,10 @@ func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location s
|
||||
}
|
||||
|
||||
// If it doesn't exist we get a new, so ignore errors
|
||||
meta, _ := globalBucketMetadataSys.Get(bucket)
|
||||
meta.ObjectLockConfigurationXML = defaultBucketObjectLockConfig
|
||||
meta := newBucketMetadata(bucket)
|
||||
if lockEnabled {
|
||||
meta.ObjectLockConfigurationXML = defaultBucketObjectLockConfig
|
||||
}
|
||||
if err := meta.Save(ctx, z); err != nil {
|
||||
return toObjectErr(err, bucket)
|
||||
}
|
||||
@ -351,8 +353,10 @@ func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location s
|
||||
}
|
||||
|
||||
// If it doesn't exist we get a new, so ignore errors
|
||||
meta, _ := globalBucketMetadataSys.Get(bucket)
|
||||
meta.ObjectLockConfigurationXML = defaultBucketObjectLockConfig
|
||||
meta := newBucketMetadata(bucket)
|
||||
if lockEnabled {
|
||||
meta.ObjectLockConfigurationXML = defaultBucketObjectLockConfig
|
||||
}
|
||||
if err := meta.Save(ctx, z); err != nil {
|
||||
return toObjectErr(err, bucket)
|
||||
}
|
||||
@ -1296,7 +1300,6 @@ func (z *xlZones) ListBuckets(ctx context.Context) (buckets []BucketInfo, err er
|
||||
if err == nil {
|
||||
buckets[i].Created = meta.Created
|
||||
}
|
||||
globalBucketMetadataSys.Set(buckets[i].Name, meta)
|
||||
}
|
||||
return buckets, nil
|
||||
}
|
||||
@ -1519,7 +1522,6 @@ func (z *xlZones) ListBucketsHeal(ctx context.Context) ([]BucketInfo, error) {
|
||||
if err == nil {
|
||||
healBuckets[i].Created = meta.Created
|
||||
}
|
||||
globalBucketMetadataSys.Set(healBuckets[i].Name, meta)
|
||||
}
|
||||
|
||||
return healBuckets, nil
|
||||
|
@ -43,32 +43,17 @@ func (t QuotaType) IsValid() bool {
|
||||
// BucketQuota holds bucket quota restrictions
|
||||
type BucketQuota struct {
|
||||
Quota uint64 `json:"quota"`
|
||||
Type QuotaType `json:"quotatype"`
|
||||
Type QuotaType `json:"quotatype,omitempty"`
|
||||
}
|
||||
|
||||
// RemoveBucketQuota - removes quota config on a bucket.
|
||||
func (adm *AdminClient) RemoveBucketQuota(ctx context.Context, bucket string) error {
|
||||
|
||||
queryValues := url.Values{}
|
||||
queryValues.Set("bucket", bucket)
|
||||
|
||||
reqData := requestData{
|
||||
relPath: adminAPIPrefix + "/remove-bucket-quota",
|
||||
queryValues: queryValues,
|
||||
// IsValid returns false if quota is invalid
|
||||
// empty quota when Quota == 0 is always true.
|
||||
func (q BucketQuota) IsValid() bool {
|
||||
if q.Quota > 0 {
|
||||
return q.Type.IsValid()
|
||||
}
|
||||
|
||||
// Execute DELETE on /minio/admin/v3/remove-bucket-quota to delete bucket quota.
|
||||
resp, err := adm.executeMethod(ctx, http.MethodDelete, reqData)
|
||||
defer closeResponse(resp)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusNoContent {
|
||||
return httpRespToErrorResponse(resp)
|
||||
}
|
||||
|
||||
return nil
|
||||
// Empty configs are valid.
|
||||
return true
|
||||
}
|
||||
|
||||
// GetBucketQuota - get info on a user
|
||||
@ -104,9 +89,9 @@ func (adm *AdminClient) GetBucketQuota(ctx context.Context, bucket string) (q Bu
|
||||
return q, nil
|
||||
}
|
||||
|
||||
// SetBucketQuota - sets a bucket's quota.
|
||||
// SetBucketQuota - sets a bucket's quota, if quota is set to '0'
|
||||
// quota is disabled.
|
||||
func (adm *AdminClient) SetBucketQuota(ctx context.Context, bucket string, quota uint64, quotaType QuotaType) error {
|
||||
|
||||
data, err := json.Marshal(BucketQuota{
|
||||
Quota: quota,
|
||||
Type: quotaType,
|
||||
|
Loading…
Reference in New Issue
Block a user