Revert heal locks (#12365)

A lot of healing is likely to be on non-existing objects and 
locks are very expensive and will slow down scanning 
significantly.

In cases where all are valid or, all are broken allow 
rejection without locking.

Keep the existing behavior, but move the check for 
dangling objects to after the lock has been acquired.

```
	_, err = getLatestFileInfo(ctx, partsMetadata, errs)
	if err != nil {
		return er.purgeObjectDangling(ctx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
	}
```

Revert "heal: Hold lock when reading xl.meta from disks (#12362)"

This reverts commit abd32065aa

.golangci.yml

@@ -12,13 +12,17 @@ linters:
     - goimports
     - misspell
     - govet
-    - golint
+    - revive
     - ineffassign
     - gosimple
     - deadcode
     - structcheck
     - gomodguard
     - gofmt
+    - unused
+    - structcheck
+    - unconvert
+    - varcheck
 
 issues:
   exclude-use-default: false
@@ -26,11 +30,5 @@ issues:
       - should have a package comment
       - error strings should not be capitalized or end with punctuation or a newline
 
-run:
-  skip-dirs:
-    - pkg/rpc
-    - pkg/argon2
-    - pkg/s3select/internal
-
 service:
   golangci-lint-version: 1.20.0 # use the fixed version to not introduce new linters unexpectedly

Makefile

@@ -16,7 +16,7 @@ checks:
 
 getdeps:
 	@mkdir -p ${GOPATH}/bin
-	@which golangci-lint 1>/dev/null || (echo "Installing golangci-lint" && curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(GOPATH)/bin v1.27.0)
+	@echo "Installing golangci-lint" && curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(GOPATH)/bin v1.40.1
 	@which msgp 1>/dev/null || (echo "Installing msgp" && go install -v github.com/tinylib/msgp@v1.1.3)
 	@which stringer 1>/dev/null || (echo "Installing stringer" && go install -v golang.org/x/tools/cmd/stringer)
 

cmd/admin-heal-ops.go

@@ -486,22 +486,6 @@ func newHealSequence(ctx context.Context, bucket, objPrefix, clientAddr string,
 	}
 }
 
