From c045ae15e785995c5bea6b1375e912380ae23a71 Mon Sep 17 00:00:00 2001 From: Anis Elleuch Date: Tue, 12 May 2020 23:20:42 +0100 Subject: [PATCH] fix: avoid undoing bucket creation and return the first err instead (#9578) --- cmd/xl-sets.go | 29 ++++------------------------- cmd/xl-v1-bucket.go | 23 ----------------------- cmd/xl-v1-healing.go | 4 ---- cmd/xl-v1-healing_test.go | 37 ------------------------------------- cmd/xl-zones.go | 24 +----------------------- 5 files changed, 5 insertions(+), 112 deletions(-) diff --git a/cmd/xl-sets.go b/cmd/xl-sets.go index 6961f5c82..8319e2797 100644 --- a/cmd/xl-sets.go +++ b/cmd/xl-sets.go @@ -507,9 +507,8 @@ func (s *xlSets) Shutdown(ctx context.Context) error { return nil } -// MakeBucketLocation - creates a new bucket across all sets simultaneously -// even if one of the sets fail to create buckets, we proceed to undo a -// successful operation. +// MakeBucketLocation - creates a new bucket across all sets simultaneously, +// then return the first encountered error func (s *xlSets) MakeBucketWithLocation(ctx context.Context, bucket, location string, lockEnabled bool) error { g := errgroup.WithNErrs(len(s.sets)) @@ -522,11 +521,10 @@ func (s *xlSets) MakeBucketWithLocation(ctx context.Context, bucket, location st } errs := g.Wait() - // Upon any error we try to undo the make bucket operation if possible - // on all sets where it succeeded. + + // Return the first encountered error for _, err := range errs { if err != nil { - undoMakeBucketSets(bucket, s.sets, errs) return err } } @@ -535,25 +533,6 @@ func (s *xlSets) MakeBucketWithLocation(ctx context.Context, bucket, location st return nil } -// This function is used to undo a successful MakeBucket operation. -func undoMakeBucketSets(bucket string, sets []*xlObjects, errs []error) { - g := errgroup.WithNErrs(len(sets)) - - // Undo previous make bucket entry on all underlying sets. - for index := range sets { - index := index - g.Go(func() error { - if errs[index] == nil { - return sets[index].DeleteBucket(GlobalContext, bucket, false) - } - return nil - }, index) - } - - // Wait for all delete bucket to finish. - g.Wait() -} - // hashes the key returning an integer based on the input algorithm. // This function currently supports // - CRCMOD diff --git a/cmd/xl-v1-bucket.go b/cmd/xl-v1-bucket.go index 8eefb1f17..b42c03f0c 100644 --- a/cmd/xl-v1-bucket.go +++ b/cmd/xl-v1-bucket.go @@ -69,10 +69,6 @@ func (xl xlObjects) MakeBucketWithLocation(ctx context.Context, bucket, location writeQuorum := getWriteQuorum(len(storageDisks)) err := reduceWriteQuorumErrs(ctx, g.Wait(), bucketOpIgnoredErrs, writeQuorum) - if err == errXLWriteQuorum { - // Purge successfully created buckets if we don't have writeQuorum. - undoMakeBucket(storageDisks, bucket) - } return toObjectErr(err, bucket) } @@ -94,25 +90,6 @@ func undoDeleteBucket(storageDisks []StorageAPI, bucket string) { g.Wait() } -// undo make bucket operation upon quorum failure. -func undoMakeBucket(storageDisks []StorageAPI, bucket string) { - g := errgroup.WithNErrs(len(storageDisks)) - // Undo previous make bucket entry on all underlying storage disks. - for index := range storageDisks { - if storageDisks[index] == nil { - continue - } - index := index - g.Go(func() error { - _ = storageDisks[index].DeleteVol(bucket, false) - return nil - }, index) - } - - // Wait for all make vol to finish. - g.Wait() -} - // getBucketInfo - returns the BucketInfo from one of the load balanced disks. func (xl xlObjects) getBucketInfo(ctx context.Context, bucketName string) (bucketInfo BucketInfo, err error) { var bucketErrs []error diff --git a/cmd/xl-v1-healing.go b/cmd/xl-v1-healing.go index e7c78403f..ef3fb1373 100644 --- a/cmd/xl-v1-healing.go +++ b/cmd/xl-v1-healing.go @@ -150,10 +150,6 @@ func healBucket(ctx context.Context, storageDisks []StorageAPI, bucket string, w reducedErr = reduceWriteQuorumErrs(ctx, errs, bucketOpIgnoredErrs, writeQuorum) if reducedErr != nil { - if reducedErr == errXLWriteQuorum { - // Purge successfully created buckets if we don't have writeQuorum. - undoMakeBucket(storageDisks, bucket) - } return res, reducedErr } diff --git a/cmd/xl-v1-healing_test.go b/cmd/xl-v1-healing_test.go index 859995cad..3afd9ed71 100644 --- a/cmd/xl-v1-healing_test.go +++ b/cmd/xl-v1-healing_test.go @@ -25,43 +25,6 @@ import ( "github.com/minio/minio/pkg/madmin" ) -// Tests undoes and validates if the undoing completes successfully. -func TestUndoMakeBucket(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - - nDisks := 16 - fsDirs, err := getRandomDisks(nDisks) - if err != nil { - t.Fatal(err) - } - defer removeRoots(fsDirs) - - // Remove format.json on 16 disks. - obj, _, err := initObjectLayer(ctx, mustGetZoneEndpoints(fsDirs...)) - if err != nil { - t.Fatal(err) - } - - bucketName := getRandomBucketName() - if err = obj.MakeBucketWithLocation(ctx, bucketName, "", false); err != nil { - t.Fatal(err) - } - z := obj.(*xlZones) - xl := z.zones[0].sets[0] - undoMakeBucket(xl.getDisks(), bucketName) - - // Validate if bucket was deleted properly. - _, err = obj.GetBucketInfo(ctx, bucketName) - if err != nil { - switch err.(type) { - case BucketNotFound: - default: - t.Fatal(err) - } - } -} - func TestHealObjectCorrupted(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/cmd/xl-zones.go b/cmd/xl-zones.go index b1e1aa42e..6901a201d 100644 --- a/cmd/xl-zones.go +++ b/cmd/xl-zones.go @@ -315,25 +315,6 @@ func (z *xlZones) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter, upd return firstErr } -// This function is used to undo a successful MakeBucket operation. -func undoMakeBucketZones(bucket string, zones []*xlSets, errs []error) { - g := errgroup.WithNErrs(len(zones)) - - // Undo previous make bucket entry on all underlying zones. - for index := range zones { - index := index - g.Go(func() error { - if errs[index] == nil { - return zones[index].DeleteBucket(GlobalContext, bucket, false) - } - return nil - }, index) - } - - // Wait for all delete bucket to finish. - g.Wait() -} - // 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. @@ -363,12 +344,9 @@ func (z *xlZones) MakeBucketWithLocation(ctx context.Context, bucket, location s } errs := g.Wait() - // Upon even a single write quorum error we undo all previously created buckets. + // Return the first encountered error for _, err := range errs { if err != nil { - if _, ok := err.(InsufficientWriteQuorum); ok { - undoMakeBucketZones(bucket, z.zones, errs) - } return err } }