fix: update storage-class properly

fixes #14005

COMPLIANCE.md

@@ -0,0 +1,6 @@
+# AGPLv3 Compliance
+We have designed MinIO as an Open Source software for the Open Source software community. This requires applications to consider whether their usage of MinIO is in compliance with the GNU AGPLv3 [license](https://github.com/minio/minio/blob/master/LICENSE).
+
+MinIO cannot make the determination as to whether your application's usage of MinIO is in compliance with the AGPLv3 license requirements. You should instead rely on your own legal counsel or licensing specialists to audit and ensure your application is in compliance with the licenses of MinIO and all other open-source projects with which your application integrates or interacts. We understand that AGPLv3 licensing is complex and nuanced. It is for that reason we strongly encourage using experts in licensing to make any such determinations around compliance instead of relying on apocryphal or anecdotal advice.
+
+[MinIO Commercial Licensing](https://min.io/pricing) is the best option for applications that trigger AGPLv3 obligations (e.g. open sourcing your application). Applications using MinIO - or any other OSS-licensed code - without validating their usage do so at their own risk.

README.md

@@ -237,13 +237,11 @@ mc admin update <minio alias, e.g., myminio>
 
 ## Important things to remember during MinIO upgrades
 
+- Container environments are advised to update the container images instead of updating binaries inside the container.
+- `mc admin update` is disabled in kubernetes/container environments, container environments provide their own mechanisms to rollout of updates.
 - `mc admin update` will only work if the user running MinIO has write access to the parent directory where the binary is located, for example if the current binary is at `/usr/local/bin/minio`, you would need write access to `/usr/local/bin`.
 - `mc admin update` updates and restarts all servers simultaneously, applications would retry and continue their respective operations upon upgrade.
-- `mc admin update` is disabled in kubernetes/container environments, container environments provide their own mechanisms to rollout of updates.
-- In the case of federated setups `mc admin update` should be run against each cluster individually. Avoid updating `mc` to any new releases until all clusters have been successfully updated.
 - If using `kes` as KMS with MinIO, just replace the binary and restart `kes` more information about `kes` can be found [here](https://github.com/minio/kes/wiki)
-- If using Vault as KMS with MinIO, ensure you have followed the Vault upgrade procedure outlined here: https://www.vaultproject.io/docs/upgrading/index.html
-- If using etcd with MinIO for the federation, ensure you have followed the etcd upgrade procedure outlined here: https://github.com/etcd-io/etcd/blob/master/Documentation/upgrades/upgrading-etcd.md
 
 # Explore Further
 - [MinIO Erasure Code QuickStart Guide](https://docs.min.io/docs/minio-erasure-code-quickstart-guide)
@@ -257,5 +255,6 @@ mc admin update <minio alias, e.g., myminio>
 Please follow MinIO [Contributor's Guide](https://github.com/minio/minio/blob/master/CONTRIBUTING.md)
 
 # License
-MinIO source is licensed under the GNU AGPLv3 license that can be found in the [LICENSE](https://github.com/minio/minio/blob/master/LICENSE) file.
-MinIO [Documentation](https://github.com/minio/minio/tree/master/docs) © 2021 by MinIO, Inc is licensed under [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/).
+- MinIO source is licensed under the GNU AGPLv3 license that can be found in the [LICENSE](https://github.com/minio/minio/blob/master/LICENSE) file.
+- MinIO [Documentation](https://github.com/minio/minio/tree/master/docs) © 2021 by MinIO, Inc is licensed under [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/).
+- [License Compliance](https://github.com/minio/minio/blob/master/COMPLIANCE.md)

cmd/erasure-bucket.go

@@ -64,8 +64,7 @@ func (er erasureObjects) MakeBucketWithLocation(ctx context.Context, bucket stri
 		}, index)
 	}
 
-	writeQuorum := getWriteQuorum(len(storageDisks))
-	err := reduceWriteQuorumErrs(ctx, g.Wait(), bucketOpIgnoredErrs, writeQuorum)
+	err := reduceWriteQuorumErrs(ctx, g.Wait(), bucketOpIgnoredErrs, er.defaultWQuorum())
 	return toObjectErr(err, bucket)
 }
 
@@ -121,8 +120,7 @@ func (er erasureObjects) getBucketInfo(ctx context.Context, bucketName string) (
 	// reduce to one error based on read quorum.
 	// `nil` is deliberately passed for ignoredErrs
 	// because these errors were already ignored.
-	readQuorum := getReadQuorum(len(storageDisks))
-	return BucketInfo{}, reduceReadQuorumErrs(ctx, errs, nil, readQuorum)
+	return BucketInfo{}, reduceReadQuorumErrs(ctx, errs, nil, er.defaultRQuorum())
 }
 
 // GetBucketInfo - returns BucketInfo for a bucket.
@@ -167,8 +165,7 @@ func (er erasureObjects) DeleteBucket(ctx context.Context, bucket string, opts D
 		return nil
 	}
 
-	writeQuorum := getWriteQuorum(len(storageDisks))
-	err := reduceWriteQuorumErrs(ctx, dErrs, bucketOpIgnoredErrs, writeQuorum)
+	err := reduceWriteQuorumErrs(ctx, dErrs, bucketOpIgnoredErrs, er.defaultWQuorum())
 	if err == errErasureWriteQuorum && !opts.NoRecreate {
 		undoDeleteBucket(storageDisks, bucket)
 	}

cmd/erasure-healing.go

@@ -286,12 +286,10 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 
 	// Initialize heal result object
 	result = madmin.HealResultItem{
-		Type:         madmin.HealItemObject,
-		Bucket:       bucket,
-		Object:       object,
-		DiskCount:    len(storageDisks),
-		ParityBlocks: er.defaultParityCount,
-		DataBlocks:   len(storageDisks) - er.defaultParityCount,
+		Type:      madmin.HealItemObject,
+		Bucket:    bucket,
+		Object:    object,
+		DiskCount: len(storageDisks),
 	}
 
 	if !opts.NoLock {
@@ -306,18 +304,27 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 
 	// Re-read when we have lock...
 	partsMetadata, errs := readAllFileInfo(ctx, storageDisks, bucket, object, versionID, true)
+	if isAllNotFound(errs) {
+		// Nothing to do, file is already gone.
+		return er.defaultHealResult(FileInfo{}, storageDisks, storageEndpoints,
+			errs, bucket, object, versionID), nil
+	}
 
-	if _, err = getLatestFileInfo(ctx, partsMetadata, errs); err != nil {
+	readQuorum, _, err := objectQuorumFromMeta(ctx, partsMetadata, errs, er.defaultParityCount)
+	if err != nil {
 		return er.purgeObjectDangling(ctx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
 	}
 
+	result.ParityBlocks = result.DiskCount - readQuorum
+	result.DataBlocks = readQuorum
+
 	// List of disks having latest version of the object er.meta
 	// (by modtime).
 	onlineDisks, modTime := listOnlineDisks(storageDisks, partsMetadata, errs)
 
 	// Latest FileInfo for reference. If a valid metadata is not
 	// present, it is as good as object not found.
-	latestMeta, err := pickValidFileInfo(ctx, partsMetadata, modTime, result.DataBlocks)
+	latestMeta, err := pickValidFileInfo(ctx, partsMetadata, modTime, readQuorum)
 	if err != nil {
 		return result, toObjectErr(err, bucket, object, versionID)
 	}
@@ -392,13 +399,9 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 	}
 
 	if isAllNotFound(errs) {
-		err = toObjectErr(errFileNotFound, bucket, object)
-		if versionID != "" {
-			err = toObjectErr(errFileVersionNotFound, bucket, object, versionID)
-		}
 		// File is fully gone, fileInfo is empty.
 		return er.defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs,
-			bucket, object, versionID), err
+			bucket, object, versionID), nil
 	}
 
 	// If less than read quorum number of disks have all the parts
@@ -663,7 +666,7 @@ func (er erasureObjects) healObjectDir(ctx context.Context, bucket, object strin
 	}
 	if dryRun || danglingObject || isAllNotFound(errs) {
 		// Nothing to do, file is already gone.
-		return hr, toObjectErr(errFileNotFound, bucket, object)
+		return hr, nil
 	}
 	for i, err := range errs {
 		if err == errVolumeNotFound || err == errFileNotFound {
@@ -768,8 +771,15 @@ func statAllDirs(ctx context.Context, storageDisks []StorageAPI, bucket, prefix
 // A 0 length slice will always return false.
 func isAllNotFound(errs []error) bool {
 	for _, err := range errs {
-		if errors.Is(err, errFileNotFound) || errors.Is(err, errVolumeNotFound) || errors.Is(err, errFileVersionNotFound) {
-			continue
+		if err != nil {
+			switch err.Error() {
+			case errFileNotFound.Error():
+				fallthrough
+			case errVolumeNotFound.Error():
+				fallthrough
+			case errFileVersionNotFound.Error():
+				continue
+			}
 		}
 		return false
 	}

cmd/erasure-healing_test.go

@@ -329,6 +329,131 @@ func TestHealingDanglingObject(t *testing.T) {
 	}
 }
 
+func TestHealCorrectQuorum(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	resetGlobalHealState()
+	defer resetGlobalHealState()
+
+	nDisks := 32
+	fsDirs, err := getRandomDisks(nDisks)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer removeRoots(fsDirs)
+
+	pools := mustGetPoolEndpoints(fsDirs[:16]...)
+	pools = append(pools, mustGetPoolEndpoints(fsDirs[16:]...)...)
+
+	// Everything is fine, should return nil
+	objLayer, _, err := initObjectLayer(ctx, pools)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	bucket := getRandomBucketName()
+	object := getRandomObjectName()
+	data := bytes.Repeat([]byte("a"), 5*1024*1024)
+	var opts ObjectOptions
+
+	err = objLayer.MakeBucketWithLocation(ctx, bucket, BucketOptions{})
+	if err != nil {
+		t.Fatalf("Failed to make a bucket - %v", err)
+	}
+
+	// Create an object with multiple parts uploaded in decreasing
+	// part number.
+	uploadID, err := objLayer.NewMultipartUpload(ctx, bucket, object, opts)
+	if err != nil {
+		t.Fatalf("Failed to create a multipart upload - %v", err)
+	}
+
+	var uploadedParts []CompletePart
+	for _, partID := range []int{2, 1} {
+		pInfo, err1 := objLayer.PutObjectPart(ctx, bucket, object, uploadID, partID, mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), opts)
+		if err1 != nil {
+			t.Fatalf("Failed to upload a part - %v", err1)
+		}
+		uploadedParts = append(uploadedParts, CompletePart{
+			PartNumber: pInfo.PartNumber,
+			ETag:       pInfo.ETag,
+		})
+	}
+
+	_, err = objLayer.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, ObjectOptions{})
+	if err != nil {
+		t.Fatalf("Failed to complete multipart upload - %v", err)
+	}
+
+	cfgFile := pathJoin(bucketConfigPrefix, bucket, ".test.bin")
+	if err = saveConfig(ctx, objLayer, cfgFile, data); err != nil {
+		t.Fatal(err)
+	}
+
+	hopts := madmin.HealOpts{
+		DryRun:   false,
+		Remove:   true,
+		ScanMode: madmin.HealNormalScan,
+	}
+
+	// Test 1: Remove the object backend files from the first disk.
+	z := objLayer.(*erasureServerPools)
+	for _, set := range z.serverPools {
+		er := set.sets[0]
+		erasureDisks := er.getDisks()
+
+		fileInfos, errs := readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
+		nfi, err := getLatestFileInfo(ctx, fileInfos, errs)
+		if errors.Is(err, errFileNotFound) {
+			continue
+		}
+		if err != nil {
+			t.Fatalf("Failed to getLatestFileInfo - %v", err)
+		}
+
+		for i := 0; i < nfi.Erasure.ParityBlocks; i++ {
+			erasureDisks[i].Delete(context.Background(), bucket, pathJoin(object, xlStorageFormatFile), false)
+		}
+
+		// Try healing now, it should heal the content properly.
+		_, err = objLayer.HealObject(ctx, bucket, object, "", hopts)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		fileInfos, errs = readAllFileInfo(ctx, erasureDisks, bucket, object, "", false)
+		if countErrs(errs, nil) != len(fileInfos) {
+			t.Fatal("Expected all xl.meta healed, but partial heal detected")
+		}
+
+		fileInfos, errs = readAllFileInfo(ctx, erasureDisks, minioMetaBucket, cfgFile, "", false)
+		nfi, err = getLatestFileInfo(ctx, fileInfos, errs)
+		if errors.Is(err, errFileNotFound) {
+			continue
+		}
+		if err != nil {
+			t.Fatalf("Failed to getLatestFileInfo - %v", err)
+		}
+
+		for i := 0; i < nfi.Erasure.ParityBlocks; i++ {
+			erasureDisks[i].Delete(context.Background(), minioMetaBucket, pathJoin(cfgFile, xlStorageFormatFile), false)
+		}
+
+		// Try healing now, it should heal the content properly.
+		_, err = objLayer.HealObject(ctx, minioMetaBucket, cfgFile, "", hopts)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		fileInfos, errs = readAllFileInfo(ctx, erasureDisks, minioMetaBucket, cfgFile, "", false)
+		if countErrs(errs, nil) != len(fileInfos) {
+			t.Fatal("Expected all xl.meta healed, but partial heal detected")
+		}
+	}
+}
+
 func TestHealObjectCorruptedPools(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

cmd/erasure-object.go

@@ -498,7 +498,7 @@ func (er erasureObjects) getObjectInfo(ctx context.Context, bucket, object strin
 func (er erasureObjects) getObjectInfoAndQuorum(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, wquorum int, err error) {
 	fi, _, _, err := er.getObjectFileInfo(ctx, bucket, object, opts, false)
 	if err != nil {
-		return objInfo, getWriteQuorum(len(er.getDisks())), toObjectErr(err, bucket, object)
+		return objInfo, er.defaultWQuorum(), toObjectErr(err, bucket, object)
 	}
 
 	wquorum = fi.Erasure.DataBlocks
@@ -1113,7 +1113,7 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
 		// class for objects which have reduced quorum
 		// storage class only needs to be honored for
 		// Read() requests alone which we already do.
-		writeQuorums[i] = getWriteQuorum(len(storageDisks))
+		writeQuorums[i] = er.defaultWQuorum()
 	}
 
 	versionsMap := make(map[string]FileInfoVersions, len(objects))
@@ -1600,7 +1600,7 @@ func (er erasureObjects) updateObjectMeta(ctx context.Context, bucket, object st
 	// Wait for all the routines.
 	mErrs := g.Wait()
 
-	return reduceWriteQuorumErrs(ctx, mErrs, objectOpIgnoredErrs, getWriteQuorum(len(onlineDisks)))
+	return reduceWriteQuorumErrs(ctx, mErrs, objectOpIgnoredErrs, er.defaultWQuorum())
 }
 
 // DeleteObjectTags - delete object tags from an existing object

cmd/erasure.go

@@ -84,6 +84,19 @@ func (er erasureObjects) Shutdown(ctx context.Context) error {
 	return nil
 }
 
+// defaultWQuorum write quorum based on setDriveCount and defaultParityCount
+func (er erasureObjects) defaultWQuorum() int {
+	dataCount := er.setDriveCount - er.defaultParityCount
+	if dataCount == er.defaultParityCount {
+		return dataCount + 1
+	}
+	return dataCount
+}
+
+func (er erasureObjects) defaultRQuorum() int {
+	return er.setDriveCount - er.defaultParityCount
+}
+
 // byDiskTotal is a collection satisfying sort.Interface.
 type byDiskTotal []madmin.Disk
 

cmd/metacache-set.go

@@ -544,7 +544,7 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul
 	// Special case: ask all disks if the drive count is 4
 	if askDisks == -1 || er.setDriveCount == 4 {
 		askDisks = len(disks) // with 'strict' quorum list on all online disks.
-		listingQuorum = getReadQuorum(er.setDriveCount)
+		listingQuorum = er.defaultRQuorum()
 	}
 	if askDisks == 0 {
 		askDisks = globalAPIConfig.getListQuorum()

docs/orchestration/docker-compose/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2021-12-20T22-07-16Z
+  image: quay.io/minio/minio:RELEASE.2021-12-27T07-23-18Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

docs/tls/README.md

@@ -148,7 +148,7 @@ DNS.1 = localhost
 Run `openssl` by specifying the configuration file and enter a passphrase if prompted:
 
 ```sh
-openssl req -new -x509 -nodes -days 730 -key private.key -out public.crt -config openssl.conf
+openssl req -new -x509 -nodes -days 730 -keyout private.key -out public.crt -config openssl.conf
 ```
 
 ### <a name="using-gnu-tls"></a>3.3 Use GnuTLS (for Windows) to Generate a Certificate

internal/config/storageclass/storage-class.go

@@ -234,7 +234,7 @@ func (sCfg Config) GetParityForSC(sc string) (parity int) {
 }
 
 // Update update storage-class with new config
-func (sCfg Config) Update(newCfg Config) {
+func (sCfg *Config) Update(newCfg Config) {
 	ConfigLock.Lock()
 	defer ConfigLock.Unlock()
 	sCfg.RRS = newCfg.RRS