<commit_before>/*
 * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cmd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path"
	"sync"
	"time"

	"github.com/minio/minio/cmd/config"
	"github.com/minio/minio/cmd/logger"
	"github.com/minio/minio/pkg/env"
	"github.com/minio/minio/pkg/event"
	"github.com/minio/minio/pkg/madmin"
)

// BucketQuotaSys - map of bucket and quota configuration.
type BucketQuotaSys struct {
	sync.RWMutex
	quotaMap map[string]madmin.BucketQuota
}

// Set - set quota configuration.
func (sys *BucketQuotaSys) Set(bucketName string, q madmin.BucketQuota) {
	sys.Lock()
	sys.quotaMap[bucketName] = q
	sys.Unlock()
}

// Get - Get quota configuration.
func (sys *BucketQuotaSys) Get(bucketName string) (q madmin.BucketQuota, ok bool) {
	sys.RLock()
	defer sys.RUnlock()
	q, ok = sys.quotaMap[bucketName]
	return
}

// Remove - removes quota configuration.
func (sys *BucketQuotaSys) Remove(bucketName string) {
	sys.Lock()
	delete(sys.quotaMap, bucketName)
	sys.Unlock()
}

// Buckets - list of buckets with quota configuration
func (sys *BucketQuotaSys) Buckets() []string {
	sys.RLock()
	defer sys.RUnlock()
	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})
		configFile := path.Join(bucketConfigPrefix, bucket.Name, bucketQuotaConfigFile)
		configData, err := readConfig(ctx, objAPI, configFile)
		if err != nil {
			if errors.Is(err, errConfigNotFound) {
				continue
			}
			return err
		}
		quotaCfg, err := parseBucketQuota(configData)
		if err != nil {
			return err
		}
		globalBucketQuotaSys.Set(bucket.Name, quotaCfg)
	}
	return nil
}

// NewBucketQuotaSys returns initialized BucketQuotaSys
func NewBucketQuotaSys() *BucketQuotaSys {
	return &BucketQuotaSys{quotaMap: map[string]madmin.BucketQuota{}}
}

// parseBucketQuota parses BucketQuota from json
func parseBucketQuota(data []byte) (quotaCfg madmin.BucketQuota, err error) {
	err = json.Unmarshal(data, &quotaCfg)
	if err != nil {
		return
	}
	if !quotaCfg.Type.IsValid() {
		return quotaCfg, fmt.Errorf("Invalid quota type %s", quotaCfg.Type)
	}
	return
}

type bucketStorageCache struct {
	bucketsSizes map[string]uint64
	lastUpdate   time.Time
	mu           sync.Mutex
}

func (b *bucketStorageCache) check(ctx context.Context, q madmin.BucketQuota, bucket string, size int64) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if time.Since(b.lastUpdate) > 10*time.Second {
		dui, err := loadDataUsageFromBackend(ctx, newObjectLayerWithoutSafeModeFn())
		if err != nil {
			return err
		}
		b.lastUpdate = time.Now()
		b.bucketsSizes = dui.BucketsSizes
	}
	currUsage := b.bucketsSizes[bucket]
	if (currUsage + uint64(size)) > q.Quota {
		return BucketQuotaExceeded{Bucket: bucket}
	}
	return nil
}
func enforceBucketQuota(ctx context.Context, bucket string, size int64) error {
	if size < 0 {
		return nil
	}
	q, ok := globalBucketQuotaSys.Get(bucket)
	if !ok {
		return nil
	}

	return globalBucketStorageCache.check(ctx, q, bucket, size)
}

const (
	bgQuotaInterval = 1 * time.Hour
)

// initQuotaEnforcement starts the routine that deletes objects in bucket
// that exceeds the FIFO quota
func initQuotaEnforcement(ctx context.Context, objAPI ObjectLayer) {
	if env.Get(envDataUsageCrawlConf, config.EnableOn) == config.EnableOn {
		go startBucketQuotaEnforcement(ctx, objAPI)
	}
}

func startBucketQuotaEnforcement(ctx context.Context, objAPI ObjectLayer) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-time.NewTimer(bgQuotaInterval).C:
			logger.LogIf(ctx, enforceFIFOQuota(ctx, objAPI))
		}

	}
}

// enforceFIFOQuota deletes objects in FIFO order until sufficient objects<commit_msg>Allow writes for bucket exceeding FIFO quota (#9575)<commit_after>/*
 * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package cmd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path"
	"sync"
	"time"

	"github.com/minio/minio/cmd/config"
	"github.com/minio/minio/cmd/logger"
	"github.com/minio/minio/pkg/env"
	"github.com/minio/minio/pkg/event"
	"github.com/minio/minio/pkg/madmin"
)

// BucketQuotaSys - map of bucket and quota configuration.
type BucketQuotaSys struct {
	sync.RWMutex
	quotaMap map[string]madmin.BucketQuota
}

// Set - set quota configuration.
func (sys *BucketQuotaSys) Set(bucketName string, q madmin.BucketQuota) {
	sys.Lock()
	sys.quotaMap[bucketName] = q
	sys.Unlock()
}

// Get - Get quota configuration.
func (sys *BucketQuotaSys) Get(bucketName string) (q madmin.BucketQuota, ok bool) {
	sys.RLock()
	defer sys.RUnlock()
	q, ok = sys.quotaMap[bucketName]
	return
}

// Remove - removes quota configuration.
func (sys *BucketQuotaSys) Remove(bucketName string) {
	sys.Lock()
	delete(sys.quotaMap, bucketName)
	sys.Unlock()
}

// Buckets - list of buckets with quota configuration
func (sys *BucketQuotaSys) Buckets() []string {
	sys.RLock()
	defer sys.RUnlock()
	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})
		configFile := path.Join(bucketConfigPrefix, bucket.Name, bucketQuotaConfigFile)
		configData, err := readConfig(ctx, objAPI, configFile)
		if err != nil {
			if errors.Is(err, errConfigNotFound) {
				continue
			}
			return err
		}
		quotaCfg, err := parseBucketQuota(configData)
		if err != nil {
			return err
		}
		globalBucketQuotaSys.Set(bucket.Name, quotaCfg)
	}
	return nil
}

// NewBucketQuotaSys returns initialized BucketQuotaSys
func NewBucketQuotaSys() *BucketQuotaSys {
	return &BucketQuotaSys{quotaMap: map[string]madmin.BucketQuota{}}
}

// parseBucketQuota parses BucketQuota from json
func parseBucketQuota(data []byte) (quotaCfg madmin.BucketQuota, err error) {
	err = json.Unmarshal(data, &quotaCfg)
	if err != nil {
		return
	}
	if !quotaCfg.Type.IsValid() {
		return quotaCfg, fmt.Errorf("Invalid quota type %s", quotaCfg.Type)
	}
	return
}

type bucketStorageCache struct {
	bucketsSizes map[string]uint64
	lastUpdate   time.Time
	mu           sync.Mutex
}

func (b *bucketStorageCache) check(ctx context.Context, q madmin.BucketQuota, bucket string, size int64) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	if time.Since(b.lastUpdate) > 10*time.Second {
		dui, err := loadDataUsageFromBackend(ctx, newObjectLayerWithoutSafeModeFn())
		if err != nil {
			return err
		}
		b.lastUpdate = time.Now()
		b.bucketsSizes = dui.BucketsSizes
	}
	currUsage := b.bucketsSizes[bucket]
	if (currUsage + uint64(size)) > q.Quota {
		return BucketQuotaExceeded{Bucket: bucket}
	}
	return nil
}
func enforceBucketQuota(ctx context.Context, bucket string, size int64) error {
	if size < 0 {
		return nil
	}
	q, ok := globalBucketQuotaSys.Get(bucket)
	if !ok {
		return nil
	}

	if q.Type == madmin.FIFOQuota {
		return nil
	}
	return globalBucketStorageCache.check(ctx, q, bucket, size)
}

const (
	bgQuotaInterval = 1 * time.Hour
)

// initQuotaEnforcement starts the routine that deletes objects in bucket
// that exceeds the FIFO quota
func initQuotaEnforcement(ctx context.Context, objAPI ObjectLayer) {
	if env.Get(envDataUsageCrawlConf, config.EnableOn) == config.EnableOn {
		go startBucketQuotaEnforcement(ctx, objAPI)
	}
}

func startBucketQuotaEnforcement(ctx context.Context, objAPI ObjectLayer) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-time.NewTimer(bgQuotaInterval).C:
			logger.LogIf(ctx, enforceFIFOQuota(ctx, objAPI))
		}

	}
}

// enforceFIFOQuota deletes objects in FIFO order until sufficient objects
{'fraction_non_alphanumeric': 0.10807184145334435, 'fraction_numerical': 0.004541701073492981, 'mean_word_length': 4.209139784946236, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7015059', 'n_tokens_mistral': 3449, 'n_tokens_neox': 3254, 'n_words': 911}