fix: heal multiple buckets in bulk (#11029)

makes server startup, orders of magnitude
faster with large number of buckets

cmd/erasure-bucket.go

@@ -31,6 +31,34 @@ var bucketOpIgnoredErrs = append(baseIgnoredErrs, errDiskAccessDenied, errUnform
 // list all errors that can be ignored in a bucket metadata operation.
 var bucketMetadataOpIgnoredErrs = append(bucketOpIgnoredErrs, errVolumeNotFound)
 
+// MakeMultipleBuckets - create a list of buckets
+func (er erasureObjects) MakeMultipleBuckets(ctx context.Context, buckets ...string) error {
+	storageDisks := er.getDisks()
+
+	g := errgroup.WithNErrs(len(storageDisks))
+
+	// Make a volume entry on all underlying storage disks.
+	for index := range storageDisks {
+		index := index
+		g.Go(func() error {
+			if storageDisks[index] != nil {
+				if err := storageDisks[index].MakeVolBulk(ctx, buckets...); err != nil {
+					if !errors.Is(err, errVolumeExists) {
+						logger.LogIf(ctx, err)
+					}
+					return err
+				}
+				return nil
+			}
+			return errDiskNotFound
+		}, index)
+	}
+
+	writeQuorum := getWriteQuorum(len(storageDisks))
+	err := reduceWriteQuorumErrs(ctx, g.Wait(), bucketOpIgnoredErrs, writeQuorum)
+	return toObjectErr(err)
+}
+
 /// Bucket operations
 
 // MakeBucket - make a bucket.

cmd/erasure-server-sets.go

@@ -424,6 +424,28 @@ func (z *erasureServerPools) CrawlAndGetDataUsage(ctx context.Context, bf *bloom
 	return firstErr
 }
 
+func (z *erasureServerPools) MakeMultipleBuckets(ctx context.Context, buckets ...string) error {
+	g := errgroup.WithNErrs(len(z.serverPools))
+
+	// Create buckets in parallel across all sets.
+	for index := range z.serverPools {
+		index := index
+		g.Go(func() error {
+			return z.serverPools[index].MakeMultipleBuckets(ctx, buckets...)
+		}, index)
+	}
+
+	errs := g.Wait()
+	// Return the first encountered error
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // MakeBucketWithLocation - creates a new bucket across all serverPools simultaneously
 // even if one of the sets fail to create buckets, we proceed all the successful
 // operations.

cmd/erasure-sets.go

@@ -531,6 +531,31 @@ func (s *erasureSets) Shutdown(ctx context.Context) error {
 	return nil
 }
 
+// MakeMultipleBuckets - make many buckets at once.
+func (s *erasureSets) MakeMultipleBuckets(ctx context.Context, buckets ...string) error {
+	g := errgroup.WithNErrs(len(s.sets))
+
+	// Create buckets in parallel across all sets.
+	for index := range s.sets {
+		index := index
+		g.Go(func() error {
+			return s.sets[index].MakeMultipleBuckets(ctx, buckets...)
+		}, index)
+	}
+
+	errs := g.Wait()
+
+	// Return the first encountered error
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+
+	// Success.
+	return nil
+}
+
 // MakeBucketLocation - creates a new bucket across all sets simultaneously,
 // then return the first encountered error
 func (s *erasureSets) MakeBucketWithLocation(ctx context.Context, bucket string, opts BucketOptions) error {

cmd/gateway-unsupported.go

@@ -51,6 +51,11 @@ func (a GatewayUnsupported) SetDriveCount() int {
 	return 0
 }
 
+// MakeMultipleBuckets is dummy stub for gateway.
+func (a GatewayUnsupported) MakeMultipleBuckets(ctx context.Context, buckets ...string) error {
+	return NotImplemented{}
+}
+
 // ListMultipartUploads lists all multipart uploads.
 func (a GatewayUnsupported) ListMultipartUploads(ctx context.Context, bucket string, prefix string, keyMarker string, uploadIDMarker string, delimiter string, maxUploads int) (lmi ListMultipartsInfo, err error) {
 	return lmi, NotImplemented{}

cmd/iam.go

@@ -466,11 +466,10 @@ func (sys *IAMSys) Init(ctx context.Context, objAPI ObjectLayer) {
 
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 
-	var err error
 	for {
 		// let one of the server acquire the lock, if not let them timeout.
 		// which shall be retried again by this loop.
-		if err = txnLk.GetLock(retryCtx, iamLockTimeout); err != nil {
+		if err := txnLk.GetLock(retryCtx, iamLockTimeout); err != nil {
 			logger.Info("Waiting for all MinIO IAM sub-system to be initialized.. trying to acquire lock")
 			time.Sleep(time.Duration(r.Float64() * float64(5*time.Second)))
 			continue
@@ -480,7 +479,7 @@ func (sys *IAMSys) Init(ctx context.Context, objAPI ObjectLayer) {
 			// ****  WARNING ****
 			// Migrating to encrypted backend on etcd should happen before initialization of
 			// IAM sub-system, make sure that we do not move the above codeblock elsewhere.
-			if err = migrateIAMConfigsEtcdToEncrypted(ctx, globalEtcdClient); err != nil {
+			if err := migrateIAMConfigsEtcdToEncrypted(ctx, globalEtcdClient); err != nil {
 				txnLk.Unlock()
 				logger.LogIf(ctx, fmt.Errorf("Unable to decrypt an encrypted ETCD backend for IAM users and policies: %w", err))
 				logger.LogIf(ctx, errors.New("IAM sub-system is partially initialized, some users may not be available"))
@@ -494,7 +493,7 @@ func (sys *IAMSys) Init(ctx context.Context, objAPI ObjectLayer) {
 		}
 
 		// Migrate IAM configuration, if necessary.
-		if err = sys.doIAMConfigMigration(ctx); err != nil {
+		if err := sys.doIAMConfigMigration(ctx); err != nil {
 			txnLk.Unlock()
 			if errors.Is(err, errDiskNotFound) ||
 				errors.Is(err, errConfigNotFound) ||
@@ -515,14 +514,27 @@ func (sys *IAMSys) Init(ctx context.Context, objAPI ObjectLayer) {
 		break
 	}
 
-	err = sys.store.loadAll(ctx, sys)
+	for {
+		if err := sys.store.loadAll(ctx, sys); err != nil {
+			if errors.Is(err, errDiskNotFound) ||
+				errors.Is(err, errConfigNotFound) ||
+				errors.Is(err, context.DeadlineExceeded) ||
+				errors.As(err, &rquorum) ||
+				errors.As(err, &wquorum) ||
+				isErrBucketNotFound(err) {
+				logger.Info("Waiting for all MinIO IAM sub-system to be initialized.. possible cause (%v)", err)
+				time.Sleep(time.Duration(r.Float64() * float64(5*time.Second)))
+				continue
+			}
+			if err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize IAM sub-system, some users may not be available %w", err))
+			}
+		}
+		break
+	}
 
 	// Invalidate the old cred always, even upon error to avoid any leakage.
 	globalOldCred = auth.Credentials{}
-	if err != nil {
-		logger.LogIf(ctx, fmt.Errorf("Unable to initialize IAM sub-system, some users may not be available %w", err))
-	}
-
 	go sys.store.watch(ctx, sys)
 }
 

cmd/metacache-set.go

@@ -614,9 +614,6 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions) (entr
 
 	if len(disks) < askDisks {
 		err = InsufficientReadQuorum{}
-		if debugPrint {
-			console.Errorf("listPath: Insufficient disks, %d of %d needed are available", len(disks), askDisks)
-		}
 		logger.LogIf(ctx, fmt.Errorf("listPath: Insufficient disks, %d of %d needed are available", len(disks), askDisks))
 		cancel()
 		return

cmd/object-api-interface.go

@@ -82,6 +82,7 @@ type ObjectLayer interface {
 	StorageInfo(ctx context.Context, local bool) (StorageInfo, []error) // local queries only local disks
 
 	// Bucket operations.
+	MakeMultipleBuckets(ctx context.Context, buckets ...string) error
 	MakeBucketWithLocation(ctx context.Context, bucket string, opts BucketOptions) error
 	GetBucketInfo(ctx context.Context, bucket string) (bucketInfo BucketInfo, err error)
 	ListBuckets(ctx context.Context) (buckets []BucketInfo, err error)

cmd/server-main.go

@@ -304,18 +304,14 @@ func initAllSubsystems(ctx context.Context, newObject ObjectLayer) (err error) {
 		if err != nil {
 			return fmt.Errorf("Unable to list buckets to heal: %w", err)
 		}
-		for _, bucket := range buckets {
-			if err = newObject.MakeBucketWithLocation(ctx, bucket.Name, BucketOptions{}); err != nil {
-				if errors.As(err, &wquorum) || errors.As(err, &rquorum) {
-					// Return the error upwards for the caller to retry.
-					return fmt.Errorf("Unable to heal bucket: %w", err)
-				}
-				if _, ok := err.(BucketExists); !ok {
-					// ignore any other error and log for investigation.
-					logger.LogIf(ctx, err)
-					continue
-				}
-				// Bucket already exists, nothing that needs to be done.
+		bucketNames := make([]string, len(buckets))
+		for i := range buckets {
+			bucketNames[i] = buckets[i].Name
+		}
+		if err = newObject.MakeMultipleBuckets(ctx, bucketNames...); err != nil {
+			if errors.As(err, &wquorum) || errors.As(err, &rquorum) {
+				// Return the error upwards for the caller to retry.
+				return fmt.Errorf("Unable to heal buckets: %w", err)
 			}
 		}
 	} else {