astelm/minio
1
0
Fork 0
mirror of https://github.com/pgsty/minio.git synced 2026-08-11 00:33:28 +03:00
Code Issues Packages Projects Releases Wiki Activity

cleanup object-lock/bucket tagging for gateways (#9548)

Browse Source 
This PR is to ensure that we call the relevant object
layer APIs for necessary S3 API level functionalities
allowing gateway implementations to return proper
errors as NotImplemented{}

This allows for all our tests in mint to behave
appropriately and can be handled appropriately as
well.
This commit is contained in:
Harshavardhana
2020-05-08 13:44:44 -07:00
committed by GitHub
parent 6885c72f32
commit a1de9cec58
49 changed files with 681 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/xl-zones.go
+94 -50
View File
@@ -31,6 +31,7 @@ import (
"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/madmin"
"github.com/minio/minio/pkg/sync/errgroup"
@@ -336,9 +337,19 @@ func undoMakeBucketZones(bucket string, zones []*xlSets, errs []error) {
// MakeBucketWithLocation - creates a new bucket across all zones simultaneously
// even if one of the sets fail to create buckets, we proceed all the successful
// operations.
func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location string) error {
func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location string, lockEnabled bool) error {
if z.SingleZone() {
return z.zones[0].MakeBucketWithLocation(ctx, bucket, location)
if err := z.zones[0].MakeBucketWithLocation(ctx, bucket, location, lockEnabled); err != nil {
return err
}
if lockEnabled {
meta := newBucketMetadata(bucket)
meta.LockEnabled = lockEnabled
if err := meta.save(ctx, z); err != nil {
return toObjectErr(err, bucket)
}
}
return nil
}
g := errgroup.WithNErrs(len(z.zones))
@@ -347,7 +358,7 @@ func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location s
for index := range z.zones {
index := index
g.Go(func() error {
return z.zones[index].MakeBucketWithLocation(ctx, bucket, location)
return z.zones[index].MakeBucketWithLocation(ctx, bucket, location, lockEnabled)
}, index)
}
@@ -362,6 +373,14 @@ func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location s
}
}
if lockEnabled {
meta := newBucketMetadata(bucket)
meta.LockEnabled = lockEnabled
if err := meta.save(ctx, z); err != nil {
return toObjectErr(err, bucket)
}
}
// Success.
return nil
@@ -405,11 +424,11 @@ func (z *xlZones) GetObjectNInfo(ctx context.Context, bucket, object string, rs
func (z *xlZones) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error {
// Lock the object before reading.
objectLock := z.NewNSLock(ctx, bucket, object)
if err := objectLock.GetRLock(globalObjectTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err := lk.GetRLock(globalObjectTimeout); err != nil {
return err
}
defer objectLock.RUnlock()
defer lk.RUnlock()
if z.SingleZone() {
return z.zones[0].GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts)
@@ -428,11 +447,11 @@ func (z *xlZones) GetObject(ctx context.Context, bucket, object string, startOff
func (z *xlZones) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
// Lock the object before reading.
objectLock := z.NewNSLock(ctx, bucket, object)
if err := objectLock.GetRLock(globalObjectTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err := lk.GetRLock(globalObjectTimeout); err != nil {
return ObjectInfo{}, err
}
defer objectLock.RUnlock()
defer lk.RUnlock()
if z.SingleZone() {
return z.zones[0].GetObjectInfo(ctx, bucket, object, opts)
@@ -453,11 +472,11 @@ func (z *xlZones) GetObjectInfo(ctx context.Context, bucket, object string, opts
// PutObject - writes an object to least used erasure zone.
func (z *xlZones) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
// Lock the object.
objectLock := z.NewNSLock(ctx, bucket, object)
if err := objectLock.GetLock(globalObjectTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err := lk.GetLock(globalObjectTimeout); err != nil {
return ObjectInfo{}, err
}
defer objectLock.Unlock()
defer lk.Unlock()
if z.SingleZone() {
return z.zones[0].PutObject(ctx, bucket, object, data, opts)
@@ -480,11 +499,11 @@ func (z *xlZones) PutObject(ctx context.Context, bucket string, object string, d
func (z *xlZones) DeleteObject(ctx context.Context, bucket string, object string) error {
// Acquire a write lock before deleting the object.
objectLock := z.NewNSLock(ctx, bucket, object)
if err := objectLock.GetLock(globalOperationTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err := lk.GetLock(globalOperationTimeout); err != nil {
return err
}
defer objectLock.Unlock()
defer lk.Unlock()
if z.SingleZone() {
return z.zones[0].DeleteObject(ctx, bucket, object)
@@ -531,11 +550,11 @@ func (z *xlZones) CopyObject(ctx context.Context, srcBucket, srcObject, destBuck
// Check if this request is only metadata update.
cpSrcDstSame := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(destBucket, destObject))
if !cpSrcDstSame {
objectLock := z.NewNSLock(ctx, destBucket, destObject)
if err := objectLock.GetLock(globalObjectTimeout); err != nil {
lk := z.NewNSLock(ctx, destBucket, destObject)
if err := lk.GetLock(globalObjectTimeout); err != nil {
return objInfo, err
}
defer objectLock.Unlock()
defer lk.Unlock()
}
if z.SingleZone() {
@@ -1110,11 +1129,11 @@ func (z *xlZones) CompleteMultipartUpload(ctx context.Context, bucket, object, u
// Hold namespace to complete the transaction, only hold
// if uploadID can be held exclusively.
objectLock := z.NewNSLock(ctx, bucket, object)
if err = objectLock.GetLock(globalOperationTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err = lk.GetLock(globalOperationTimeout); err != nil {
return objInfo, err
}
defer objectLock.Unlock()
defer lk.Unlock()
if z.SingleZone() {
return z.zones[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
@@ -1201,6 +1220,31 @@ func (z *xlZones) SetBucketSSEConfig(ctx context.Context, bucket string, config
return saveBucketSSEConfig(ctx, z, bucket, config)
}
// SetBucketObjectLockConfig enables/clears default object lock configuration
func (z *xlZones) SetBucketObjectLockConfig(ctx context.Context, bucket string, config *objectlock.Config) error {
return saveBucketObjectLockConfig(ctx, z, bucket, config)
}
// GetBucketObjectLockConfig - returns current defaults for object lock configuration
func (z *xlZones) GetBucketObjectLockConfig(ctx context.Context, bucket string) (*objectlock.Config, error) {
return readBucketObjectLockConfig(ctx, z, bucket)
}
// SetBucketTagging sets bucket tags on given bucket
func (z *xlZones) SetBucketTagging(ctx context.Context, bucket string, t *tags.Tags) error {
return saveBucketTagging(ctx, z, bucket, t)
}
// GetBucketTagging get bucket tags set on given bucket
func (z *xlZones) GetBucketTagging(ctx context.Context, bucket string) (*tags.Tags, error) {
return readBucketTagging(ctx, z, bucket)
}
// DeleteBucketTagging delete bucket tags set if any.
func (z *xlZones) DeleteBucketTagging(ctx context.Context, bucket string) error {
return deleteBucketTagging(ctx, z, bucket)
}
// DeleteBucketSSEConfig deletes bucket encryption config on given bucket
func (z *xlZones) DeleteBucketSSEConfig(ctx context.Context, bucket string) error {
return removeBucketSSEConfig(ctx, z, bucket)
@@ -1245,27 +1289,14 @@ func (z *xlZones) DeleteBucket(ctx context.Context, bucket string, forceDelete b
errs := g.Wait()
if forceDelete {
for _, err := range errs {
if err != nil {
if _, ok := err.(InsufficientWriteQuorum); ok {
undoDeleteBucketZones(bucket, z.zones, errs)
}
return err
}
}
return nil
}
// For any write quorum failure, we undo all the delete buckets operation
// by creating all the buckets again.
// For any write quorum failure, we undo all the delete
// buckets operation by creating all the buckets again.
for _, err := range errs {
if err != nil {
if _, ok := err.(InsufficientWriteQuorum); ok {
undoDeleteBucketZones(bucket, z.zones, errs)
}
return err
}
}
@@ -1283,7 +1314,7 @@ func undoDeleteBucketZones(bucket string, zones []*xlSets, errs []error) {
index := index
g.Go(func() error {
if errs[index] == nil {
return zones[index].MakeBucketWithLocation(GlobalContext, bucket, "")
return zones[index].MakeBucketWithLocation(GlobalContext, bucket, "", false)
}
return nil
}, index)
@@ -1297,17 +1328,30 @@ func undoDeleteBucketZones(bucket string, zones []*xlSets, errs []error) {
// that all buckets are present on all zones.
func (z *xlZones) ListBuckets(ctx context.Context) (buckets []BucketInfo, err error) {
if z.SingleZone() {
return z.zones[0].ListBuckets(ctx)
}
for _, zone := range z.zones {
buckets, err := zone.ListBuckets(ctx)
if err != nil {
logger.LogIf(ctx, err)
continue
buckets, err = z.zones[0].ListBuckets(ctx)
} else {
for _, zone := range z.zones {
buckets, err = zone.ListBuckets(ctx)
if err != nil {
logger.LogIf(ctx, err)
continue
}
break
}
return buckets, nil
}
return buckets, InsufficientReadQuorum{}
if err != nil {
return nil, err
}
for i := range buckets {
meta, err := loadBucketMetadata(ctx, z, buckets[i].Name)
if err == nil {
buckets[i].Created = meta.Created
}
if err != errMetaDataConverted {
logger.LogIf(ctx, err)
}
}
return buckets, nil
}
func (z *xlZones) ReloadFormat(ctx context.Context, dryRun bool) error {
@@ -1488,11 +1532,11 @@ func (z *xlZones) HealObjects(ctx context.Context, bucket, prefix string, opts m
func (z *xlZones) HealObject(ctx context.Context, bucket, object string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
// Lock the object before healing. Use read lock since healing
// will only regenerate parts & xl.json of outdated disks.
objectLock := z.NewNSLock(ctx, bucket, object)
if err := objectLock.GetRLock(globalHealingTimeout); err != nil {
lk := z.NewNSLock(ctx, bucket, object)
if err := lk.GetRLock(globalHealingTimeout); err != nil {
return madmin.HealResultItem{}, err
}
defer objectLock.RUnlock()
defer lk.RUnlock()
if z.SingleZone() {
return z.zones[0].HealObject(ctx, bucket, object, opts)