-// resetHealStatusCounters - reset the healSequence status counters between
-// each monthly background heal scanning activity.
-// This is used only in case of Background healing scenario, where
-// we use a single long running healSequence which reactively heals
-// objects passed to the SourceCh.
-func (h *healSequence) resetHealStatusCounters() {
-	h.mutex.Lock()
-	defer h.mutex.Unlock()
-
-	h.currentStatus.Items = []madmin.HealResultItem{}
-	h.lastSentResultIndex = 0
-	h.scannedItemsMap = make(map[madmin.HealItemType]int64)
-	h.healedItemsMap = make(map[madmin.HealItemType]int64)
-	h.healFailedItemsMap = make(map[string]int64)
-}
-
 // getScannedItemsCount - returns a count of all scanned items
 func (h *healSequence) getScannedItemsCount() int64 {
 	var count int64

cmd/api-headers_test.go

@@ -31,7 +31,7 @@ func TestNewRequestID(t *testing.T) {
 
 	var e rune
 	for _, char := range id {
-		e = rune(char)
+		e = char
 
 		// Ensure that it is alphanumeric, in this case, between 0-9 and A-Z.
 		if !(('0' <= e && e <= '9') || ('A' <= e && e <= 'Z')) {

cmd/benchmark-utils_test.go

@@ -111,7 +111,7 @@ func runPutObjectPartBenchmark(b *testing.B, obj ObjectLayer, partSize int) {
 			} else {
 				textPartData = textData[j*partSize:]
 			}
-			md5hex := getMD5Hash([]byte(textPartData))
+			md5hex := getMD5Hash(textPartData)
 			var partInfo PartInfo
 			partInfo, err = obj.PutObjectPart(context.Background(), bucket, object, uploadID, j,
 				mustGetPutObjReader(b, bytes.NewReader(textPartData), int64(len(textPartData)), md5hex, sha256hex), ObjectOptions{})
@@ -206,7 +206,7 @@ func runPutObjectBenchmarkParallel(b *testing.B, obj ObjectLayer, objSize int) {
 	// generate md5sum for the generated data.
 	// md5sum of the data to written is required as input for PutObject.
 
-	md5hex := getMD5Hash([]byte(textData))
+	md5hex := getMD5Hash(textData)
 	sha256hex := ""
 
 	// benchmark utility which helps obtain number of allocations and bytes allocated per ops.

cmd/bitrot-streaming.go

@@ -31,14 +31,6 @@ import (
 	"github.com/minio/minio/pkg/ioutil"
 )
 
-type errHashMismatch struct {
-	message string
-}
-
-func (err *errHashMismatch) Error() string {
-	return err.message
-}
-
 // Calculates bitrot in chunks and writes the hash into the stream.
 type streamingBitrotWriter struct {
 	iow          io.WriteCloser

cmd/bucket-targets.go

@@ -352,7 +352,7 @@ func (sys *BucketTargetSys) getRemoteARN(bucket string, target *madmin.BucketTar
 			return tgt.Arn
 		}
 	}
-	if !madmin.ServiceType(target.Type).IsValid() {
+	if !target.Type.IsValid() {
 		return ""
 	}
 	return generateARN(target)

cmd/config-migrate.go

@@ -45,7 +45,7 @@ import (
 	"github.com/minio/minio/pkg/quick"
 )
 
-// DO NOT EDIT following message template, please open a github issue to discuss instead.
+// DO NOT EDIT following message template, please open a GitHub issue to discuss instead.
 var configMigrateMSGTemplate = "Configuration file %s migrated from version '%s' to '%s' successfully."
 
 // Save config file to corresponding backend

cmd/config/config.go

@@ -315,10 +315,7 @@ func (c Config) DelFrom(r io.Reader) error {
 			return err
 		}
 	}
-	if err := scanner.Err(); err != nil {
-		return err
-	}
-	return nil
+	return scanner.Err()
 }
 
 // ReadConfig - read content from input and write into c.

cmd/config/dns/etcd_dns.go

@@ -137,7 +137,7 @@ func (c *CoreDNS) list(key string, domain bool) ([]SrvRecord, error) {
 	var srvRecords []SrvRecord
 	for _, n := range r.Kvs {
 		var srvRecord SrvRecord
-		if err = json.Unmarshal([]byte(n.Value), &srvRecord); err != nil {
+		if err = json.Unmarshal(n.Value, &srvRecord); err != nil {
 			return nil, err
 		}
 		srvRecord.Key = strings.TrimPrefix(string(n.Key), key)

cmd/config/identity/ldap/config.go

@@ -189,7 +189,7 @@ func (l *Config) lookupBind(conn *ldap.Conn) error {
 		err = conn.Bind(l.LookupBindDN, l.LookupBindPassword)
 	}
 	if ldap.IsErrorWithCode(err, 49) {
-		return fmt.Errorf("LDAP Lookup Bind user invalid credentials error: %v", err)
+		return fmt.Errorf("LDAP Lookup Bind user invalid credentials error: %w", err)
 	}
 	return err
 }
@@ -214,7 +214,7 @@ func (l *Config) usernameFormatsBind(conn *ldap.Conn, username, password string)
 			bindDistNames = append(bindDistNames, bindDN)
 			successCount++
 		} else if !ldap.IsErrorWithCode(errs[i], 49) {
-			return "", fmt.Errorf("LDAP Bind request failed with unexpected error: %v", errs[i])
+			return "", fmt.Errorf("LDAP Bind request failed with unexpected error: %w", errs[i])
 		}
 	}
 	if successCount == 0 {
@@ -279,7 +279,7 @@ func (l *Config) searchForUserGroups(conn *ldap.Conn, username, bindDN string) (
 			var newGroups []string
 			newGroups, err := getGroups(conn, searchRequest)
 			if err != nil {
-				errRet := fmt.Errorf("Error finding groups of %s: %v", bindDN, err)
+				errRet := fmt.Errorf("Error finding groups of %s: %w", bindDN, err)
 				return nil, errRet
 			}
 
@@ -341,14 +341,14 @@ func (l *Config) Bind(username, password string) (string, []string, error) {
 		// Lookup user DN
 		bindDN, err = l.lookupUserDN(conn, username)
 		if err != nil {
-			errRet := fmt.Errorf("Unable to find user DN: %s", err)
+			errRet := fmt.Errorf("Unable to find user DN: %w", err)
 			return "", nil, errRet
 		}
 
 		// Authenticate the user credentials.
 		err = conn.Bind(bindDN, password)
 		if err != nil {
-			errRet := fmt.Errorf("LDAP auth failed for DN %s: %v", bindDN, err)
+			errRet := fmt.Errorf("LDAP auth failed for DN %s: %w", bindDN, err)
 			return "", nil, errRet
 		}
 
@@ -366,7 +366,7 @@ func (l *Config) Bind(username, password string) (string, []string, error) {
 		// Bind to the successful bindDN again.
 		err = conn.Bind(bindDN, password)
 		if err != nil {
-			errRet := fmt.Errorf("LDAP conn failed though auth for DN %s succeeded: %v", bindDN, err)
+			errRet := fmt.Errorf("LDAP conn failed though auth for DN %s succeeded: %w", bindDN, err)
 			return "", nil, errRet
 		}
 	}
@@ -423,12 +423,12 @@ func (l Config) GetExpiryDuration() time.Duration {
 func (l Config) testConnection() error {
 	conn, err := l.Connect()
 	if err != nil {
-		return fmt.Errorf("Error creating connection to LDAP server: %v", err)
+		return fmt.Errorf("Error creating connection to LDAP server: %w", err)
 	}
 	defer conn.Close()
 	if l.isUsingLookupBind {
 		if err = l.lookupBind(conn); err != nil {
-			return fmt.Errorf("Error connecting as LDAP Lookup Bind user: %v", err)
+			return fmt.Errorf("Error connecting as LDAP Lookup Bind user: %w", err)
 		}
 		return nil
 	}
@@ -449,7 +449,7 @@ func (l Config) testConnection() error {
 	} else if strings.HasPrefix(err.Error(), "All username formats failed due to invalid credentials: ") {
 		return nil
 	}
-	return fmt.Errorf("LDAP connection test error: %v", err)
+	return fmt.Errorf("LDAP connection test error: %w", err)
 }
 
 // Enabled returns if jwks is enabled.
@@ -547,7 +547,7 @@ func Lookup(kvs config.KVS, rootCAs *x509.CertPool) (l Config, err error) {
 
 	// Test connection to LDAP server.
 	if err := l.testConnection(); err != nil {
-		return l, fmt.Errorf("Connection test for LDAP server failed: %v", err)
+		return l, fmt.Errorf("Connection test for LDAP server failed: %w", err)
 	}
 
 	// Group search params configuration

cmd/config/notify/legacy.go

@@ -27,7 +27,7 @@ import (
 )
 
 // SetNotifyKafka - helper for config migration from older config.
-func SetNotifyKafka(s config.Config, kName string, cfg target.KafkaArgs) error {
+func SetNotifyKafka(s config.Config, name string, cfg target.KafkaArgs) error {
 	if !cfg.Enable {
 		return nil
 	}
@@ -36,7 +36,7 @@ func SetNotifyKafka(s config.Config, kName string, cfg target.KafkaArgs) error {
 		return err
 	}
 
-	s[config.NotifyKafkaSubSys][kName] = config.KVS{
+	s[config.NotifyKafkaSubSys][name] = config.KVS{
 		config.KV{
 			Key:   config.Enable,
 			Value: config.EnableOn,

cmd/data-usage-cache.go

@@ -218,12 +218,6 @@ func (h dataUsageHash) mod(cycle uint32, cycles uint32) bool {
 	return uint32(xxhash.Sum64String(string(h)))%cycles == cycle%cycles
 }
 
-// addChildString will add a child based on its name.
-// If it already exists it will not be added again.
-func (e *dataUsageEntry) addChildString(name string) {
-	e.addChild(hashPath(name))
-}
-
 // addChild will add a child based on its hash.
 // If it already exists it will not be added again.
 func (e *dataUsageEntry) addChild(hash dataUsageHash) {
@@ -291,17 +285,6 @@ func (d *dataUsageCache) searchParent(h dataUsageHash) *dataUsageHash {
 	return nil
 }
 
-// Returns nil if not found.
-func (d *dataUsageCache) subCache(path string) dataUsageCache {
-	dst := dataUsageCache{Info: dataUsageCacheInfo{
-		Name:        path,
-		LastUpdate:  d.Info.LastUpdate,
-		BloomFilter: d.Info.BloomFilter,
-	}}
-	dst.copyWithChildren(d, dataUsageHash(hashPath(path).Key()), nil)
-	return dst
-}
-
 // deleteRecursive will delete an entry recursively, but not change its parent.
 func (d *dataUsageCache) deleteRecursive(h dataUsageHash) {
 	if existing, ok := d.Cache[h.String()]; ok {
@@ -313,37 +296,6 @@ func (d *dataUsageCache) deleteRecursive(h dataUsageHash) {
 	}
 }
 
-// deleteChildren will delete any children, but not the entry itself.
-func (d *dataUsageCache) deleteChildren(h dataUsageHash) {
-	if existing, ok := d.Cache[h.String()]; ok {
-		for child := range existing.Children {
-			d.deleteRecursive(dataUsageHash(child))
-		}
-	}
-}
-
-// replaceRootChild will replace the child of root in d with the root of 'other'.
-func (d *dataUsageCache) replaceRootChild(other dataUsageCache) {
-	otherRoot := other.root()
-	if otherRoot == nil {
-		logger.LogIf(GlobalContext, errors.New("replaceRootChild: Source has no root"))
-		return
-	}
-	thisRoot := d.root()
-	if thisRoot == nil {
-		logger.LogIf(GlobalContext, errors.New("replaceRootChild: Root of current not found"))
-		return
-	}
-	thisRootHash := d.rootHash()
-	otherRootHash := other.rootHash()
-	if thisRootHash == otherRootHash {
-		logger.LogIf(GlobalContext, errors.New("replaceRootChild: Root of child matches root of destination"))
-		return
-	}
-	d.deleteRecursive(other.rootHash())
-	d.copyWithChildren(&other, other.rootHash(), &thisRootHash)
-}
-
 // keepBuckets will keep only the buckets specified specified by delete all others.
 func (d *dataUsageCache) keepBuckets(b []BucketInfo) {
 	lu := make(map[dataUsageHash]struct{})
@@ -415,16 +367,6 @@ func (d *dataUsageCache) replace(path, parent string, e dataUsageEntry) {
 	}
 }
 
-// listCache will return all cache paths.
-func (d *dataUsageCache) listCache() []string {
-	dst := make([]string, 0, len(d.Cache))
-	for k := range d.Cache {
-		dst = append(dst, k)
-	}
-	sort.Strings(dst)
-	return dst
-}
-
 // replaceHashed add or replaces an entry to the cache based on its hash.
 // If a parent is specified it will be added to that if not already there.
 // If the parent does not exist, it will be added.

cmd/erasure-common.go

@@ -80,24 +80,6 @@ func (er erasureObjects) getOnlineDisks() (newDisks []StorageAPI) {
 	return newDisks
 }
 
-// getLoadBalancedNDisks - fetches load balanced (sufficiently randomized) disk slice
-// with N disks online. If ndisks is zero or negative, then it will returns all disks,
-// same if ndisks is greater than the number of all disks.
-func (er erasureObjects) getLoadBalancedNDisks(ndisks int) (newDisks []StorageAPI) {
-	disks := er.getLoadBalancedDisks(ndisks != -1)
-	for _, disk := range disks {
-		if disk == nil {
-			continue
-		}
-		newDisks = append(newDisks, disk)
-		ndisks--
-		if ndisks == 0 {
-			break
-		}
-	}
-	return
-}
-
 // getLoadBalancedDisks - fetches load balanced (sufficiently randomized) disk slice.
 // ensures to skip disks if they are not healing and online.
 func (er erasureObjects) getLoadBalancedDisks(optimized bool) []StorageAPI {

cmd/erasure-healing-common.go

@@ -189,6 +189,35 @@ func getLatestFileInfo(ctx context.Context, partsMetadata []FileInfo, errs []err
 	return latestFileInfo, nil
 }
 
+// fileInfoConsistent whether all fileinfos are consistent with each other.
+// Will return false if any fileinfo mismatches.
+func fileInfoConsistent(ctx context.Context, partsMetadata []FileInfo, errs []error) bool {
+	// There should be atleast half correct entries, if not return failure
+	if reducedErr := reduceReadQuorumErrs(ctx, errs, nil, len(partsMetadata)/2); reducedErr != nil {
+		return false
+	}
+	if len(partsMetadata) == 1 {
+		return true
+	}
+	// Reference
+	ref := partsMetadata[0]
+	if !ref.IsValid() {
+		return false
+	}
+	for _, meta := range partsMetadata[1:] {
+		if !meta.IsValid() {
+			return false
+		}
+		if !meta.ModTime.Equal(ref.ModTime) {
+			return false
+		}
+		if meta.DataDir != ref.DataDir {
+			return false
+		}
+	}
+	return true
+}
+
 // disksWithAllParts - This function needs to be called with
 // []StorageAPI returned by listOnlineDisks. Returns,
 //

cmd/erasure-healing.go

@@ -267,6 +267,10 @@ 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)
 
+	_, err = getLatestFileInfo(ctx, partsMetadata, errs)
+	if err != nil {
+		return er.purgeObjectDangling(ctx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
+	}
 	// List of disks having latest version of the object er.meta
 	// (by modtime).
 	_, modTime, dataDir := listOnlineDisks(storageDisks, partsMetadata, errs)
@@ -530,6 +534,7 @@ func (er erasureObjects) healObject(ctx context.Context, bucket string, object s
 		if partsMetadata[i].IsRemote() {
 			partsMetadata[i].DataDir = ""
 		}
+
 		// Attempt a rename now from healed data to final location.
 		if err = disk.RenameData(ctx, minioMetaTmpBucket, tmpID, partsMetadata[i], bucket, object); err != nil {
 			logger.LogIf(ctx, err)
@@ -642,8 +647,19 @@ func defaultHealResult(lfi FileInfo, storageDisks []StorageAPI, storageEndpoints
 		VersionID: versionID,
 		DiskCount: len(storageDisks),
 	}
+
 	if lfi.IsValid() {
 		result.ObjectSize = lfi.Size
+		result.ParityBlocks = lfi.Erasure.ParityBlocks
+	} else {
+		// Default to most common configuration for erasure blocks.
+		result.ParityBlocks = defaultParityCount
+	}
+	result.DataBlocks = len(storageDisks) - result.ParityBlocks
+
+	if errs == nil {
+		// No disks related errors are provided, quit
+		return result
 	}
 
 	for index, disk := range storageDisks {
@@ -758,8 +774,13 @@ func (er erasureObjects) purgeObjectDangling(ctx context.Context, bucket, object
 	// remove we simply delete it from namespace.
 	m, ok := isObjectDangling(metaArr, errs, dataErrs)
 	if ok {
-		writeQuorum := m.Erasure.DataBlocks
-		if m.Erasure.DataBlocks == 0 || m.Erasure.DataBlocks == m.Erasure.ParityBlocks {
+		parityBlocks := m.Erasure.ParityBlocks
+		if m.Erasure.ParityBlocks == 0 {
+			parityBlocks = er.defaultParityCount
+		}
+		dataBlocks := len(storageDisks) - parityBlocks
+		writeQuorum := dataBlocks
+		if dataBlocks == parityBlocks {
 			writeQuorum++
 		}
 		var err error
@@ -879,15 +900,14 @@ func (er erasureObjects) HealObject(ctx context.Context, bucket, object, version
 	storageDisks := er.getDisks()
 	storageEndpoints := er.getEndpoints()
 
-	// Read metadata files from all the disks
-
 	// When versionID is empty, we read directly from the `null` versionID for healing.
 	if versionID == "" {
 		versionID = nullVersionID
 	}
 
+	// Perform quick read without lock.
+	// This allows to quickly check if all is ok or all are missing.
 	partsMetadata, errs := readAllFileInfo(healCtx, storageDisks, bucket, object, versionID, false)
-
 	if isAllNotFound(errs) {
 		err = toObjectErr(errFileNotFound, bucket, object)
 		if versionID != "" {
@@ -897,9 +917,12 @@ func (er erasureObjects) HealObject(ctx context.Context, bucket, object, version
 		return defaultHealResult(FileInfo{}, storageDisks, storageEndpoints, errs, bucket, object, versionID, er.defaultParityCount), err
 	}
 
-	_, err = getLatestFileInfo(healCtx, partsMetadata, errs)
-	if err != nil {
-		return er.purgeObjectDangling(healCtx, bucket, object, versionID, partsMetadata, errs, []error{}, opts)
+	// Return early if all ok and not deep scanning.
+	if opts.ScanMode == madmin.HealNormalScan && fileInfoConsistent(ctx, partsMetadata, errs) {
+		fi, err := getLatestFileInfo(ctx, partsMetadata, errs)
+		if err == nil && fi.VersionID == versionID {
+			return defaultHealResult(fi, storageDisks, storageEndpoints, errs, bucket, object, versionID, fi.Erasure.ParityBlocks), nil
+		}
 	}
 
 	// Heal the object.

cmd/erasure-metadata.go

@@ -239,7 +239,11 @@ func (fi FileInfo) ObjectToPartOffset(ctx context.Context, offset int64) (partIn
 	return 0, 0, InvalidRange{}
 }
 
-func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.Time, dataDir string, quorum int) (xmv FileInfo, e error) {
+func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.Time, dataDir string, quorum int) (FileInfo, error) {
+	// with less quorum return error.
+	if quorum < 2 {
+		return FileInfo{}, errErasureReadQuorum
+	}
 	metaHashes := make([]string, len(metaArr))
 	h := sha256.New()
 	for i, meta := range metaArr {
@@ -278,7 +282,9 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.
 
 	for i, hash := range metaHashes {
 		if hash == maxHash {
-			return metaArr[i], nil
+			if metaArr[i].IsValid() {
+				return metaArr[i], nil
+			}
 		}
 	}
 
@@ -287,7 +293,7 @@ func findFileInfoInQuorum(ctx context.Context, metaArr []FileInfo, modTime time.
 
 // pickValidFileInfo - picks one valid FileInfo content and returns from a
 // slice of FileInfo.
-func pickValidFileInfo(ctx context.Context, metaArr []FileInfo, modTime time.Time, dataDir string, quorum int) (xmv FileInfo, e error) {
+func pickValidFileInfo(ctx context.Context, metaArr []FileInfo, modTime time.Time, dataDir string, quorum int) (FileInfo, error) {
 	return findFileInfoInQuorum(ctx, metaArr, modTime, dataDir, quorum)
 }
 
@@ -329,12 +335,16 @@ func objectQuorumFromMeta(ctx context.Context, partsMetaData []FileInfo, errs []
 		return 0, 0, err
 	}
 
-	dataBlocks := latestFileInfo.Erasure.DataBlocks
+	if !latestFileInfo.IsValid() {
+		return 0, 0, errErasureReadQuorum
+	}
+
 	parityBlocks := globalStorageClass.GetParityForSC(latestFileInfo.Metadata[xhttp.AmzStorageClass])
 	if parityBlocks <= 0 {
 		parityBlocks = defaultParityCount
 	}
 
+	dataBlocks := len(partsMetaData) - parityBlocks
 	writeQuorum := dataBlocks
 	if dataBlocks == parityBlocks {
 		writeQuorum++

cmd/erasure-metadata_test.go

@@ -176,29 +176,39 @@ func TestFindFileInfoInQuorum(t *testing.T) {
 	}
 
 	tests := []struct {
-		fis         []FileInfo
-		modTime     time.Time
-		dataDir     string
-		expectedErr error
+		fis            []FileInfo
+		modTime        time.Time
+		dataDir        string
+		expectedErr    error
+		expectedQuorum int
 	}{
 		{
-			fis:         getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21"),
-			modTime:     time.Unix(1603863445, 0),
-			dataDir:     "36a21454-a2ca-11eb-bbaa-93a81c686f21",
-			expectedErr: nil,
+			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21"),
+			modTime:        time.Unix(1603863445, 0),
+			dataDir:        "36a21454-a2ca-11eb-bbaa-93a81c686f21",
+			expectedErr:    nil,
+			expectedQuorum: 8,
 		},
 		{
-			fis:         getNFInfo(16, 7, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21"),
-			modTime:     time.Unix(1603863445, 0),
-			dataDir:     "36a21454-a2ca-11eb-bbaa-93a81c686f21",
-			expectedErr: errErasureReadQuorum,
+			fis:            getNFInfo(16, 7, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21"),
+			modTime:        time.Unix(1603863445, 0),
+			dataDir:        "36a21454-a2ca-11eb-bbaa-93a81c686f21",
+			expectedErr:    errErasureReadQuorum,
+			expectedQuorum: 8,
+		},
+		{
+			fis:            getNFInfo(16, 16, 1603863445, "36a21454-a2ca-11eb-bbaa-93a81c686f21"),
+			modTime:        time.Unix(1603863445, 0),
+			dataDir:        "36a21454-a2ca-11eb-bbaa-93a81c686f21",
+			expectedErr:    errErasureReadQuorum,
+			expectedQuorum: 0,
 		},
 	}
 
 	for _, test := range tests {
 		test := test
 		t.Run("", func(t *testing.T) {
-			_, err := findFileInfoInQuorum(context.Background(), test.fis, test.modTime, test.dataDir, 8)
+			_, err := findFileInfoInQuorum(context.Background(), test.fis, test.modTime, test.dataDir, test.expectedQuorum)
 			if err != test.expectedErr {
 				t.Errorf("Expected %s, got %s", test.expectedErr, err)
 			}

cmd/erasure-object.go

@@ -214,37 +214,6 @@ func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object stri
 	return fn(pr, h, opts.CheckPrecondFn, pipeCloser, nsUnlocker)
 }
 
-// GetObject - reads an object erasured coded across multiple
-// disks. Supports additional parameters like offset and length
-// which are synonymous with HTTP Range requests.
-//
-// startOffset indicates the starting read location of the object.
-// length indicates the total length of the object.
-func (er erasureObjects) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) {
-	// Lock the object before reading.
-	lk := er.NewNSLock(bucket, object)
-	lkctx, err := lk.GetRLock(ctx, globalOperationTimeout)
-	if err != nil {
-		return err
-	}
-	ctx = lkctx.Context()
-	defer lk.RUnlock(lkctx.Cancel)
-
-	// Start offset cannot be negative.
-	if startOffset < 0 {
-		logger.LogIf(ctx, errUnexpected, logger.Application)
-		return errUnexpected
-	}
-
-	// Writer cannot be nil.
-	if writer == nil {
-		logger.LogIf(ctx, errUnexpected)
-		return errUnexpected
-	}
-
-	return er.getObject(ctx, bucket, object, startOffset, length, writer, opts)
-}
-
 func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, fi FileInfo, metaArr []FileInfo, onlineDisks []StorageAPI) error {
 	// Reorder online disks based on erasure distribution order.
 	// Reorder parts metadata based on erasure distribution order.
@@ -284,6 +253,27 @@ func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, obje
 	if err != nil {
 		return toObjectErr(err, bucket, object)
 	}
+
+	// This hack is needed to avoid a bug found when overwriting
+	// the inline data object with a un-inlined version, for the
+	// time being we need this as we have released inline-data
+	// version already and this bug is already present in newer
+	// releases.
+	//
+	// This mainly happens with objects < smallFileThreshold when
+	// they are overwritten with un-inlined objects >= smallFileThreshold,
+	// due to a bug in RenameData() the fi.Data is not niled leading to
+	// GetObject thinking that fi.Data is valid while fi.Size has
+	// changed already.
+	if _, ok := fi.Metadata[ReservedMetadataPrefixLower+"inline-data"]; ok {
+		shardFileSize := erasure.ShardFileSize(fi.Size)
+		if shardFileSize >= 0 && shardFileSize >= smallFileThreshold {
+			for i := range metaArr {
+				metaArr[i].Data = nil
+			}
+		}
+	}
+
 	var healOnce sync.Once
 
 	// once we have obtained a common FileInfo i.e latest, we should stick
@@ -373,23 +363,6 @@ func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, obje
 	return nil
 }
 
-// getObject wrapper for erasure GetObject
-func (er erasureObjects) getObject(ctx context.Context, bucket, object string, startOffset, length int64, writer io.Writer, opts ObjectOptions) error {
-	fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, bucket, object, opts, true)
-	if err != nil {
-		return toObjectErr(err, bucket, object)
-	}
-	if fi.Deleted {
-		if opts.VersionID == "" {
-			return toObjectErr(errFileNotFound, bucket, object)
-		}
-		// Make sure to return object info to provide extra information.
-		return toObjectErr(errMethodNotAllowed, bucket, object)
-	}
-
-	return er.getObjectWithFileInfo(ctx, bucket, object, startOffset, length, writer, fi, metaArr, onlineDisks)
-}
-
 // GetObjectInfo - reads object metadata and replies back ObjectInfo.
 func (er erasureObjects) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (info ObjectInfo, err error) {
 	if !opts.NoLock {
@@ -797,6 +770,13 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		partsMetadata[index].ModTime = modTime
 	}
 
+	if len(inlineBuffers) > 0 {
+		// Set an additional header when data is inlined.
+		for index := range partsMetadata {
+			partsMetadata[index].Metadata[ReservedMetadataPrefixLower+"inline-data"] = "true"
+		}
+	}
+
 	// Rename the successfully written temporary object to final location.
 	if onlineDisks, err = renameData(ctx, onlineDisks, minioMetaTmpBucket, tempObj, partsMetadata, bucket, object, writeQuorum); err != nil {
 		logger.LogIf(ctx, err)

cmd/erasure-object_test.go

@@ -323,9 +323,18 @@ func TestGetObjectNoQuorum(t *testing.T) {
 		}
 	}
 
-	err = xl.GetObject(ctx, bucket, object, 0, int64(len(buf)), ioutil.Discard, "", opts)
-	if err != toObjectErr(errErasureReadQuorum, bucket, object) {
-		t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureReadQuorum, bucket, object), err)
+	gr, err := xl.GetObjectNInfo(ctx, bucket, object, nil, nil, readLock, opts)
+	if err != nil {
+		if err != toObjectErr(errErasureReadQuorum, bucket, object) {
+			t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureReadQuorum, bucket, object), err)
+		}
+	}
+	if gr != nil {
+		_, err = io.Copy(ioutil.Discard, gr)
+		if err != toObjectErr(errErasureReadQuorum, bucket, object) {
+			t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureReadQuorum, bucket, object), err)
+		}
+		gr.Close()
 	}
 
 	// Test use case 2: Make 9 disks offline, which leaves less than quorum number of disks
@@ -359,9 +368,18 @@ func TestGetObjectNoQuorum(t *testing.T) {
 		}
 		z.serverPools[0].erasureDisksMu.Unlock()
 		// Fetch object from store.
-		err = xl.GetObject(ctx, bucket, object, 0, int64(len("abcd")), ioutil.Discard, "", opts)
-		if err != toObjectErr(errErasureReadQuorum, bucket, object) {
-			t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureWriteQuorum, bucket, object), err)
+		gr, err := xl.GetObjectNInfo(ctx, bucket, object, nil, nil, readLock, opts)
+		if err != nil {
+			if err != toObjectErr(errErasureReadQuorum, bucket, object) {
+				t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureReadQuorum, bucket, object), err)
+			}
+		}
+		if gr != nil {
+			_, err = io.Copy(ioutil.Discard, gr)
+			if err != toObjectErr(errErasureReadQuorum, bucket, object) {
+				t.Errorf("Expected GetObject to fail with %v, but failed with %v", toObjectErr(errErasureReadQuorum, bucket, object), err)
+			}
+			gr.Close()
 		}
 	}
 

cmd/erasure-server-pool.go

@@ -622,7 +622,7 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
 	}
 	wg.Wait()
 
-	var found int = -1
+	var found = -1
 	for i, err := range errs {
 		if err == nil {
 			found = i
@@ -683,7 +683,7 @@ func (z *erasureServerPools) GetObjectInfo(ctx context.Context, bucket, object s
 	}
 	wg.Wait()
 
-	var found int = -1
+	var found = -1
 	for i, err := range errs {
 		if err == nil {
 			found = i

cmd/format-disk-cache.go

@@ -495,8 +495,5 @@ func migrateCacheFormatJSON(cacheFormatPath string) error {
 	formatV2.Version = formatMetaVersion1
 	formatV2.Cache = formatV1.Cache
 	formatV2.Cache.Version = formatCacheVersionV2
-	if err := jsonSave(f, formatV2); err != nil {
-		return err
-	}
-	return nil
+	return jsonSave(f, formatV2)
 }

cmd/fs-v1.go

@@ -801,82 +801,6 @@ func (fs *FSObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
 	return objReaderFn(reader, h, opts.CheckPrecondFn, closeFn, rwPoolUnlocker, nsUnlocker)
 }
 
-// getObject - wrapper for GetObject
-func (fs *FSObjects) getObject(ctx context.Context, bucket, object string, offset int64, length int64, writer io.Writer, etag string, lock bool) (err error) {
-	if _, err = fs.statBucketDir(ctx, bucket); err != nil {
-		return toObjectErr(err, bucket)
-	}
-
-	// Offset cannot be negative.
-	if offset < 0 {
-		logger.LogIf(ctx, errUnexpected, logger.Application)
-		return toObjectErr(errUnexpected, bucket, object)
-	}
-
-	// Writer cannot be nil.
-	if writer == nil {
-		logger.LogIf(ctx, errUnexpected, logger.Application)
-		return toObjectErr(errUnexpected, bucket, object)
-	}
-
-	// If its a directory request, we return an empty body.
-	if HasSuffix(object, SlashSeparator) {
-		_, err = writer.Write([]byte(""))
-		logger.LogIf(ctx, err)
-		return toObjectErr(err, bucket, object)
-	}
-
-	if bucket != minioMetaBucket {
-		fsMetaPath := pathJoin(fs.fsPath, minioMetaBucket, bucketMetaPrefix, bucket, object, fs.metaJSONFile)
-		if lock {
-			_, err = fs.rwPool.Open(fsMetaPath)
-			if err != nil && err != errFileNotFound {
-				logger.LogIf(ctx, err)
-				return toObjectErr(err, bucket, object)
-			}
-			defer fs.rwPool.Close(fsMetaPath)
-		}
-	}
-
-	if etag != "" && etag != defaultEtag {
-		objEtag, perr := fs.getObjectETag(ctx, bucket, object, lock)
-		if perr != nil {
-			return toObjectErr(perr, bucket, object)
-		}
-		if objEtag != etag {
-			logger.LogIf(ctx, InvalidETag{}, logger.Application)
-			return toObjectErr(InvalidETag{}, bucket, object)
-		}
-	}
-
-	// Read the object, doesn't exist returns an s3 compatible error.
-	fsObjPath := pathJoin(fs.fsPath, bucket, object)
-	reader, size, err := fsOpenFile(ctx, fsObjPath, offset)
-	if err != nil {
-		return toObjectErr(err, bucket, object)
-	}
-	defer reader.Close()
-
-	// For negative length we read everything.
-	if length < 0 {
-		length = size - offset
-	}
-
-	// Reply back invalid range if the input offset and length fall out of range.
-	if offset > size || offset+length > size {
-		err = InvalidRange{offset, length, size}
-		logger.LogIf(ctx, err, logger.Application)
-		return err
-	}
-
-	_, err = io.Copy(writer, io.LimitReader(reader, length))
-	// The writer will be closed incase of range queries, which will emit ErrClosedPipe.
-	if err == io.ErrClosedPipe {
-		err = nil
-	}
-	return toObjectErr(err, bucket, object)
-}
-
 // Create a new fs.json file, if the existing one is corrupt. Should happen very rarely.
 func (fs *FSObjects) createFsJSON(object, fsMetaPath string) error {
 	fsMeta := newFSMetaV1()
@@ -1377,77 +1301,6 @@ func (fs *FSObjects) isObjectDir(bucket, prefix string) bool {
 	return len(entries) == 0
 }
 
-// getObjectETag is a helper function, which returns only the md5sum
-// of the file on the disk.
-func (fs *FSObjects) getObjectETag(ctx context.Context, bucket, entry string, lock bool) (string, error) {
-	fsMetaPath := pathJoin(fs.fsPath, minioMetaBucket, bucketMetaPrefix, bucket, entry, fs.metaJSONFile)
-
-	var reader io.Reader
-	var fi os.FileInfo
-	var size int64
-	if lock {
-		// Read `fs.json` to perhaps contend with
-		// parallel Put() operations.
-		rlk, err := fs.rwPool.Open(fsMetaPath)
-		// Ignore if `fs.json` is not available, this is true for pre-existing data.
-		if err != nil && err != errFileNotFound {
-			logger.LogIf(ctx, err)
-			return "", toObjectErr(err, bucket, entry)
-		}
-
-		// If file is not found, we don't need to proceed forward.
-		if err == errFileNotFound {
-			return "", nil
-		}
-
-		// Read from fs metadata only if it exists.
-		defer fs.rwPool.Close(fsMetaPath)
-
-		// Fetch the size of the underlying file.
-		fi, err = rlk.LockedFile.Stat()
-		if err != nil {
-			logger.LogIf(ctx, err)
-			return "", toObjectErr(err, bucket, entry)
-		}
-
-		size = fi.Size()
-		reader = io.NewSectionReader(rlk.LockedFile, 0, fi.Size())
-	} else {
-		var err error
-		reader, size, err = fsOpenFile(ctx, fsMetaPath, 0)
-		if err != nil {
-			return "", toObjectErr(err, bucket, entry)
-		}
-	}
-
-	// `fs.json` can be empty due to previously failed
-	// PutObject() transaction, if we arrive at such
-	// a situation we just ignore and continue.
-	if size == 0 {
-		return "", nil
-	}
-
-	fsMetaBuf, err := ioutil.ReadAll(reader)
-	if err != nil {
-		logger.LogIf(ctx, err)
-		return "", toObjectErr(err, bucket, entry)
-	}
-
-	var fsMeta fsMetaV1
-	var json = jsoniter.ConfigCompatibleWithStandardLibrary
-	if err = json.Unmarshal(fsMetaBuf, &fsMeta); err != nil {
-		return "", err
-	}
-
-	// Check if FS metadata is valid, if not return error.
-	if !isFSMetaValid(fsMeta.Version) {
-		logger.LogIf(ctx, errCorruptedFormat)
-		return "", toObjectErr(errCorruptedFormat, bucket, entry)
-	}
-
-	return extractETag(fsMeta.Meta), nil
-}
-
 // ListObjectVersions not implemented for FS mode.
 func (fs *FSObjects) ListObjectVersions(ctx context.Context, bucket, prefix, marker, versionMarker, delimiter string, maxKeys int) (loi ListObjectVersionsInfo, e error) {
 	return loi, NotImplemented{}

cmd/globals.go

@@ -215,9 +215,6 @@ var (
 	// Hold the old server credentials passed by the environment
 	globalOldCred auth.Credentials
 
-	// Indicates if config is to be encrypted
-	globalConfigEncrypted bool
-
 	globalPublicCerts []*x509.Certificate
 
 	globalDomainNames []string      // Root domains for virtual host style requests

cmd/http-tracer.go

@@ -218,7 +218,7 @@ func Trace(f http.HandlerFunc, logBody bool, w http.ResponseWriter, r *http.Requ
 		Time:     now,
 		Proto:    r.Proto,
 		Method:   r.Method,
-		Path:     r.URL.Path,
+		Path:     r.URL.RawPath,
 		RawQuery: redactLDAPPwd(r.URL.RawQuery),
 		Client:   handlers.GetSourceIP(r),
 		Headers:  reqHeaders,

cmd/iam-etcd-store.go

@@ -57,7 +57,7 @@ func etcdKvsToSet(prefix string, kvs []*mvccpb.KeyValue) set.StringSet {
 //  suffix := "config.json"
 //  result is foo
 func extractPathPrefixAndSuffix(s string, prefix string, suffix string) string {
-	return pathClean(strings.TrimSuffix(strings.TrimPrefix(string(s), prefix), suffix))
+	return pathClean(strings.TrimSuffix(strings.TrimPrefix(s, prefix), suffix))
 }
 
 // IAMEtcdStore implements IAMStorageAPI
@@ -331,7 +331,7 @@ func (ies *IAMEtcdStore) addUser(ctx context.Context, user string, userType IAMU
 				return []byte(globalOldCred.SecretKey), nil
 			}
 			if _, err := jwtgo.ParseWithClaims(u.Credentials.SessionToken, m, stsTokenCallback); err == nil {
-				jwt := jwtgo.NewWithClaims(jwtgo.SigningMethodHS512, jwtgo.MapClaims(m))
+				jwt := jwtgo.NewWithClaims(jwtgo.SigningMethodHS512, m)
 				if token, err := jwt.SignedString([]byte(globalActiveCred.SecretKey)); err == nil {
 					u.Credentials.SessionToken = token
 					err := ies.saveIAMConfig(ctx, &u, getUserIdentityPath(user, userType))

cmd/metacache-bucket.go

@@ -386,23 +386,6 @@ func interestingCaches(root string, cachesRoot map[string][]string) []string {
 	return interesting
 }
 
-// updateCache will update a cache by id.
-// If the cache cannot be found nil is returned.
-// The bucket cache will be locked until the done .
-func (b *bucketMetacache) updateCache(id string) (cache *metacache, done func()) {
-	b.mu.Lock()
-	c, ok := b.caches[id]
-	if !ok {
-		b.mu.Unlock()
-		return nil, func() {}
-	}
-	return &c, func() {
-		c.lastUpdate = UTCNow()
-		b.caches[id] = c
-		b.mu.Unlock()
-	}
-}
-
 // updateCacheEntry will update a cache.
 // Returns the updated status.
 func (b *bucketMetacache) updateCacheEntry(update metacache) (metacache, error) {
@@ -437,18 +420,6 @@ func (b *bucketMetacache) cloneCaches() (map[string]metacache, map[string][]stri
 	return dst, dst2
 }
 
-// getCache will return a clone of a specific metacache.
-// Will return nil if the cache doesn't exist.
-func (b *bucketMetacache) getCache(id string) *metacache {
-	b.mu.RLock()
-	c, ok := b.caches[id]
-	b.mu.RUnlock()
-	if !ok {
-		return nil
-	}
-	return &c
-}
-
 // deleteAll will delete all on disk data for ALL caches.
 // Deletes are performed concurrently.
 func (b *bucketMetacache) deleteAll() {

cmd/metacache-entries.go

@@ -19,7 +19,6 @@ package cmd
 
 import (
 	"bytes"
-	"io"
 	"os"
 	"sort"
 	"strings"
@@ -275,16 +274,6 @@ type metaCacheEntriesSorted struct {
 	listID string
 }
 
-// writeTo will write all objects to the provided output.
-func (m metaCacheEntriesSorted) writeTo(writer io.Writer) error {
-	w := newMetacacheWriter(writer, 1<<20)
-	if err := w.write(m.o...); err != nil {
-		w.Close()
-		return err
-	}
-	return w.Close()
-}
-
 // shallowClone will create a shallow clone of the array objects,
 // but object metadata will not be cloned.
 func (m metaCacheEntriesSorted) shallowClone() metaCacheEntriesSorted {
@@ -293,19 +282,6 @@ func (m metaCacheEntriesSorted) shallowClone() metaCacheEntriesSorted {
 	return m
 }
 
-// iterate the entries in order.
-// If the iterator function returns iterating stops.
-func (m *metaCacheEntriesSorted) iterate(fn func(entry metaCacheEntry) (cont bool)) {
-	if m == nil {
-		return
-	}
-	for _, o := range m.o {
-		if !fn(o) {
-			return
-		}
-	}
-}
-
 // fileInfoVersions converts the metadata to FileInfoVersions where possible.
 // Metadata that cannot be decoded is skipped.
 func (m *metaCacheEntriesSorted) fileInfoVersions(bucket, prefix, delimiter, afterV string) (versions []ObjectInfo) {
@@ -488,17 +464,6 @@ func (m *metaCacheEntriesSorted) merge(other metaCacheEntriesSorted, limit int)
 	m.o = merged
 }
 
-// filter allows selective filtering with the provided function.
-func (m *metaCacheEntriesSorted) filter(fn func(entry *metaCacheEntry) bool) {
-	dst := m.o[:0]
-	for _, o := range m.o {
-		if fn(&o) {
-			dst = append(dst, o)
-		}
-	}
-	m.o = dst
-}
-
 // filterPrefix will filter m to only contain entries with the specified prefix.
 func (m *metaCacheEntriesSorted) filterPrefix(s string) {
 	if s == "" {

cmd/metrics-v2.go

@@ -952,7 +952,7 @@ func getMinioProcMetrics() MetricsGroup {
 			metrics = append(metrics,
 				Metric{
 					Description: getMinIOProcessCPUTime(),
-					Value:       float64(stat.CPUTime()),
+					Value:       stat.CPUTime(),
 				})
 			return
 		},

cmd/metrics.go

@@ -181,7 +181,7 @@ func healingMetricsPrometheus(ch chan<- prometheus.Metric) {
 				"Objects for which healing failed in current self healing run",
 				[]string{"mount_path", "volume_status"}, nil),
 			prometheus.GaugeValue,
-			float64(v), string(s[0]), string(s[1]),
+			float64(v), s[0], s[1],
 		)
 	}
 }

cmd/notification.go

@@ -498,75 +498,6 @@ func (sys *NotificationSys) updateBloomFilter(ctx context.Context, current uint6
 	return bf, nil
 }
 
-// collectBloomFilter will collect bloom filters from all servers from the specified cycle.
-func (sys *NotificationSys) collectBloomFilter(ctx context.Context, from uint64) (*bloomFilter, error) {
-	var req = bloomFilterRequest{
-		Current: 0,
-		Oldest:  from,
-	}
-
-	// Load initial state from local...
-	var bf *bloomFilter
-	bfr, err := intDataUpdateTracker.cycleFilter(ctx, req)
-	logger.LogIf(ctx, err)
-	if err == nil && bfr.Complete {
-		nbf := intDataUpdateTracker.newBloomFilter()
-		bf = &nbf
-		_, err = bf.ReadFrom(bytes.NewReader(bfr.Filter))
-		logger.LogIf(ctx, err)
-	}
-	if !bfr.Complete {
-		// If local isn't complete just return early
-		return nil, nil
-	}
-
-	var mu sync.Mutex
-	g := errgroup.WithNErrs(len(sys.peerClients))
-	for idx, client := range sys.peerClients {
-		if client == nil {
-			continue
-		}
-		client := client
-		g.Go(func() error {
-			serverBF, err := client.cycleServerBloomFilter(ctx, req)
-			if false && intDataUpdateTracker.debug {
-				b, _ := json.MarshalIndent(serverBF, "", "  ")
-				logger.Info("Disk %v, Bloom filter: %v", client.host.Name, string(b))
-			}
-			// Keep lock while checking result.
-			mu.Lock()
-			defer mu.Unlock()
-
-			if err != nil || !serverBF.Complete || bf == nil {
-				logger.LogIf(ctx, err)
-				bf = nil
-				return nil
-			}
-
-			var tmp bloom.BloomFilter
-			_, err = tmp.ReadFrom(bytes.NewReader(serverBF.Filter))
-			if err != nil {
-				logger.LogIf(ctx, err)
-				bf = nil
-				return nil
-			}
-			if bf.BloomFilter == nil {
-				bf.BloomFilter = &tmp
-			} else {
-				err = bf.Merge(&tmp)
-				if err != nil {
-					logger.LogIf(ctx, err)
-					bf = nil
-					return nil
-				}
-			}
-			return nil
-		}, idx)
-	}
-	g.Wait()
-	return bf, nil
-}
-
 // findEarliestCleanBloomFilter will find the earliest bloom filter across the cluster
 // where the directory is clean.
 // Due to how objects are stored this can include object names.

cmd/object-api-putobject_test.go

@@ -33,7 +33,7 @@ import (
 )
 
 func md5Header(data []byte) map[string]string {
-	return map[string]string{"etag": getMD5Hash([]byte(data))}
+	return map[string]string{"etag": getMD5Hash(data)}
 }
 
 // Wrapper for calling PutObject tests for both Erasure multiple disks and single node setup.

cmd/storage-rest-common.go

@@ -18,7 +18,7 @@
 package cmd
 
 const (
-	storageRESTVersion       = "v34" // Streaming Usage Updates
+	storageRESTVersion       = "v35" // Inline bugfix needs all servers to be updated
 	storageRESTVersionPrefix = SlashSeparator + storageRESTVersion
 	storageRESTPrefix        = minioReservedBucketPath + "/storage"
 )

cmd/sts-handlers.go

@@ -119,12 +119,12 @@ func checkAssumeRoleAuth(ctx context.Context, r *http.Request) (user auth.Creden
 		return user, true, ErrSTSAccessDenied
 	case authTypeSigned:
 		s3Err := isReqAuthenticated(ctx, r, globalServerRegion, serviceSTS)
-		if APIErrorCode(s3Err) != ErrNone {
+		if s3Err != ErrNone {
 			return user, false, STSErrorCode(s3Err)
 		}
 
 		user, _, s3Err = getReqAccessKeyV4(r, globalServerRegion, serviceSTS)
-		if APIErrorCode(s3Err) != ErrNone {
+		if s3Err != ErrNone {
 			return user, false, STSErrorCode(s3Err)
 		}
 

cmd/test-utils_test.go

@@ -78,8 +78,6 @@ func TestMain(m *testing.M) {
 		SecretKey: auth.DefaultSecretKey,
 	}
 
-	globalConfigEncrypted = true
-
 	// disable ENVs which interfere with tests.
 	for _, env := range []string{
 		crypto.EnvKMSAutoEncryption,
@@ -1278,35 +1276,6 @@ func getRandomBucketName() string {
 
 }
 
-// NewEOFWriter returns a Writer that writes to w,
-// but returns EOF error after writing n bytes.
-func NewEOFWriter(w io.Writer, n int64) io.Writer {
-	return &EOFWriter{w, n}
-}
-
-type EOFWriter struct {
-	w io.Writer
-	n int64
-}
-
-// io.Writer implementation designed to error out with io.EOF after reading `n` bytes.
-func (t *EOFWriter) Write(p []byte) (n int, err error) {
-	if t.n <= 0 {
-		return -1, io.EOF
-	}
-	// real write
-	n = len(p)
-	if int64(n) > t.n {
-		n = int(t.n)
-	}
-	n, err = t.w.Write(p[0:n])
-	t.n -= int64(n)
-	if err == nil {
-		n = len(p)
-	}
-	return
-}
-
 // construct URL for http requests for bucket operations.
 func makeTestTargetURL(endPoint, bucketName, objectName string, queryValues url.Values) string {
 	urlStr := endPoint + SlashSeparator

cmd/tier.go

@@ -49,7 +49,7 @@ const (
 )
 
 // tierConfigPath refers to remote tier config object name
-var tierConfigPath string = path.Join(minioConfigPrefix, tierConfigFile)
+var tierConfigPath = path.Join(minioConfigPrefix, tierConfigFile)
 
 // TierConfigMgr holds the collection of remote tiers configured in this deployment.
 type TierConfigMgr struct {

cmd/xl-storage-format-v2.go

@@ -211,6 +211,9 @@ type xlMetaV2Version struct {
 
 // Valid xl meta xlMetaV2Version is valid
 func (j xlMetaV2Version) Valid() bool {
+	if !j.Type.valid() {
+		return false
+	}
 	switch j.Type {
 	case LegacyType:
 		return j.ObjectV1 != nil &&

cmd/xl-storage.go

@@ -758,25 +758,6 @@ func (s *xlStorage) DeleteVol(ctx context.Context, volume string, forceDelete bo
 	return nil
 }
 
-func (s *xlStorage) isLeaf(volume string, leafPath string) bool {
-	volumeDir, err := s.getVolDir(volume)
-	if err != nil {
-		return false
-	}
-
-	if err = Access(pathJoin(volumeDir, leafPath, xlStorageFormatFile)); err == nil {
-		return true
-	}
-	if osIsNotExist(err) {
-		// We need a fallback code where directory might contain
-		// legacy `xl.json`, in such situation we just rename
-		// and proceed if rename is successful we know that it
-		// is the leaf since `xl.json` was present.
-		return s.renameLegacyMetadata(volumeDir, leafPath) == nil
-	}
-	return false
-}
-
 // ListDir - return all the entries at the given directory path.
 // If an entry is a directory it will be returned with a trailing SlashSeparator.
 func (s *xlStorage) ListDir(ctx context.Context, volume, dirPath string, count int) (entries []string, err error) {
@@ -1095,6 +1076,11 @@ func (s *xlStorage) ReadVersion(ctx context.Context, volume, path, versionID str
 
 	if readData {
 		if len(fi.Data) > 0 || fi.Size == 0 {
+			if len(fi.Data) > 0 {
+				if _, ok := fi.Metadata[ReservedMetadataPrefixLower+"inline-data"]; !ok {
+					fi.Metadata[ReservedMetadataPrefixLower+"inline-data"] = "true"
+				}
+			}
 			return fi, nil
 		}
 
@@ -2042,6 +2028,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 				// Purge the destination path as we are not preserving anything
 				// versioned object was not requested.
 				oldDstDataPath = pathJoin(dstVolumeDir, dstPath, ofi.DataDir)
+				xlMeta.data.remove(nullVersionID, ofi.DataDir)
 			}
 		}
 	}

docs/bucket/replication/DESIGN.md

@@ -37,7 +37,7 @@ Existing object replication, replica modification sync for 2-way replication and
 
 ### Internal metadata for replication
 
-`xl.meta` that is in use for [versioning](https://raw.githubusercontent.com/minio/minio/master/docs/bucket/versioning/DESIGN.md) has additional metadata for replication of objects,delete markers and versioned deletes.
+`xl.meta` that is in use for [versioning](https://github.com/minio/minio/blob/master/docs/bucket/versioning/DESIGN.md) has additional metadata for replication of objects,delete markers and versioned deletes.
 
 ### Metadata for object replication
 

docs/bucket/versioning/README.md

@@ -2,7 +2,7 @@
 
 MinIO versioning is designed to keep multiple versions of an object in one bucket. For example, you could store `spark.csv` (version `ede336f2`) and `spark.csv` (version `fae684da`) in a single bucket. Versioning protects you from unintended overwrites, deletions, protect objects with retention policies.
 
-To control data retention and storage usage, use object versioning with [object lifecycle management](https://github.com/minio/minio/master/docs/bucket/lifecycle/README.md).  If you have an object expiration lifecycle policy in your non-versioned bucket and you want to maintain the same permanent delete behavior when on versioning-enabled bucket, you must add a noncurrent expiration policy. The noncurrent expiration lifecycle policy will manage the deletes of the noncurrent object versions in the versioning-enabled bucket. (A version-enabled bucket maintains one current and zero or more noncurrent object versions.)
+To control data retention and storage usage, use object versioning with [object lifecycle management](https://github.com/minio/minio/blob/master/docs/bucket/lifecycle/README.md).  If you have an object expiration lifecycle policy in your non-versioned bucket and you want to maintain the same permanent delete behavior when on versioning-enabled bucket, you must add a noncurrent expiration policy. The noncurrent expiration lifecycle policy will manage the deletes of the noncurrent object versions in the versioning-enabled bucket. (A version-enabled bucket maintains one current and zero or more noncurrent object versions.)
 
 Versioning must be explicitly enabled on a bucket, versioning is not enabled by default. Object locking enabled buckets have versioning enabled automatically. Enabling and suspending versioning is done at the bucket level.
 

docs/minio-limits.md

@@ -2,41 +2,35 @@
 
 ### Erasure Code (Multiple Drives / Servers)
 
-|Item|Specification|
-|:---|:---|
-|Maximum number of servers per cluster| no-limit|
-|Maximum number of federated clusters | no-limit|
-|Minimum number of servers| 02|
-|Minimum number of drives per server when server count is 1 | 04 |
-|Minimum number of drives per server when server count is 2 or 3 | 02|
-|Minimum number of drives per server when server count is 4 | 01|
-|Maximum number of drives per server| no-limit|
-|Read quorum| N/2|
-|Write quorum| N/2+1|
-
-### Browser Access
-
-|Item|Specification|
-|:---|:---|
-|Web browser upload size limit| 5 TiB|
+| Item                                                            | Specification |
+|:----------------------------------------------------------------|:--------------|
+| Maximum number of servers per cluster                           | no-limit      |
+| Maximum number of federated clusters                            | no-limit      |
+| Minimum number of servers                                       | 02            |
+| Minimum number of drives per server when server count is 1      | 04            |
+| Minimum number of drives per server when server count is 2 or 3 | 02            |
+| Minimum number of drives per server when server count is 4      | 01            |
+| Maximum number of drives per server                             | no-limit      |
+| Read quorum                                                     | N/2           |
+| Write quorum                                                    | N/2+1         |
 
 ### Limits of S3 API
 
-|Item|Specification|
-|:---|:---|
-|Maximum number of buckets| no-limit|
-|Maximum number of objects per bucket| no-limit|
-|Maximum object size| 5 TiB|
-|Minimum object size| 0 B|
-|Maximum object size per PUT operation| 5 TiB|
-|Maximum number of parts per upload| 	10,000|
-|Part size|5 MiB to 5 GiB. Last part can be 0 B to 5 GiB|
-|Maximum number of parts returned per list parts request| 10000|
-|Maximum number of objects returned per list objects request| 10000|
-|Maximum number of multipart uploads returned per list multipart uploads request| 1000|
+| Item                                                                            | Specification                                 |
+|:--------------------------------------------------------------------------------|:----------------------------------------------|
+| Maximum number of buckets                                                       | no-limit                                      |
+| Maximum number of objects per bucket                                            | no-limit                                      |
+| Maximum object size                                                             | 5 TiB                                         |
+| Minimum object size                                                             | 0 B                                           |
+| Maximum object size per PUT operation                                           | 5 TiB                                         |
+| Maximum number of parts per upload                                              | 10,000                                        |
+| Part size range                                                                 | 5 MiB to 5 GiB. Last part can be 0 B to 5 GiB |
+| Maximum number of parts returned per list parts request                         | 10000                                         |
+| Maximum number of objects returned per list objects request                     | 4500                                          |
+| Maximum number of multipart uploads returned per list multipart uploads request | 1000                                          |
 
 ### List of Amazon S3 API's not supported on MinIO
-We found the following APIs to be redundant or less useful outside of AWS S3. If you have a different view on any of the APIs we missed, please open a [github issue](https://github.com/minio/minio/issues).
+We found the following APIs to be redundant or less useful outside of AWS S3. If you have a different view on any of the APIs we missed, please open a [GitHub issue](https://github.com/minio/minio/issues).
 
 #### List of Amazon S3 Bucket API's not supported on MinIO
 
@@ -52,4 +46,6 @@ We found the following APIs to be redundant or less useful outside of AWS S3. If
 - ObjectTorrent
 
 ### Object name restrictions on MinIO
-Object names that contain characters `^*|\/&";` are unsupported on Windows and other file systems which do not support filenames with these characters. Note that this list is not exhaustive, and depends on the maintainers of the filesystem itself.
+Object names that contain characters `^*|\/&";` are unsupported on Windows and other file systems which do not support filenames with these characters. NOTE: This list is not an exhaustive, it depends on the operating system and filesystem under use.
+
+For best experience we recommend that you use distributions that ship fairly recent Linux kernel such as CentOS 8, Ubuntu 18.04 LTS along with XFS as the choice for your backend filesystem.

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

@@ -5,7 +5,7 @@ version: '3.7'
 # it through port 9000.
 services:
   minio1:
-    image: minio/minio:RELEASE.2021-05-20T22-31-44Z
+    image: minio/minio:RELEASE.2021-05-22T02-34-39Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2021-05-20T22-31-44Z
+    image: minio/minio:RELEASE.2021-05-22T02-34-39Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2021-05-20T22-31-44Z
+    image: minio/minio:RELEASE.2021-05-22T02-34-39Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2021-05-20T22-31-44Z
+    image: minio/minio:RELEASE.2021-05-22T02-34-39Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

pkg/bucket/lifecycle/lifecycle_test.go

@@ -135,12 +135,12 @@ func TestMarshalLifecycleConfig(t *testing.T) {
 			{
 				Status:     "Enabled",
 				Filter:     Filter{Prefix: Prefix{string: "prefix-1", set: true}},
-				Expiration: Expiration{Date: ExpirationDate(midnightTS)},
+				Expiration: Expiration{Date: midnightTS},
 			},
 			{
 				Status:                      "Enabled",
 				Filter:                      Filter{Prefix: Prefix{string: "prefix-1", set: true}},
-				Expiration:                  Expiration{Date: ExpirationDate(midnightTS)},
+				Expiration:                  Expiration{Date: midnightTS},
 				NoncurrentVersionTransition: NoncurrentVersionTransition{NoncurrentDays: 2, StorageClass: "TEST"},
 			},
 		},

pkg/bucket/lifecycle/rule.go

@@ -65,7 +65,7 @@ func getNewUUID() (string, error) {
 
 // validateID - checks if ID is valid or not.
 func (r Rule) validateID() error {
-	IDLen := len(string(r.ID))
+	IDLen := len(r.ID)
 	// generate new ID when not provided
 	// cannot be longer than 255 characters
 	if IDLen == 0 {

pkg/bucket/replication/rule.go

@@ -140,10 +140,7 @@ func (r Rule) validateStatus() error {
 }
 
 func (r Rule) validateFilter() error {
-	if err := r.Filter.Validate(); err != nil {
-		return err
-	}
-	return nil
+	return r.Filter.Validate()
 }
 
 // Prefix - a rule can either have prefix under <filter></filter> or under

pkg/cgroup/linux.go

@@ -35,7 +35,7 @@ import (
 )
 
 // DO NOT EDIT following constants are chosen defaults for any kernel
-// after 3.x, please open a github issue https://github.com/minio/minio/issues
+// after 3.x, please open a GitHub issue https://github.com/minio/minio/issues
 // and discuss first if you wish to change this.
 const (
 	// Default string for looking for kernel memory param.

pkg/disk/stat_linux.go

@@ -33,10 +33,11 @@ func GetInfo(path string) (info Info, err error) {
 	}
 	reservedBlocks := s.Bfree - s.Bavail
 	info = Info{
-		Total:  uint64(s.Frsize) * (s.Blocks - reservedBlocks),
-		Free:   uint64(s.Frsize) * s.Bavail,
-		Files:  s.Files,
-		Ffree:  s.Ffree,
+		Total: uint64(s.Frsize) * (s.Blocks - reservedBlocks),
+		Free:  uint64(s.Frsize) * s.Bavail,
+		Files: s.Files,
+		Ffree: s.Ffree,
+		//nolint:unconvert
 		FSType: getFSType(int64(s.Type)),
 	}
 	// Check for overflows.

pkg/event/target/amqp.go

@@ -205,16 +205,12 @@ func (target *AMQPTarget) send(eventData event.Event, ch *amqp.Channel) error {
 		return err
 	}
 
-	if err := ch.Publish(target.args.Exchange, target.args.RoutingKey, target.args.Mandatory,
+	return ch.Publish(target.args.Exchange, target.args.RoutingKey, target.args.Mandatory,
 		target.args.Immediate, amqp.Publishing{
 			ContentType:  "application/json",
 			DeliveryMode: target.args.DeliveryMode,
 			Body:         data,
-		}); err != nil {
-		return err
-	}
-
-	return nil
+		})
 }
 
 // Save - saves the events to the store which will be replayed when the amqp connection is active.

pkg/s3select/csv/record.go

@@ -103,11 +103,7 @@ func (r *Record) WriteCSV(writer io.Writer, opts sql.WriteCSVOpts) error {
 		return err
 	}
 	w.Flush()
-	if err := w.Error(); err != nil {
-		return err
-	}
-
-	return nil
+	return w.Error()
 }
 
 // WriteJSON - encodes to JSON data.

pkg/s3select/json/record.go

@@ -145,11 +145,7 @@ func (r *Record) WriteCSV(writer io.Writer, opts sql.WriteCSVOpts) error {
 		return err
 	}
 	w.Flush()
-	if err := w.Error(); err != nil {
-		return err
-	}
-
-	return nil
+	return w.Error()
 }
 
 // Raw - returns the underlying representation.

pkg/s3select/simdj/record.go

@@ -181,11 +181,7 @@ allElems:
 		return err
 	}
 	w.Flush()
-	if err := w.Error(); err != nil {
-		return err
-	}
-
-	return nil
+	return w.Error()
 }
 
 // Raw - returns the underlying representation.

pkg/s3select/sql/funceval.go

@@ -65,7 +65,7 @@ func (e *FuncExpr) getFunctionName() FuncName {
 	case e.SFunc != nil:
 		return FuncName(strings.ToUpper(e.SFunc.FunctionName))
 	case e.Count != nil:
-		return FuncName(aggFnCount)
+		return aggFnCount
 	case e.Cast != nil:
 		return sqlFnCast
 	case e.Substring != nil:

pkg/smart/types.go

@@ -113,43 +113,6 @@ type nvmeIdentController struct {
 	Vs           [1024]byte              // Vendor Specific
 } // 4096 bytes
 
-type nvmeLBAF struct {
-	Ms uint16
-	Ds uint8
-	Rp uint8
-}
-
-//nolint:deadcode
-type nvmeIdentNamespace struct {
-	Nsze    uint64
-	Ncap    uint64
-	Nuse    uint64
-	Nsfeat  uint8
-	Nlbaf   uint8
-	Flbas   uint8
-	Mc      uint8
-	Dpc     uint8
-	Dps     uint8
-	Nmic    uint8
-	Rescap  uint8
-	Fpi     uint8
-	Rsvd33  uint8
-	Nawun   uint16
-	Nawupf  uint16
-	Nacwu   uint16
-	Nabsn   uint16
-	Nabo    uint16
-	Nabspf  uint16
-	Rsvd46  [2]byte
-	Nvmcap  [16]byte
-	Rsvd64  [40]byte
-	Nguid   [16]byte
-	EUI64   [8]byte
-	Lbaf    [16]nvmeLBAF
-	Rsvd192 [192]byte
-	Vs      [3712]byte
-} // 4096 bytes
-
 //nolint:deadcode
 type nvmeSMARTLog struct {
 	CritWarning      uint8

pkg/sys/stats_linux.go

@@ -70,6 +70,7 @@ func getSysinfoMemoryLimit() (limit uint64, err error) {
 
 	// Total RAM is always the multiplicative value
 	// of unit size and total ram.
+	//nolint:unconvert
 	return uint64(unit) * uint64(totalRAM), nil
 }