minio/cmd/bucket-metadata-sys.go

497 lines
15 KiB
Go
Raw Normal View History

// Copyright (c) 2015-2021 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/minio/madmin-go"
"github.com/minio/minio-go/v7/pkg/tags"
bucketsse "github.com/minio/minio/internal/bucket/encryption"
"github.com/minio/minio/internal/bucket/lifecycle"
objectlock "github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/replication"
"github.com/minio/minio/internal/bucket/versioning"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/kms"
"github.com/minio/minio/internal/logger"
"github.com/minio/minio/internal/sync/errgroup"
"github.com/minio/pkg/bucket/policy"
)
// BucketMetadataSys captures all bucket metadata for a given cluster.
type BucketMetadataSys struct {
sync.RWMutex
metadataMap map[string]BucketMetadata
}
// Remove bucket metadata from memory.
func (sys *BucketMetadataSys) Remove(bucket string) {
if globalIsGateway {
return
}
sys.Lock()
delete(sys.metadataMap, bucket)
globalBucketMonitor.DeleteBucket(bucket)
sys.Unlock()
}
// Set - sets a new metadata in-memory.
// Only a shallow copy is saved and fields with references
// cannot be modified without causing a race condition,
// so they should be replaced atomically and not appended to, etc.
// Data is not persisted to disk.
func (sys *BucketMetadataSys) Set(bucket string, meta BucketMetadata) {
if globalIsGateway {
return
}
if bucket != minioMetaBucket {
sys.Lock()
sys.metadataMap[bucket] = meta
sys.Unlock()
}
}
// Update update bucket metadata for the specified config file.
// The configData data should not be modified after being sent here.
func (sys *BucketMetadataSys) Update(ctx context.Context, bucket string, configFile string, configData []byte) (updatedAt time.Time, err error) {
objAPI := newObjectLayerFn()
if objAPI == nil {
return updatedAt, errServerNotInitialized
}
if globalIsGateway && globalGatewayName != NASBackendGateway {
if configFile == bucketPolicyConfig {
if configData == nil {
return updatedAt, objAPI.DeleteBucketPolicy(ctx, bucket)
}
config, err := policy.ParseConfig(bytes.NewReader(configData), bucket)
if err != nil {
return updatedAt, err
}
return updatedAt, objAPI.SetBucketPolicy(ctx, bucket, config)
}
return updatedAt, NotImplemented{}
}
if bucket == minioMetaBucket {
return updatedAt, errInvalidArgument
}
meta, err := loadBucketMetadata(ctx, objAPI, bucket)
if err != nil {
if !globalIsErasure && !globalIsDistErasure && errors.Is(err, errVolumeNotFound) {
// Only single drive mode needs this fallback.
meta = newBucketMetadata(bucket)
} else {
return updatedAt, err
}
}
updatedAt = UTCNow()
switch configFile {
case bucketPolicyConfig:
meta.PolicyConfigJSON = configData
meta.PolicyConfigUpdatedAt = updatedAt
case bucketNotificationConfig:
meta.NotificationConfigXML = configData
case bucketLifecycleConfig:
meta.LifecycleConfigXML = configData
case bucketSSEConfig:
meta.EncryptionConfigXML = configData
meta.EncryptionConfigUpdatedAt = updatedAt
case bucketTaggingConfig:
meta.TaggingConfigXML = configData
meta.TaggingConfigUpdatedAt = updatedAt
case bucketQuotaConfigFile:
meta.QuotaConfigJSON = configData
meta.QuotaConfigUpdatedAt = updatedAt
case objectLockConfig:
meta.ObjectLockConfigXML = configData
meta.ObjectLockConfigUpdatedAt = updatedAt
case bucketVersioningConfig:
meta.VersioningConfigXML = configData
meta.VersioningConfigUpdatedAt = updatedAt
case bucketReplicationConfig:
meta.ReplicationConfigXML = configData
meta.ReplicationConfigUpdatedAt = updatedAt
case bucketTargetsFile:
meta.BucketTargetsConfigJSON, meta.BucketTargetsConfigMetaJSON, err = encryptBucketMetadata(meta.Name, configData, kms.Context{
bucket: meta.Name,
bucketTargetsFile: bucketTargetsFile,
})
if err != nil {
return updatedAt, fmt.Errorf("Error encrypting bucket target metadata %w", err)
}
default:
return updatedAt, fmt.Errorf("Unknown bucket %s metadata update requested %s", bucket, configFile)
}
if err := meta.Save(ctx, objAPI); err != nil {
return updatedAt, err
}
sys.Set(bucket, meta)
globalNotificationSys.LoadBucketMetadata(bgContext(ctx), bucket) // Do not use caller context here
return updatedAt, nil
}
// Get metadata for a bucket.
// If no metadata exists errConfigNotFound is returned and a new metadata is returned.
// Only a shallow copy is returned, so referenced data should not be modified,
// but can be replaced atomically.
//
// This function should only be used with
// - GetBucketInfo
// - ListBuckets
// For all other bucket specific metadata, use the relevant
// calls implemented specifically for each of those features.
func (sys *BucketMetadataSys) Get(bucket string) (BucketMetadata, error) {
if globalIsGateway || bucket == minioMetaBucket {
return newBucketMetadata(bucket), errConfigNotFound
}
sys.RLock()
defer sys.RUnlock()
meta, ok := sys.metadataMap[bucket]
if !ok {
return newBucketMetadata(bucket), errConfigNotFound
}
return meta, nil
}
// GetVersioningConfig returns configured versioning config
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetVersioningConfig(bucket string) (*versioning.Versioning, time.Time, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return &versioning.Versioning{XMLNS: "http://s3.amazonaws.com/doc/2006-03-01/"}, meta.Created, nil
}
return &versioning.Versioning{XMLNS: "http://s3.amazonaws.com/doc/2006-03-01/"}, time.Time{}, err
}
return meta.versioningConfig, meta.VersioningConfigUpdatedAt, nil
}
// GetTaggingConfig returns configured tagging config
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetTaggingConfig(bucket string) (*tags.Tags, time.Time, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketTaggingNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
if meta.taggingConfig == nil {
return nil, time.Time{}, BucketTaggingNotFound{Bucket: bucket}
}
return meta.taggingConfig, meta.TaggingConfigUpdatedAt, nil
}
// GetObjectLockConfig returns configured object lock config
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetObjectLockConfig(bucket string) (*objectlock.Config, time.Time, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketObjectLockConfigNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
if meta.objectLockConfig == nil {
return nil, time.Time{}, BucketObjectLockConfigNotFound{Bucket: bucket}
}
return meta.objectLockConfig, meta.ObjectLockConfigUpdatedAt, nil
}
// GetLifecycleConfig returns configured lifecycle config
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetLifecycleConfig(bucket string) (*lifecycle.Lifecycle, error) {
if globalIsGateway && globalGatewayName == NASBackendGateway {
// Only needed in case of NAS gateway.
objAPI := newObjectLayerFn()
if objAPI == nil {
return nil, errServerNotInitialized
}
meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
if err != nil {
return nil, err
}
if meta.lifecycleConfig == nil {
return nil, BucketLifecycleNotFound{Bucket: bucket}
}
return meta.lifecycleConfig, nil
}
meta, err := sys.GetConfig(GlobalContext, 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) {
if globalIsGateway && globalGatewayName == NASBackendGateway {
// Only needed in case of NAS gateway.
objAPI := newObjectLayerFn()
if objAPI == nil {
return nil, errServerNotInitialized
}
meta, err := loadBucketMetadata(GlobalContext, objAPI, bucket)
if err != nil {
return nil, err
}
return meta.notificationConfig, nil
}
meta, err := sys.GetConfig(GlobalContext, 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, time.Time, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketSSEConfigNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
if meta.sseConfig == nil {
return nil, time.Time{}, BucketSSEConfigNotFound{Bucket: bucket}
}
return meta.sseConfig, meta.EncryptionConfigUpdatedAt, nil
}
// CreatedAt returns the time of creation of bucket
func (sys *BucketMetadataSys) CreatedAt(bucket string) (time.Time, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
return time.Time{}, err
}
return meta.Created.UTC(), nil
}
// GetPolicyConfig returns configured bucket policy
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetPolicyConfig(bucket string) (*policy.Policy, time.Time, error) {
if globalIsGateway {
objAPI := newObjectLayerFn()
if objAPI == nil {
return nil, time.Time{}, errServerNotInitialized
}
p, err := objAPI.GetBucketPolicy(GlobalContext, bucket)
return p, UTCNow(), err
}
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketPolicyNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
if meta.policyConfig == nil {
return nil, time.Time{}, BucketPolicyNotFound{Bucket: bucket}
}
return meta.policyConfig, meta.PolicyConfigUpdatedAt, nil
}
// GetQuotaConfig returns configured bucket quota
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetQuotaConfig(ctx context.Context, bucket string) (*madmin.BucketQuota, time.Time, error) {
meta, err := sys.GetConfig(ctx, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketQuotaConfigNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
return meta.quotaConfig, meta.QuotaConfigUpdatedAt, nil
}
// GetReplicationConfig returns configured bucket replication config
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetReplicationConfig(ctx context.Context, bucket string) (*replication.Config, time.Time, error) {
meta, err := sys.GetConfig(ctx, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, time.Time{}, BucketReplicationConfigNotFound{Bucket: bucket}
}
return nil, time.Time{}, err
}
if meta.replicationConfig == nil {
return nil, time.Time{}, BucketReplicationConfigNotFound{Bucket: bucket}
}
return meta.replicationConfig, meta.ReplicationConfigUpdatedAt, nil
}
// GetBucketTargetsConfig returns configured bucket targets for this bucket
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetBucketTargetsConfig(bucket string) (*madmin.BucketTargets, error) {
meta, err := sys.GetConfig(GlobalContext, bucket)
if err != nil {
if errors.Is(err, errConfigNotFound) {
return nil, BucketRemoteTargetNotFound{Bucket: bucket}
}
return nil, err
}
if meta.bucketTargetConfig == nil {
return nil, BucketRemoteTargetNotFound{Bucket: bucket}
}
return meta.bucketTargetConfig, nil
}
// GetConfig returns a specific configuration from the bucket metadata.
// The returned object may not be modified.
func (sys *BucketMetadataSys) GetConfig(ctx context.Context, bucket string) (BucketMetadata, error) {
objAPI := newObjectLayerFn()
if objAPI == nil {
return newBucketMetadata(bucket), errServerNotInitialized
}
if globalIsGateway {
return newBucketMetadata(bucket), NotImplemented{}
}
if bucket == minioMetaBucket {
return newBucketMetadata(bucket), errInvalidArgument
}
sys.RLock()
meta, ok := sys.metadataMap[bucket]
sys.RUnlock()
if ok {
return meta, nil
}
meta, err := loadBucketMetadata(ctx, objAPI, bucket)
if err != nil {
return meta, err
}
sys.Lock()
sys.metadataMap[bucket] = meta
sys.Unlock()
return meta, nil
}
// Init - initializes bucket metadata system for all buckets.
func (sys *BucketMetadataSys) Init(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) error {
if objAPI == nil {
return errServerNotInitialized
}
// In gateway mode, we don't need to load bucket metadata except
// NAS gateway backend.
if globalIsGateway && !objAPI.IsNotificationSupported() {
return nil
}
// Load bucket metadata sys in background
go sys.load(ctx, buckets, objAPI)
return nil
}
// concurrently load bucket metadata to speed up loading bucket metadata.
func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) {
g := errgroup.WithNErrs(len(buckets))
for index := range buckets {
index := index
g.Go(func() error {
_, _ = objAPI.HealBucket(ctx, buckets[index].Name, madmin.HealOpts{
// Ensure heal opts for bucket metadata be deep healed all the time.
ScanMode: madmin.HealDeepScan,
Recreate: true,
})
meta, err := loadBucketMetadata(ctx, objAPI, buckets[index].Name)
if err != nil {
if !globalIsErasure && !globalIsDistErasure && errors.Is(err, errVolumeNotFound) {
meta = newBucketMetadata(buckets[index].Name)
} else {
return err
}
}
sys.Lock()
sys.metadataMap[buckets[index].Name] = meta
sys.Unlock()
globalNotificationSys.set(buckets[index], meta) // set notification targets
globalBucketTargetSys.set(buckets[index], meta) // set remote replication targets
return nil
}, index)
}
for _, err := range g.Wait() {
if err != nil {
logger.LogIf(ctx, err)
}
}
}
// Loads bucket metadata for all buckets into BucketMetadataSys.
func (sys *BucketMetadataSys) load(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) {
count := 100 // load 100 bucket metadata at a time.
for {
if len(buckets) < count {
sys.concurrentLoad(ctx, buckets, objAPI)
return
}
sys.concurrentLoad(ctx, buckets[:count], objAPI)
buckets = buckets[count:]
}
}
tests: Fix occasional data race (#11223) CI tests could trigger a data race. Servers are generally not expected to reinitialize, so tests could trigger data races when reinitializing and async operations are running. We add the option to safely reset global vars instead of overwriting. Fixes races like: ``` WARNING: DATA RACE Read at 0x00000477ab18 by goroutine 1159: github.com/minio/minio/cmd.FileInfo.ToObjectInfo() /home/runner/work/minio/minio/cmd/erasure-metadata.go:105 +0x16d github.com/minio/minio/cmd.erasureObjects.putObject() /home/runner/work/minio/minio/cmd/erasure-object.go:748 +0x13f8 github.com/minio/minio/cmd.(*erasureObjects).listPath.func3.2() /home/runner/work/minio/minio/cmd/metacache-set.go:682 +0x7d3 github.com/minio/minio/cmd.newMetacacheBlockWriter.func1.2() /home/runner/work/minio/minio/cmd/metacache-stream.go:777 +0x1c4 github.com/minio/minio/cmd.newMetacacheBlockWriter.func1() /home/runner/work/minio/minio/cmd/metacache-stream.go:806 +0x614 Previous write at 0x00000477ab18 by goroutine 1269: [failed to restore the stack] Goroutine 1159 (running) created at: github.com/minio/minio/cmd.newMetacacheBlockWriter() /home/runner/work/minio/minio/cmd/metacache-stream.go:760 +0x112 github.com/minio/minio/cmd.(*erasureObjects).listPath.func3() /home/runner/work/minio/minio/cmd/metacache-set.go:672 +0xe22 Goroutine 1269 (running) created at: testing.(*T).Run() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1095 +0x537 testing.runTests.func1() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1339 +0xa6 testing.tRunner() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1050 +0x1eb testing.runTests() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1337 +0x594 testing.(*M).Run() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1252 +0x2ff github.com/minio/minio/cmd.TestMain() /home/runner/work/minio/minio/cmd/test-utils_test.go:120 +0x44e main.main() _testmain.go:1408 +0x223 ================== ================== WARNING: DATA RACE Read at 0x00000477aae8 by goroutine 1159: github.com/minio/minio/cmd.(*BucketVersioningSys).Enabled() /home/runner/work/minio/minio/cmd/bucket-versioning.go:26 +0x52 github.com/minio/minio/cmd.FileInfo.ToObjectInfo() /home/runner/work/minio/minio/cmd/erasure-metadata.go:105 +0x197 github.com/minio/minio/cmd.erasureObjects.putObject() /home/runner/work/minio/minio/cmd/erasure-object.go:748 +0x13f8 github.com/minio/minio/cmd.(*erasureObjects).listPath.func3.2() /home/runner/work/minio/minio/cmd/metacache-set.go:682 +0x7d3 github.com/minio/minio/cmd.newMetacacheBlockWriter.func1.2() /home/runner/work/minio/minio/cmd/metacache-stream.go:777 +0x1c4 github.com/minio/minio/cmd.newMetacacheBlockWriter.func1() /home/runner/work/minio/minio/cmd/metacache-stream.go:806 +0x614 Previous write at 0x00000477aae8 by goroutine 1269: [failed to restore the stack] Goroutine 1159 (running) created at: github.com/minio/minio/cmd.newMetacacheBlockWriter() /home/runner/work/minio/minio/cmd/metacache-stream.go:760 +0x112 github.com/minio/minio/cmd.(*erasureObjects).listPath.func3() /home/runner/work/minio/minio/cmd/metacache-set.go:672 +0xe22 Goroutine 1269 (running) created at: testing.(*T).Run() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1095 +0x537 testing.runTests.func1() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1339 +0xa6 testing.tRunner() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1050 +0x1eb testing.runTests() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1337 +0x594 testing.(*M).Run() /opt/hostedtoolcache/go/1.14.13/x64/src/testing/testing.go:1252 +0x2ff github.com/minio/minio/cmd.TestMain() /home/runner/work/minio/minio/cmd/test-utils_test.go:120 +0x44e main.main() _testmain.go:1408 +0x223 ================== ```
2021-01-05 13:45:26 -05:00
// Reset the state of the BucketMetadataSys.
func (sys *BucketMetadataSys) Reset() {
sys.Lock()
for k := range sys.metadataMap {
delete(sys.metadataMap, k)
}
sys.Unlock()
}
// NewBucketMetadataSys - creates new policy system.
func NewBucketMetadataSys() *BucketMetadataSys {
return &BucketMetadataSys{
metadataMap: make(map[string]BucketMetadata),
}
}