non-empty dirs should not be listed as objects (#9129)

This PR fixes a regression introduced in a1c7c9ea73

Dockerfile

@@ -22,6 +22,7 @@ ENV MINIO_ACCESS_KEY_FILE=access_key \
 EXPOSE 9000
 
 COPY --from=0 /go/bin/minio /usr/bin/minio
+COPY CREDITS /third_party/
 COPY dockerscripts/docker-entrypoint.sh /usr/bin/
 
 RUN  \

Dockerfile.arm.release

@@ -0,0 +1,41 @@
+FROM golang:1.13-alpine as builder
+
+WORKDIR /home
+
+ENV GOPATH /go
+ENV CGO_ENABLED 0
+ENV GO111MODULE on
+
+RUN  \
+     apk add --no-cache git 'curl>7.61.0' && \
+     git clone https://github.com/minio/minio && \
+     curl -L https://github.com/balena-io/qemu/releases/download/v3.0.0%2Bresin/qemu-3.0.0+resin-arm.tar.gz | tar zxvf - -C . && mv qemu-3.0.0+resin-arm/qemu-arm-static . 
+
+FROM arm32v7/alpine:3.10
+
+LABEL maintainer="MinIO Inc <dev@min.io>"
+
+COPY dockerscripts/docker-entrypoint.sh /usr/bin/
+COPY CREDITS /third_party/
+COPY --from=builder /home/qemu-arm-static /usr/bin/qemu-arm-static
+
+ENV MINIO_UPDATE off
+ENV MINIO_ACCESS_KEY_FILE=access_key \
+    MINIO_SECRET_KEY_FILE=secret_key \
+    MINIO_KMS_MASTER_KEY_FILE=kms_master_key \
+    MINIO_SSE_MASTER_KEY_FILE=sse_master_key
+
+RUN \
+     apk add --no-cache ca-certificates 'curl>7.61.0' 'su-exec>=0.2' && \
+     echo 'hosts: files mdns4_minimal [NOTFOUND=return] dns mdns4' >> /etc/nsswitch.conf && \
+     curl https://dl.min.io/server/minio/release/linux-arm/minio > /usr/bin/minio && \
+     chmod +x /usr/bin/minio  && \
+     chmod +x /usr/bin/docker-entrypoint.sh
+
+EXPOSE 9000
+
+ENTRYPOINT ["/usr/bin/docker-entrypoint.sh"]
+
+VOLUME ["/data"]
+
+CMD ["minio"]

Dockerfile.arm64.release

@@ -0,0 +1,41 @@
+FROM golang:1.13-alpine as builder
+
+WORKDIR /home
+
+ENV GOPATH /go
+ENV CGO_ENABLED 0
+ENV GO111MODULE on
+
+RUN  \
+     apk add --no-cache git 'curl>7.61.0' && \
+     git clone https://github.com/minio/minio && \
+     curl -L https://github.com/balena-io/qemu/releases/download/v3.0.0%2Bresin/qemu-3.0.0+resin-arm.tar.gz | tar zxvf - -C . && mv qemu-3.0.0+resin-arm/qemu-arm-static . 
+
+FROM arm64v8/alpine:3.10
+
+LABEL maintainer="MinIO Inc <dev@min.io>"
+
+COPY dockerscripts/docker-entrypoint.sh /usr/bin/
+COPY CREDITS /third_party/
+COPY --from=builder /home/qemu-arm-static /usr/bin/qemu-arm-static
+
+ENV MINIO_UPDATE off
+ENV MINIO_ACCESS_KEY_FILE=access_key \
+    MINIO_SECRET_KEY_FILE=secret_key \
+    MINIO_KMS_MASTER_KEY_FILE=kms_master_key \
+    MINIO_SSE_MASTER_KEY_FILE=sse_master_key
+
+RUN \
+     apk add --no-cache ca-certificates 'curl>7.61.0' 'su-exec>=0.2' && \
+     echo 'hosts: files mdns4_minimal [NOTFOUND=return] dns mdns4' >> /etc/nsswitch.conf && \
+     curl https://dl.min.io/server/minio/release/linux-arm64/minio > /usr/bin/minio && \
+     chmod +x /usr/bin/minio  && \
+     chmod +x /usr/bin/docker-entrypoint.sh
+
+EXPOSE 9000
+
+ENTRYPOINT ["/usr/bin/docker-entrypoint.sh"]
+
+VOLUME ["/data"]
+
+CMD ["minio"]

Dockerfile.dev

@@ -4,6 +4,7 @@ LABEL maintainer="MinIO Inc <dev@min.io>"
 
 COPY dockerscripts/docker-entrypoint.sh /usr/bin/
 COPY minio /usr/bin/
+COPY CREDITS /third_party/
 
 ENV MINIO_UPDATE off
 ENV MINIO_ACCESS_KEY_FILE=access_key \

Dockerfile.release

@@ -13,6 +13,7 @@ FROM alpine:3.10
 LABEL maintainer="MinIO Inc <dev@min.io>"
 
 COPY dockerscripts/docker-entrypoint.sh /usr/bin/
+COPY CREDITS /third_party/
 
 ENV MINIO_UPDATE off
 ENV MINIO_ACCESS_KEY_FILE=access_key \

cmd/admin-handlers.go

@@ -141,9 +141,9 @@ func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Req
 		}
 
 		if runtime.GOOS == "windows" {
-			u.Path = path.Dir(u.Path) + "minio.exe"
+			u.Path = path.Dir(u.Path) + SlashSeparator + "minio.exe"
 		} else {
-			u.Path = path.Dir(u.Path) + "minio"
+			u.Path = path.Dir(u.Path) + SlashSeparator + "minio"
 		}
 
 		updateURL = u.String()
@@ -982,13 +982,24 @@ func (a adminAPIHandlers) BackgroundHealStatusHandler(w http.ResponseWriter, r *
 	}
 
 	// Aggregate healing result
-	var aggregatedHealStateResult = madmin.BgHealState{}
+	var aggregatedHealStateResult = madmin.BgHealState{
+		ScannedItemsCount: bgHealStates[0].ScannedItemsCount,
+		LastHealActivity:  bgHealStates[0].LastHealActivity,
+		NextHealRound:     bgHealStates[0].NextHealRound,
+	}
+
+	bgHealStates = bgHealStates[1:]
+
 	for _, state := range bgHealStates {
 		aggregatedHealStateResult.ScannedItemsCount += state.ScannedItemsCount
-		if aggregatedHealStateResult.LastHealActivity.Before(state.LastHealActivity) {
+		if !state.LastHealActivity.IsZero() && aggregatedHealStateResult.LastHealActivity.Before(state.LastHealActivity) {
 			aggregatedHealStateResult.LastHealActivity = state.LastHealActivity
+			// The node which has the last heal activity means its
+			// is the node that is orchestrating self healing operations,
+			// which also means it is the same node which decides when
+			// the next self healing operation will be done.
+			aggregatedHealStateResult.NextHealRound = state.NextHealRound
 		}
-
 	}
 
 	if err := json.NewEncoder(w).Encode(aggregatedHealStateResult); err != nil {
@@ -1398,10 +1409,10 @@ func (a adminAPIHandlers) ServerInfoHandler(w http.ResponseWriter, r *http.Reque
 		} else if ldapConn == nil {
 			ldap.Status = "Not Configured"
 		} else {
+			// Close ldap connection to avoid leaks.
+			ldapConn.Close()
 			ldap.Status = "online"
 		}
-		// Close ldap connection to avoid leaks.
-		defer ldapConn.Close()
 	}
 
 	log, audit := fetchLoggerInfo(cfg)

cmd/admin-heal-ops.go

@@ -575,8 +575,6 @@ func (h *healSequence) queueHealTask(path string, healType madmin.HealItemType)
 }
 
 func (h *healSequence) healItemsFromSourceCh() error {
-	h.lastHealActivity = UTCNow()
-
 	bucketsOnly := true // heal buckets only, not objects.
 	if err := h.healItems(bucketsOnly); err != nil {
 		logger.LogIf(h.ctx, err)

cmd/api-errors.go

@@ -1641,7 +1641,7 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 		apiErr = ErrKMSNotConfigured
 	case crypto.ErrKMSAuthLogin:
 		apiErr = ErrKMSAuthFailure
-	case errOperationTimedOut, context.Canceled, context.DeadlineExceeded:
+	case context.Canceled, context.DeadlineExceeded:
 		apiErr = ErrOperationTimedOut
 	case errDiskNotFound:
 		apiErr = ErrSlowDown
@@ -1769,6 +1769,8 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 		apiErr = ErrOverlappingFilterNotification
 	case *event.ErrUnsupportedConfiguration:
 		apiErr = ErrUnsupportedNotification
+	case OperationTimedOut:
+		apiErr = ErrOperationTimedOut
 	case BackendDown:
 		apiErr = ErrBackendDown
 	case ObjectNameTooLong:

cmd/api-headers.go

@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/minio/minio/cmd/crypto"
@@ -81,6 +82,10 @@ func setObjectHeaders(w http.ResponseWriter, objInfo ObjectInfo, rs *HTTPRangeSp
 		w.Header()[xhttp.ETag] = []string{"\"" + objInfo.ETag + "\""}
 	}
 
+	if strings.Contains(objInfo.ETag, "-") && len(objInfo.Parts) > 0 {
+		w.Header().Set(xhttp.AmzMpPartsCount, strconv.Itoa(len(objInfo.Parts)))
+	}
+
 	if objInfo.ContentType != "" {
 		w.Header().Set(xhttp.ContentType, objInfo.ContentType)
 	}

cmd/data-usage.go

@@ -74,15 +74,15 @@ func timeToCrawl(ctx context.Context, objAPI ObjectLayer) time.Duration {
 	if dataUsageInfo.LastUpdate.IsZero() {
 		return 1 * time.Second
 	}
-	waitDuration := dataUsageInfo.LastUpdate.Sub(UTCNow())
-	if waitDuration > dataUsageCrawlInterval {
+	timeSinceLastUpdate := UTCNow().Sub(dataUsageInfo.LastUpdate)
+	if timeSinceLastUpdate > dataUsageCrawlInterval {
 		// Waited long enough start crawl in a 1 second
 		return 1 * time.Second
 	}
 	// No crawling needed, ask the routine to wait until
 	// the daily interval 12hrs - delta between last update
 	// with current time.
-	return dataUsageCrawlInterval - waitDuration
+	return dataUsageCrawlInterval - timeSinceLastUpdate
 }
 
 var dataUsageLockTimeout = lifecycleLockTimeout

cmd/disk-cache-backend.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2019 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2019-2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,7 +37,6 @@ import (
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/disk"
 	"github.com/minio/sio"
-	"github.com/ncw/directio"
 	"go.uber.org/atomic"
 )
 
@@ -158,7 +157,7 @@ func newDiskCache(dir string, quotaPct, after, lowWatermark, highWatermark int)
 		onlineMutex:   &sync.RWMutex{},
 		pool: sync.Pool{
 			New: func() interface{} {
-				b := directio.AlignedBlock(int(cacheBlkSize))
+				b := disk.AlignedBlock(int(cacheBlkSize))
 				return &b
 			},
 		},

cmd/endpoint-ellipses.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2018 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2018-2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -108,12 +108,24 @@ func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint6
 		return ss
 	}
 
+	setCounts := possibleSetCounts(commonSize)
+	if len(setCounts) == 0 {
+		msg := fmt.Sprintf("Incorrect number of endpoints provided %s, number of disks %d is not divisible by any supported erasure set sizes %d", args, commonSize, setSizes)
+		return nil, config.ErrInvalidNumberOfErasureEndpoints(nil).Msg(msg)
+	}
+
 	if customSetDriveCount > 0 {
-		msg := fmt.Sprintf("Invalid set drive count, leads to non-uniform distribution for the given number of disks. Possible values for custom set count are %d", possibleSetCounts(setSize))
+		msg := fmt.Sprintf("Invalid set drive count. Acceptable values for %d number drives are %d", commonSize, setCounts)
 		if customSetDriveCount > setSize {
 			return nil, config.ErrInvalidErasureSetSize(nil).Msg(msg)
 		}
-		if setSize%customSetDriveCount != 0 {
+		var found bool
+		for _, ss := range setCounts {
+			if ss == customSetDriveCount {
+				found = true
+			}
+		}
+		if !found {
 			return nil, config.ErrInvalidErasureSetSize(nil).Msg(msg)
 		}
 		setSize = customSetDriveCount
@@ -121,7 +133,7 @@ func getSetIndexes(args []string, totalSizes []uint64, customSetDriveCount uint6
 
 	// Check whether setSize is with the supported range.
 	if !isValidSetSize(setSize) {
-		msg := fmt.Sprintf("Incorrect number of endpoints provided %s", args)
+		msg := fmt.Sprintf("Incorrect number of endpoints provided %s, number of disks %d is not divisible by any supported erasure set sizes %d", args, commonSize, setSizes)
 		return nil, config.ErrInvalidNumberOfErasureEndpoints(nil).Msg(msg)
 	}
 
@@ -261,9 +273,6 @@ func createServerEndpoints(serverAddr string, args ...string) (
 		if err != nil {
 			return nil, -1, -1, config.ErrInvalidErasureSetSize(err)
 		}
-		if !isValidSetSize(uint64(setDriveCount)) {
-			return nil, -1, -1, config.ErrInvalidErasureSetSize(nil)
-		}
 	}
 
 	if !ellipses.HasEllipses(args...) {

cmd/endpoint-ellipses_test.go

@@ -104,6 +104,36 @@ func TestGetSetIndexesEnvOverride(t *testing.T) {
 			8,
 			true,
 		},
+		{
+			[]string{"http://host{1...12}/data{1...12}"},
+			[]uint64{144},
+			[][]uint64{{16, 16, 16, 16, 16, 16, 16, 16, 16}},
+			16,
+			true,
+		},
+		{
+			[]string{"http://host{0...5}/data{1...28}"},
+			[]uint64{168},
+			[][]uint64{{12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12, 12}},
+			12,
+			true,
+		},
+		// Incorrect custom set drive count.
+		{
+			[]string{"http://host{0...5}/data{1...28}"},
+			[]uint64{168},
+			nil,
+			10,
+			false,
+		},
+		// Failure not divisible number of disks.
+		{
+			[]string{"http://host{1...11}/data{1...11}"},
+			[]uint64{121},
+			nil,
+			11,
+			false,
+		},
 		{
 			[]string{"data{1...60}"},
 			nil,

cmd/format-xl.go

@@ -426,7 +426,7 @@ func checkFormatXLValue(formatXL *formatXLV3) error {
 }
 
 // Check all format values.
-func checkFormatXLValues(formats []*formatXLV3) error {
+func checkFormatXLValues(formats []*formatXLV3, drivesPerSet int) error {
 	for i, formatXL := range formats {
 		if formatXL == nil {
 			continue
@@ -438,6 +438,9 @@ func checkFormatXLValues(formats []*formatXLV3) error {
 			return fmt.Errorf("%s disk is already being used in another erasure deployment. (Number of disks specified: %d but the number of disks found in the %s disk's format.json: %d)",
 				humanize.Ordinal(i+1), len(formats), humanize.Ordinal(i+1), len(formatXL.XL.Sets)*len(formatXL.XL.Sets[0]))
 		}
+		if len(formatXL.XL.Sets[0]) != drivesPerSet {
+			return fmt.Errorf("%s disk is already formatted with %d drives per erasure set. This cannot be changed to %d, please revert your MINIO_ERASURE_SET_DRIVE_COUNT setting", humanize.Ordinal(i+1), len(formatXL.XL.Sets[0]), drivesPerSet)
+		}
 	}
 	return nil
 }

cmd/fs-v1-helpers.go

@@ -429,7 +429,7 @@ func fsDeleteFile(ctx context.Context, basePath, deletePath string) error {
 		return err
 	}
 
-	if err := deleteFile(basePath, deletePath); err != nil {
+	if err := deleteFile(basePath, deletePath, false); err != nil {
 		if err != errFileNotFound {
 			logger.LogIf(ctx, err)
 		}

cmd/fs-v1.go

@@ -1058,15 +1058,18 @@ func (fs *FSObjects) DeleteObject(ctx context.Context, bucket, object string) er
 // is a leaf or non-leaf entry.
 func (fs *FSObjects) listDirFactory() ListDirFunc {
 	// listDir - lists all the entries at a given prefix and given entry in the prefix.
-	listDir := func(bucket, prefixDir, prefixEntry string) (entries []string) {
+	listDir := func(bucket, prefixDir, prefixEntry string) (emptyDir bool, entries []string) {
 		var err error
 		entries, err = readDir(pathJoin(fs.fsPath, bucket, prefixDir))
 		if err != nil && err != errFileNotFound {
 			logger.LogIf(context.Background(), err)
-			return
+			return false, nil
+		}
+		if len(entries) == 0 {
+			return true, nil
 		}
 		sort.Strings(entries)
-		return filterMatchingPrefix(entries, prefixEntry)
+		return false, filterMatchingPrefix(entries, prefixEntry)
 	}
 
 	// Return list factory instance.

cmd/gateway/hdfs/gateway-hdfs.go

@@ -326,7 +326,7 @@ func (n *hdfsObjects) ListBuckets(ctx context.Context) (buckets []minio.BucketIn
 
 func (n *hdfsObjects) listDirFactory() minio.ListDirFunc {
 	// listDir - lists all the entries at a given prefix and given entry in the prefix.
-	listDir := func(bucket, prefixDir, prefixEntry string) (entries []string) {
+	listDir := func(bucket, prefixDir, prefixEntry string) (emptyDir bool, entries []string) {
 		f, err := n.clnt.Open(minio.PathJoin(hdfsSeparator, bucket, prefixDir))
 		if err != nil {
 			if os.IsNotExist(err) {
@@ -341,6 +341,9 @@ func (n *hdfsObjects) listDirFactory() minio.ListDirFunc {
 			logger.LogIf(context.Background(), err)
 			return
 		}
+		if len(fis) == 0 {
+			return true, nil
+		}
 		for _, fi := range fis {
 			if fi.IsDir() {
 				entries = append(entries, fi.Name()+hdfsSeparator)
@@ -348,7 +351,7 @@ func (n *hdfsObjects) listDirFactory() minio.ListDirFunc {
 				entries = append(entries, fi.Name())
 			}
 		}
-		return minio.FilterMatchingPrefix(entries, prefixEntry)
+		return false, minio.FilterMatchingPrefix(entries, prefixEntry)
 	}
 
 	// Return list factory instance.

cmd/global-heal.go

@@ -29,7 +29,6 @@ import (
 const (
 	bgHealingUUID = "0000-0000-0000-0000"
 	leaderTick    = time.Hour
-	healTick      = time.Hour
 	healInterval  = 30 * 24 * time.Hour
 )
 
@@ -75,6 +74,7 @@ func getLocalBackgroundHealStatus() madmin.BgHealState {
 	return madmin.BgHealState{
 		ScannedItemsCount: bgSeq.scannedItemsCount,
 		LastHealActivity:  bgSeq.lastHealActivity,
+		NextHealRound:     UTCNow().Add(durationToNextHealRound(bgSeq.lastHealActivity)),
 	}
 }
 
@@ -112,6 +112,19 @@ func healErasureSet(ctx context.Context, setIndex int, xlObj *xlObjects) error {
 	return nil
 }
 
+// Returns the duration to the next background healing round
+func durationToNextHealRound(lastHeal time.Time) time.Duration {
+	if lastHeal.IsZero() {
+		lastHeal = globalBootTime
+	}
+
+	d := lastHeal.Add(healInterval).Sub(UTCNow())
+	if d < 0 {
+		return time.Second
+	}
+	return d
+}
+
 // Healing leader will take the charge of healing all erasure sets
 func execLeaderTasks(z *xlZones) {
 	ctx := context.Background()
@@ -132,10 +145,7 @@ func execLeaderTasks(z *xlZones) {
 
 	lastScanTime := time.Now() // So that we don't heal immediately, but after one month.
 	for {
-		if time.Since(lastScanTime) < healInterval {
-			time.Sleep(healTick)
-			continue
-		}
+		time.Sleep(durationToNextHealRound(lastScanTime))
 		for _, zone := range z.zones {
 			// Heal set by set
 			for i, set := range zone.sets {

cmd/http/headers.go

@@ -77,6 +77,9 @@ const (
 	AmzObjectLockLegalHold        = "X-Amz-Object-Lock-Legal-Hold"
 	AmzObjectLockBypassGovernance = "X-Amz-Bypass-Governance-Retention"
 
+	// Multipart parts count
+	AmzMpPartsCount = "x-amz-mp-parts-count"
+
 	// Dummy putBucketACL
 	AmzACL = "x-amz-acl"
 

cmd/naughty-disk_test.go

@@ -196,6 +196,13 @@ func (d *naughtyDisk) DeleteFileBulk(volume string, paths []string) ([]error, er
 	return errs, nil
 }
 
+func (d *naughtyDisk) DeletePrefixes(volume string, paths []string) ([]error, error) {
+	if err := d.calcError(); err != nil {
+		return nil, err
+	}
+	return d.disk.DeletePrefixes(volume, paths)
+}
+
 func (d *naughtyDisk) WriteAll(volume string, path string, reader io.Reader) (err error) {
 	if err := d.calcError(); err != nil {
 		return err

cmd/object-api-common.go

@@ -147,78 +147,6 @@ func cleanupDir(ctx context.Context, storage StorageAPI, volume, dirPath string)
 	return err
 }
 
-// Cleanup objects in bulk and recursively: each object will have a list of sub-files to delete in the backend
-func cleanupObjectsBulk(storage StorageAPI, volume string, objsPaths []string, errs []error) ([]error, error) {
-	// The list of files in disk to delete
-	var filesToDelete []string
-	// Map files to delete to the passed objsPaths
-	var filesToDeleteObjsIndexes []int
-
-	// Traverse and return the list of sub entries
-	var traverse func(string) ([]string, error)
-	traverse = func(entryPath string) ([]string, error) {
-		var output = make([]string, 0)
-		if !HasSuffix(entryPath, SlashSeparator) {
-			output = append(output, entryPath)
-			return output, nil
-		}
-		entries, err := storage.ListDir(volume, entryPath, -1, "")
-		if err != nil {
-			if err == errFileNotFound {
-				return nil, nil
-			}
-			return nil, err
-		}
-
-		for _, entry := range entries {
-			subEntries, err := traverse(pathJoin(entryPath, entry))
-			if err != nil {
-				return nil, err
-			}
-			output = append(output, subEntries...)
-		}
-		return output, nil
-	}
-
-	// Find and collect the list of files to remove associated
-	// to the passed objects paths
-	for idx, objPath := range objsPaths {
-		if errs[idx] != nil {
-			continue
-		}
-		output, err := traverse(retainSlash(pathJoin(objPath)))
-		if err != nil {
-			errs[idx] = err
-			continue
-		} else {
-			errs[idx] = nil
-		}
-		filesToDelete = append(filesToDelete, output...)
-		for i := 0; i < len(output); i++ {
-			filesToDeleteObjsIndexes = append(filesToDeleteObjsIndexes, idx)
-		}
-	}
-
-	// Reverse the list so remove can succeed
-	reverseStringSlice(filesToDelete)
-
-	dErrs, err := storage.DeleteFileBulk(volume, filesToDelete)
-	if err != nil {
-		return nil, err
-	}
-
-	// Map files deletion errors to the correspondent objects
-	for i := range dErrs {
-		if dErrs[i] != nil {
-			if errs[filesToDeleteObjsIndexes[i]] != nil {
-				errs[filesToDeleteObjsIndexes[i]] = dErrs[i]
-			}
-		}
-	}
-
-	return errs, nil
-}
-
 // Removes notification.xml for a given bucket, only used during DeleteBucket.
 func removeNotificationConfig(ctx context.Context, objAPI ObjectLayer, bucket string) error {
 	// Verify bucket is valid.

cmd/object-api-listobjects_test.go

@@ -45,6 +45,8 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		// Will not store any objects in this bucket,
 		// Its to test ListObjects on an empty bucket.
 		"empty-bucket",
+		// Listing the case where the marker > last object.
+		"test-bucket-single-object",
 	}
 	for _, bucket := range testBuckets {
 		err := obj.MakeBucketWithLocation(context.Background(), bucket, "")
@@ -72,6 +74,7 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{testBuckets[1], "obj1", "obj1", nil},
 		{testBuckets[1], "obj2", "obj2", nil},
 		{testBuckets[1], "temporary/0/", "", nil},
+		{testBuckets[3], "A/B", "contentstring", nil},
 	}
 	for _, object := range testObjects {
 		md5Bytes := md5.Sum([]byte(object.content))
@@ -445,6 +448,11 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 				{Name: "temporary/0/"},
 			},
 		},
+		// ListObjectsResult-34 Listing with marker > last object should return empty
+		{
+			IsTruncated: false,
+			Objects:     []ObjectInfo{},
+		},
 	}
 
 	testCases := []struct {
@@ -559,6 +567,8 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{"test-bucket-empty-dir", "", "", SlashSeparator, 10, resultCases[32], nil, true},
 		// Test listing a directory which contains an empty directory (64)
 		{"test-bucket-empty-dir", "", "temporary/", "", 10, resultCases[33], nil, true},
+		// Test listing with marker > last object such that response should be empty (65)
+		{"test-bucket-single-object", "", "A/C", "", 1000, resultCases[34], nil, true},
 	}
 
 	for i, testCase := range testCases {

cmd/posix-diskid-check.go

@@ -179,6 +179,13 @@ func (p *posixDiskIDCheck) DeleteFileBulk(volume string, paths []string) (errs [
 	return p.storage.DeleteFileBulk(volume, paths)
 }
 
+func (p *posixDiskIDCheck) DeletePrefixes(volume string, paths []string) (errs []error, err error) {
+	if p.isDiskStale() {
+		return nil, errDiskNotFound
+	}
+	return p.storage.DeletePrefixes(volume, paths)
+}
+
 func (p *posixDiskIDCheck) VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
 	if p.isDiskStale() {
 		return errDiskNotFound

cmd/posix.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2016-2019 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2016-2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -44,7 +44,6 @@ import (
 	"github.com/minio/minio/pkg/disk"
 	xioutil "github.com/minio/minio/pkg/ioutil"
 	"github.com/minio/minio/pkg/mountinfo"
-	"github.com/ncw/directio"
 )
 
 const (
@@ -213,7 +212,7 @@ func newPosix(path string) (*posix, error) {
 		diskPath: path,
 		pool: sync.Pool{
 			New: func() interface{} {
-				b := directio.AlignedBlock(readBlockSize)
+				b := disk.AlignedBlock(readBlockSize)
 				return &b
 			},
 		},
@@ -670,13 +669,16 @@ func (s *posix) Walk(volume, dirPath, marker string, recursive bool, leafFile st
 	ch = make(chan FileInfo)
 	go func() {
 		defer close(ch)
-		listDir := func(volume, dirPath, dirEntry string) (entries []string) {
+		listDir := func(volume, dirPath, dirEntry string) (emptyDir bool, entries []string) {
 			entries, err := s.ListDir(volume, dirPath, -1, leafFile)
 			if err != nil {
 				return
 			}
+			if len(entries) == 0 {
+				return true, nil
+			}
 			sort.Strings(entries)
-			return filterMatchingPrefix(entries, dirEntry)
+			return false, filterMatchingPrefix(entries, dirEntry)
 		}
 
 		walkResultCh := startTreeWalk(context.Background(), volume, dirPath, marker, recursive, listDir, endWalkCh)
@@ -1265,16 +1267,28 @@ func (s *posix) StatFile(volume, path string) (file FileInfo, err error) {
 	}, nil
 }
 
-// deleteFile deletes a file path if its empty. If it's successfully deleted,
-// it will recursively move up the tree, deleting empty parent directories
-// until it finds one with files in it. Returns nil for a non-empty directory.
-func deleteFile(basePath, deletePath string) error {
-	if basePath == deletePath {
+// deleteFile deletes a file or a directory if its empty unless recursive
+// is set to true. If the target is successfully deleted, it will recursively
+// move up the tree, deleting empty parent directories until it finds one
+// with files in it. Returns nil for a non-empty directory even when
+// recursive is set to false.
+func deleteFile(basePath, deletePath string, recursive bool) error {
+	if basePath == "" || deletePath == "" {
+		return nil
+	}
+	basePath = filepath.Clean(basePath)
+	deletePath = filepath.Clean(deletePath)
+	if !strings.HasPrefix(deletePath, basePath) || deletePath == basePath {
 		return nil
 	}
 
-	// Attempt to remove path.
-	if err := os.Remove((deletePath)); err != nil {
+	var err error
+	if recursive {
+		err = os.RemoveAll(deletePath)
+	} else {
+		err = os.Remove(deletePath)
+	}
+	if err != nil {
 		switch {
 		case isSysErrNotEmpty(err):
 			// Ignore errors if the directory is not empty. The server relies on
@@ -1297,12 +1311,58 @@ func deleteFile(basePath, deletePath string) error {
 	deletePath = strings.TrimSuffix(deletePath, SlashSeparator)
 	deletePath = slashpath.Dir(deletePath)
 
-	// Delete parent directory. Errors for parent directories shouldn't trickle down.
-	deleteFile(basePath, deletePath)
+	// Delete parent directory obviously not recursively. Errors for
+	// parent directories shouldn't trickle down.
+	deleteFile(basePath, deletePath, false)
 
 	return nil
 }
 
+// DeletePrefixes forcibly deletes all the contents of a set of specified paths.
+// Parent directories are automatically removed if they become empty. err can
+// bil nil while errs can contain some errors for corresponding objects. No error
+// is set if a specified prefix path does not exist.
+func (s *posix) DeletePrefixes(volume string, paths []string) (errs []error, err error) {
+	atomic.AddInt32(&s.activeIOCount, 1)
+	defer func() {
+		atomic.AddInt32(&s.activeIOCount, -1)
+	}()
+
+	volumeDir, err := s.getVolDir(volume)
+	if err != nil {
+		return nil, err
+	}
+
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, errVolumeNotFound
+		} else if os.IsPermission(err) {
+			return nil, errVolumeAccessDenied
+		} else if isSysErrIO(err) {
+			return nil, errFaultyDisk
+		}
+		return nil, err
+	}
+
+	errs = make([]error, len(paths))
+	// Following code is needed so that we retain SlashSeparator
+	// suffix if any in path argument.
+	for idx, path := range paths {
+		filePath := pathJoin(volumeDir, path)
+		errs[idx] = checkPathLength(filePath)
+		if errs[idx] != nil {
+			continue
+		}
+		// Delete file or a directory recursively, delete parent
+		// directory as well if its empty.
+		errs[idx] = deleteFile(volumeDir, filePath, true)
+	}
+
+	return
+}
+
 // DeleteFile - delete a file at path.
 func (s *posix) DeleteFile(volume, path string) (err error) {
 	atomic.AddInt32(&s.activeIOCount, 1)
@@ -1316,7 +1376,7 @@ func (s *posix) DeleteFile(volume, path string) (err error) {
 	}
 
 	// Stat a volume entry.
-	_, err = os.Stat((volumeDir))
+	_, err = os.Stat(volumeDir)
 	if err != nil {
 		if os.IsNotExist(err) {
 			return errVolumeNotFound
@@ -1331,18 +1391,49 @@ func (s *posix) DeleteFile(volume, path string) (err error) {
 	// Following code is needed so that we retain SlashSeparator suffix if any in
 	// path argument.
 	filePath := pathJoin(volumeDir, path)
-	if err = checkPathLength((filePath)); err != nil {
+	if err = checkPathLength(filePath); err != nil {
 		return err
 	}
 
 	// Delete file and delete parent directory as well if its empty.
-	return deleteFile(volumeDir, filePath)
+	return deleteFile(volumeDir, filePath, false)
 }
 
 func (s *posix) DeleteFileBulk(volume string, paths []string) (errs []error, err error) {
+	atomic.AddInt32(&s.activeIOCount, 1)
+	defer func() {
+		atomic.AddInt32(&s.activeIOCount, -1)
+	}()
+
+	volumeDir, err := s.getVolDir(volume)
+	if err != nil {
+		return nil, err
+	}
+
+	// Stat a volume entry.
+	_, err = os.Stat(volumeDir)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, errVolumeNotFound
+		} else if os.IsPermission(err) {
+			return nil, errVolumeAccessDenied
+		} else if isSysErrIO(err) {
+			return nil, errFaultyDisk
+		}
+		return nil, err
+	}
+
 	errs = make([]error, len(paths))
+	// Following code is needed so that we retain SlashSeparator
+	// suffix if any in path argument.
 	for idx, path := range paths {
-		errs[idx] = s.DeleteFile(volume, path)
+		filePath := pathJoin(volumeDir, path)
+		errs[idx] = checkPathLength(filePath)
+		if errs[idx] != nil {
+			continue
+		}
+		// Delete file and delete parent directory as well if its empty.
+		errs[idx] = deleteFile(volumeDir, filePath, false)
 	}
 	return
 }
@@ -1429,7 +1520,7 @@ func (s *posix) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) (err e
 
 	// Remove parent dir of the source file if empty
 	if parentDir := slashpath.Dir(srcFilePath); isDirEmpty(parentDir) {
-		deleteFile(srcVolumeDir, parentDir)
+		deleteFile(srcVolumeDir, parentDir, false)
 	}
 
 	return nil

cmd/prepare-storage.go

@@ -244,6 +244,10 @@ func connectLoadInitFormats(retryCount int, firstDisk bool, endpoints Endpoints,
 		if _, ok := formatCriticalErrors[sErr]; ok {
 			return nil, fmt.Errorf("Disk %s: %w", endpoints[i], sErr)
 		}
+		// not critical error but still print the error, nonetheless, which is perhaps unhandled
+		if sErr != errUnformattedDisk && sErr != errDiskNotFound && retryCount >= 5 {
+			logger.Info("Unable to read 'format.json' from %s: %v\n", endpoints[i], sErr)
+		}
 	}
 
 	// Pre-emptively check if one of the formatted disks
@@ -251,7 +255,7 @@ func connectLoadInitFormats(retryCount int, firstDisk bool, endpoints Endpoints,
 	// most part unless one of the formats is not consistent
 	// with expected XL format. For example if a user is
 	// trying to pool FS backend into an XL set.
-	if err := checkFormatXLValues(formatConfigs); err != nil {
+	if err := checkFormatXLValues(formatConfigs, drivesPerSet); err != nil {
 		return nil, err
 	}
 

cmd/storage-interface.go

@@ -55,6 +55,7 @@ type StorageAPI interface {
 	StatFile(volume string, path string) (file FileInfo, err error)
 	DeleteFile(volume string, path string) (err error)
 	DeleteFileBulk(volume string, paths []string) (errs []error, err error)
+	DeletePrefixes(volume string, paths []string) (errs []error, err error)
 	VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error
 
 	// Write all data, syncs the data to disk.

cmd/storage-rest-client.go

@@ -414,13 +414,54 @@ func (client *storageRESTClient) DeleteFileBulk(volume string, paths []string) (
 
 	respBody, err := client.call(storageRESTMethodDeleteFileBulk, values, &buffer, -1)
 	defer http.DrainBody(respBody)
+	if err != nil {
+		return nil, err
+	}
 
+	reader, err := clearLeadingSpaces(respBody)
 	if err != nil {
 		return nil, err
 	}
 
 	dErrResp := &DeleteFileBulkErrsResp{}
-	if err = gob.NewDecoder(respBody).Decode(dErrResp); err != nil {
+	if err = gob.NewDecoder(reader).Decode(dErrResp); err != nil {
+		return nil, err
+	}
+
+	for _, dErr := range dErrResp.Errs {
+		errs = append(errs, toStorageErr(dErr))
+	}
+
+	return errs, nil
+}
+
+// DeletePrefixes - deletes prefixes in bulk.
+func (client *storageRESTClient) DeletePrefixes(volume string, paths []string) (errs []error, err error) {
+	if len(paths) == 0 {
+		return errs, err
+	}
+	values := make(url.Values)
+	values.Set(storageRESTVolume, volume)
+
+	var buffer bytes.Buffer
+	for _, path := range paths {
+		buffer.WriteString(path)
+		buffer.WriteString("\n")
+	}
+
+	respBody, err := client.call(storageRESTMethodDeletePrefixes, values, &buffer, -1)
+	defer http.DrainBody(respBody)
+	if err != nil {
+		return nil, err
+	}
+
+	reader, err := clearLeadingSpaces(respBody)
+	if err != nil {
+		return nil, err
+	}
+
+	dErrResp := &DeletePrefixesErrsResp{}
+	if err = gob.NewDecoder(reader).Decode(dErrResp); err != nil {
 		return nil, err
 	}
 
@@ -443,6 +484,22 @@ func (client *storageRESTClient) RenameFile(srcVolume, srcPath, dstVolume, dstPa
 	return err
 }
 
+// clearLeadingSpaces removes all the first spaces returned from a reader.
+func clearLeadingSpaces(r io.Reader) (io.Reader, error) {
+	reader := bufio.NewReader(r)
+	for {
+		b, err := reader.ReadByte()
+		if err != nil {
+			return nil, err
+		}
+		if b != ' ' {
+			reader.UnreadByte()
+			break
+		}
+	}
+	return reader, nil
+}
+
 func (client *storageRESTClient) VerifyFile(volume, path string, size int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
 	values := make(url.Values)
 	values.Set(storageRESTVolume, volume)
@@ -457,16 +514,9 @@ func (client *storageRESTClient) VerifyFile(volume, path string, size int64, alg
 	if err != nil {
 		return err
 	}
-	reader := bufio.NewReader(respBody)
-	for {
-		b, err := reader.ReadByte()
-		if err != nil {
-			return err
-		}
-		if b != ' ' {
-			reader.UnreadByte()
-			break
-		}
+	reader, err := clearLeadingSpaces(respBody)
+	if err != nil {
+		return err
 	}
 	verifyResp := &VerifyFileResp{}
 	if err = gob.NewDecoder(reader).Decode(verifyResp); err != nil {

cmd/storage-rest-common.go

@@ -17,7 +17,7 @@
 package cmd
 
 const (
-	storageRESTVersion       = "v14" // DeleteFileBulk API change
+	storageRESTVersion       = "v15" // Adding DeletePrefixes API
 	storageRESTVersionPrefix = SlashSeparator + storageRESTVersion
 	storageRESTPrefix        = minioReservedBucketPath + "/storage"
 )
@@ -42,6 +42,7 @@ const (
 	storageRESTMethodWalk           = "/walk"
 	storageRESTMethodDeleteFile     = "/deletefile"
 	storageRESTMethodDeleteFileBulk = "/deletefilebulk"
+	storageRESTMethodDeletePrefixes = "/deleteprefixes"
 	storageRESTMethodRenameFile     = "/renamefile"
 	storageRESTMethodVerifyFile     = "/verifyfile"
 )

cmd/storage-rest-server.go

@@ -508,20 +508,69 @@ func (s *storageRESTServer) DeleteFileBulkHandler(w http.ResponseWriter, r *http
 		return
 	}
 
+	w.Header().Set(xhttp.ContentType, "text/event-stream")
+	encoder := gob.NewEncoder(w)
+	doneCh := sendWhiteSpaceToHTTPResponse(w)
 	errs, err := s.storage.DeleteFileBulk(volume, filePaths)
+	<-doneCh
 	if err != nil {
 		s.writeErrorResponse(w, err)
 		return
 	}
 
-	derrsResp := &DeleteFileBulkErrsResp{Errs: make([]error, len(errs))}
+	dErrsResp := &DeleteFileBulkErrsResp{Errs: make([]error, len(errs))}
 	for idx, err := range errs {
 		if err != nil {
-			derrsResp.Errs[idx] = StorageErr(err.Error())
+			dErrsResp.Errs[idx] = StorageErr(err.Error())
 		}
 	}
 
-	gob.NewEncoder(w).Encode(derrsResp)
+	encoder.Encode(dErrsResp)
+	w.(http.Flusher).Flush()
+}
+
+// DeletePrefixesErrsResp - collection of delete errors
+// for bulk prefixes deletes
+type DeletePrefixesErrsResp struct {
+	Errs []error
+}
+
+// DeletePrefixesHandler - delete a set of a prefixes.
+func (s *storageRESTServer) DeletePrefixesHandler(w http.ResponseWriter, r *http.Request) {
+	if !s.IsValid(w, r) {
+		return
+	}
+	vars := r.URL.Query()
+	volume := vars.Get(storageRESTVolume)
+
+	bio := bufio.NewScanner(r.Body)
+	var prefixes []string
+	for bio.Scan() {
+		prefixes = append(prefixes, bio.Text())
+	}
+
+	if err := bio.Err(); err != nil {
+		s.writeErrorResponse(w, err)
+		return
+	}
+
+	w.Header().Set(xhttp.ContentType, "text/event-stream")
+	encoder := gob.NewEncoder(w)
+	doneCh := sendWhiteSpaceToHTTPResponse(w)
+	errs, err := s.storage.DeletePrefixes(volume, prefixes)
+	<-doneCh
+	if err != nil {
+		s.writeErrorResponse(w, err)
+		return
+	}
+
+	dErrsResp := &DeletePrefixesErrsResp{Errs: make([]error, len(errs))}
+	for idx, err := range errs {
+		if err != nil {
+			dErrsResp.Errs[idx] = StorageErr(err.Error())
+		}
+	}
+	encoder.Encode(dErrsResp)
 	w.(http.Flusher).Flush()
 }
 
@@ -665,6 +714,8 @@ func registerStorageRESTHandlers(router *mux.Router, endpointZones EndpointZones
 				Queries(restQueries(storageRESTVolume, storageRESTDirPath, storageRESTCount, storageRESTLeafFile)...)
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodWalk).HandlerFunc(httpTraceHdrs(server.WalkHandler)).
 				Queries(restQueries(storageRESTVolume, storageRESTDirPath, storageRESTMarkerPath, storageRESTRecursive, storageRESTLeafFile)...)
+			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodDeletePrefixes).HandlerFunc(httpTraceHdrs(server.DeletePrefixesHandler)).
+				Queries(restQueries(storageRESTVolume)...)
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodDeleteFile).HandlerFunc(httpTraceHdrs(server.DeleteFileHandler)).
 				Queries(restQueries(storageRESTVolume, storageRESTFilePath)...)
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodDeleteFileBulk).HandlerFunc(httpTraceHdrs(server.DeleteFileBulkHandler)).

cmd/tree-walk.go

@@ -56,10 +56,10 @@ func filterMatchingPrefix(entries []string, prefixEntry string) []string {
 }
 
 // ListDirFunc - "listDir" function of type listDirFunc returned by listDirFactory() - explained below.
-type ListDirFunc func(bucket, prefixDir, prefixEntry string) (entries []string)
+type ListDirFunc func(bucket, prefixDir, prefixEntry string) (emptyDir bool, entries []string)
 
 // treeWalk walks directory tree recursively pushing TreeWalkResult into the channel as and when it encounters files.
-func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker string, recursive bool, listDir ListDirFunc, resultCh chan TreeWalkResult, endWalkCh <-chan struct{}, isEnd bool) (totalNum int, treeErr error) {
+func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker string, recursive bool, listDir ListDirFunc, resultCh chan TreeWalkResult, endWalkCh <-chan struct{}, isEnd bool) (emptyDir bool, treeErr error) {
 	// Example:
 	// if prefixDir="one/two/three/" and marker="four/five.txt" treeWalk is recursively
 	// called with prefixDir="one/two/three/four/" and marker="five.txt"
@@ -75,10 +75,10 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 		}
 	}
 
-	entries := listDir(bucket, prefixDir, entryPrefixMatch)
+	emptyDir, entries := listDir(bucket, prefixDir, entryPrefixMatch)
 	// For an empty list return right here.
-	if len(entries) == 0 {
-		return 0, nil
+	if emptyDir {
+		return true, nil
 	}
 
 	// example:
@@ -90,7 +90,7 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 	entries = entries[idx:]
 	// For an empty list after search through the entries, return right here.
 	if len(entries) == 0 {
-		return 0, nil
+		return false, nil
 	}
 
 	for i, entry := range entries {
@@ -123,16 +123,16 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 			// markIsEnd is passed to this entry's treeWalk() so that treeWalker.end can be marked
 			// true at the end of the treeWalk stream.
 			markIsEnd := i == len(entries)-1 && isEnd
-			totalFound, err := doTreeWalk(ctx, bucket, pentry, prefixMatch, markerArg, recursive,
+			emptyDir, err := doTreeWalk(ctx, bucket, pentry, prefixMatch, markerArg, recursive,
 				listDir, resultCh, endWalkCh, markIsEnd)
 			if err != nil {
-				return 0, err
+				return false, err
 			}
 
 			// A nil totalFound means this is an empty directory that
 			// needs to be sent to the result channel, otherwise continue
 			// to the next entry.
-			if totalFound > 0 {
+			if !emptyDir {
 				continue
 			}
 		}
@@ -141,13 +141,13 @@ func doTreeWalk(ctx context.Context, bucket, prefixDir, entryPrefixMatch, marker
 		isEOF := ((i == len(entries)-1) && isEnd)
 		select {
 		case <-endWalkCh:
-			return 0, errWalkAbort
+			return false, errWalkAbort
 		case resultCh <- TreeWalkResult{entry: pentry, end: isEOF}:
 		}
 	}
 
 	// Everything is listed.
-	return len(entries), nil
+	return false, nil
 }
 
 // Initiate a new treeWalk in a goroutine.

cmd/tree-walk_test.go

@@ -260,7 +260,7 @@ func TestListDir(t *testing.T) {
 	}
 
 	// Should list "file1" from fsDir1.
-	entries := listDir(volume, "", "")
+	_, entries := listDir(volume, "", "")
 	if len(entries) != 2 {
 		t.Fatal("Expected the number of entries to be 2")
 	}
@@ -278,7 +278,7 @@ func TestListDir(t *testing.T) {
 	}
 
 	// Should list "file2" from fsDir2.
-	entries = listDir(volume, "", "")
+	_, entries = listDir(volume, "", "")
 	if len(entries) != 1 {
 		t.Fatal("Expected the number of entries to be 1")
 	}

cmd/typed-errors.go

@@ -50,9 +50,6 @@ var errRPCAPIVersionUnsupported = errors.New("Unsupported rpc API version")
 // errServerTimeMismatch - server times are too far apart.
 var errServerTimeMismatch = errors.New("Server times are too far apart")
 
-// errOperationTimedOut
-var errOperationTimedOut = errors.New("Operation timed out")
-
 // errInvalidBucketName - bucket name is reserved for MinIO, usually
 // returned for 'minio', '.minio.sys', buckets with capital letters.
 var errInvalidBucketName = errors.New("The specified bucket is not valid")

cmd/utils.go

@@ -324,10 +324,10 @@ func startProfiler(profilerType string) (minioProfiler, error) {
 		}
 	case madmin.ProfilerGoroutines:
 		prof.ext = "txt"
-		prof.recordBase("goroutines", 1)
+		prof.recordBase("goroutine", 1)
 		prof.stopFn = func() ([]byte, error) {
 			var buf bytes.Buffer
-			err := pprof.Lookup("goroutines").WriteTo(&buf, 1)
+			err := pprof.Lookup("goroutine").WriteTo(&buf, 1)
 			return buf.Bytes(), err
 		}
 	case madmin.ProfilerTrace:

cmd/web-handlers.go

@@ -690,6 +690,17 @@ next:
 				}
 			}
 			if authErr == errNoAuthToken {
+				// Check if object is allowed to be deleted anonymously
+				if !globalPolicySys.IsAllowed(policy.Args{
+					Action:          policy.DeleteObjectAction,
+					BucketName:      args.BucketName,
+					ConditionValues: getConditionValues(r, "", "", nil),
+					IsOwner:         false,
+					ObjectName:      objectName,
+				}) {
+					return toJSONError(ctx, errAccessDenied)
+				}
+
 				// Check if object is allowed to be deleted anonymously
 				if globalPolicySys.IsAllowed(policy.Args{
 					Action:          policy.BypassGovernanceRetentionAction,
@@ -710,16 +721,29 @@ next:
 			continue
 		}
 
-		if !globalIAMSys.IsAllowed(iampolicy.Args{
-			AccountName:     claims.AccessKey,
-			Action:          iampolicy.DeleteObjectAction,
-			BucketName:      args.BucketName,
-			ConditionValues: getConditionValues(r, "", claims.AccessKey, claims.Map()),
-			IsOwner:         owner,
-			ObjectName:      objectName,
-			Claims:          claims.Map(),
-		}) {
-			return toJSONError(ctx, errAccessDenied)
+		if authErr == errNoAuthToken {
+			// Check if object is allowed to be deleted anonymously
+			if !globalPolicySys.IsAllowed(policy.Args{
+				Action:          iampolicy.DeleteObjectAction,
+				BucketName:      args.BucketName,
+				ConditionValues: getConditionValues(r, "", "", nil),
+				IsOwner:         false,
+				ObjectName:      objectName,
+			}) {
+				return toJSONError(ctx, errAccessDenied)
+			}
+		} else {
+			if !globalIAMSys.IsAllowed(iampolicy.Args{
+				AccountName:     claims.AccessKey,
+				Action:          iampolicy.DeleteObjectAction,
+				BucketName:      args.BucketName,
+				ConditionValues: getConditionValues(r, "", claims.AccessKey, claims.Map()),
+				IsOwner:         owner,
+				ObjectName:      objectName,
+				Claims:          claims.Map(),
+			}) {
+				return toJSONError(ctx, errAccessDenied)
+			}
 		}
 
 		// For directories, list the contents recursively and remove.

cmd/xl-sets.go

@@ -1293,7 +1293,7 @@ func (s *xlSets) ReloadFormat(ctx context.Context, dryRun bool) (err error) {
 	}(storageDisks)
 
 	formats, sErrs := loadFormatXLAll(storageDisks)
-	if err = checkFormatXLValues(formats); err != nil {
+	if err = checkFormatXLValues(formats, s.drivesPerSet); err != nil {
 		return err
 	}
 
@@ -1406,7 +1406,7 @@ func (s *xlSets) HealFormat(ctx context.Context, dryRun bool) (res madmin.HealRe
 	markRootDisksAsDown(storageDisks)
 
 	formats, sErrs := loadFormatXLAll(storageDisks)
-	if err = checkFormatXLValues(formats); err != nil {
+	if err = checkFormatXLValues(formats, s.drivesPerSet); err != nil {
 		return madmin.HealResultItem{}, err
 	}
 

cmd/xl-v1-list-objects.go

@@ -25,7 +25,7 @@ import (
 // disks - used for doing disk.ListDir()
 func listDirFactory(ctx context.Context, disks ...StorageAPI) ListDirFunc {
 	// Returns sorted merged entries from all the disks.
-	listDir := func(bucket, prefixDir, prefixEntry string) (mergedEntries []string) {
+	listDir := func(bucket, prefixDir, prefixEntry string) (emptyDir bool, mergedEntries []string) {
 		for _, disk := range disks {
 			if disk == nil {
 				continue
@@ -38,6 +38,10 @@ func listDirFactory(ctx context.Context, disks ...StorageAPI) ListDirFunc {
 				continue
 			}
 
+			if len(entries) == 0 {
+				return true, nil
+			}
+
 			// Find elements in entries which are not in mergedEntries
 			for _, entry := range entries {
 				idx := sort.SearchStrings(mergedEntries, entry)
@@ -54,7 +58,7 @@ func listDirFactory(ctx context.Context, disks ...StorageAPI) ListDirFunc {
 				sort.Strings(mergedEntries)
 			}
 		}
-		return filterMatchingPrefix(mergedEntries, prefixEntry)
+		return false, filterMatchingPrefix(mergedEntries, prefixEntry)
 	}
 	return listDir
 }

cmd/xl-v1-object.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net/http"
 	"path"
+	"sync"
 
 	xhttp "github.com/minio/minio/cmd/http"
 	"github.com/minio/minio/cmd/logger"
@@ -747,48 +748,60 @@ func (xl xlObjects) deleteObject(ctx context.Context, bucket, object string, wri
 // object.
 func (xl xlObjects) doDeleteObjects(ctx context.Context, bucket string, objects []string, errs []error, writeQuorums []int, isDirs []bool) ([]error, error) {
 	var tmpObjs = make([]string, len(objects))
-	disks := xl.getDisks()
 	if bucket == minioMetaTmpBucket {
 		copy(tmpObjs, objects)
 	} else {
-		for i, object := range objects {
-			if errs[i] != nil {
+		for idx := range objects {
+			if errs[idx] != nil {
 				continue
 			}
+			tmpObjs[idx] = mustGetUUID()
 			var err error
-			tmpObjs[i] = mustGetUUID()
-			// Rename the current object while requiring write quorum, but also consider
-			// that a non found object in a given disk as a success since it already
-			// confirms that the object doesn't have a part in that disk (already removed)
-			if isDirs[i] {
-				disks, err = rename(ctx, disks, bucket, object, minioMetaTmpBucket, tmpObjs[i], true, writeQuorums[i],
+			// Rename the current object while requiring
+			// write quorum, but also consider that a non
+			// found object in a given disk as a success
+			// since it already confirms that the object
+			// doesn't have a part in that disk (already removed)
+			if isDirs[idx] {
+				_, err = rename(ctx, xl.getDisks(), bucket, objects[idx],
+					minioMetaTmpBucket, tmpObjs[idx], true, writeQuorums[idx],
 					[]error{errFileNotFound, errFileAccessDenied})
 			} else {
-				disks, err = rename(ctx, disks, bucket, object, minioMetaTmpBucket, tmpObjs[i], true, writeQuorums[i],
+				_, err = rename(ctx, xl.getDisks(), bucket, objects[idx],
+					minioMetaTmpBucket, tmpObjs[idx], true, writeQuorums[idx],
 					[]error{errFileNotFound})
 			}
 			if err != nil {
-				errs[i] = err
+				errs[idx] = err
 			}
 		}
 	}
 
+	disks := xl.getDisks()
+
 	// Initialize list of errors.
 	var opErrs = make([]error, len(disks))
 	var delObjErrs = make([][]error, len(disks))
+	var wg = sync.WaitGroup{}
 
 	// Remove objects in bulk for each disk
-	for index, disk := range disks {
-		if disk == nil {
-			opErrs[index] = errDiskNotFound
+	for i, d := range disks {
+		if d == nil {
+			opErrs[i] = errDiskNotFound
 			continue
 		}
-		delObjErrs[index], opErrs[index] = cleanupObjectsBulk(disk, minioMetaTmpBucket, tmpObjs, errs)
-		if opErrs[index] == errVolumeNotFound || opErrs[index] == errFileNotFound {
-			opErrs[index] = nil
-		}
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			delObjErrs[index], opErrs[index] = disk.DeletePrefixes(minioMetaTmpBucket, tmpObjs)
+			if opErrs[index] == errVolumeNotFound || opErrs[index] == errFileNotFound {
+				opErrs[index] = nil
+			}
+		}(i, d)
 	}
 
+	wg.Wait()
+
 	// Return errors if any during deletion
 	if err := reduceWriteQuorumErrs(ctx, opErrs, objectOpIgnoredErrs, len(disks)/2+1); err != nil {
 		return nil, err
@@ -800,9 +813,14 @@ func (xl xlObjects) doDeleteObjects(ctx context.Context, bucket string, objects
 			continue
 		}
 		listErrs := make([]error, len(disks))
+		// Iterate over disks to fetch the error
+		// of deleting of the current object
 		for i := range delObjErrs {
+			// delObjErrs[i] is not nil when disks[i] is also not nil
 			if delObjErrs[i] != nil {
-				listErrs[i] = delObjErrs[i][objIndex]
+				if delObjErrs[i][objIndex] != errFileNotFound {
+					listErrs[i] = delObjErrs[i][objIndex]
+				}
 			}
 		}
 		errs[objIndex] = reduceWriteQuorumErrs(ctx, listErrs, objectOpIgnoredErrs, writeQuorums[objIndex])
@@ -840,23 +858,17 @@ func (xl xlObjects) deleteObjects(ctx context.Context, bucket string, objects []
 		}
 	}
 
-	for i, object := range objects {
+	for i := range objects {
 		if errs[i] != nil {
 			continue
 		}
-		if isObjectDirs[i] {
-			writeQuorums[i] = len(xl.getDisks())/2 + 1
-		} else {
-			var err error
-			// Read metadata associated with the object from all disks.
-			partsMetadata, readXLErrs := readAllXLMetadata(ctx, xl.getDisks(), bucket, object)
-			// get Quorum for this object
-			_, writeQuorums[i], err = objectQuorumFromMeta(ctx, xl, partsMetadata, readXLErrs)
-			if err != nil {
-				errs[i] = toObjectErr(err, bucket, object)
-				continue
-			}
-		}
+		// Assume (N/2 + 1) quorums for all objects
+		// this is a theoretical assumption such that
+		// for delete's we do not need to honor storage
+		// class for objects which have reduced quorum
+		// storage class only needs to be honored for
+		// Read() requests alone which we already do.
+		writeQuorums[i] = len(xl.getDisks())/2 + 1
 	}
 
 	return xl.doDeleteObjects(ctx, bucket, objects, errs, writeQuorums, isObjectDirs)

docs/bucket/notifications/README.md

@@ -38,7 +38,9 @@ notify_elasticsearch  publish bucket notifications to Elasticsearch endpoints
 notify_redis          publish bucket notifications to Redis datastores
 ```
 
-> NOTE: '*' at the end of arg means its mandatory, '*' at the end of the values, means its the default value for the arg.
+> NOTE: '\*' at the end of arg means its mandatory.  
+> NOTE: '\*' at the end of the values, means its the default value for the arg.  
+> NOTE: When configured using environment variables, the `:name` can be specified using this format `MINIO_NOTIFY_WEBHOOK_ENABLE_<name>`.
 
 <a name="AMQP"></a>
 
@@ -1203,8 +1205,6 @@ kafkacat -b localhost:9092 -t bucketevents
 
 MinIO supports persistent event store. The persistent store will backup events when the webhook goes offline and replays it when the broker comes back online. The event store can be configured by setting the directory path in `queue_dir` field and the maximum limit of events in the queue_dir in `queue_limit` field. For eg, the `queue_dir` can be `/home/events` and `queue_limit` can be `1000`. By default, the `queue_limit` is set to 10000.
 
-To update the configuration, use `mc admin config get` command to get the current configuration.
-
 ```
 KEY:
 notify_webhook[:name]  publish bucket notifications to webhook endpoints
@@ -1232,7 +1232,8 @@ MINIO_NOTIFY_WEBHOOK_COMMENT      (sentence)  optionally add a comment to this s
 ```
 
 ```sh
-$ mc admin config get myminio/ notify_webhook notify_webhook:1 queue_limit="0"  endpoint="" queue_dir=""
+$ mc admin config get myminio/ notify_webhook
+notify_webhook:1 queue_limit="0"  endpoint="" queue_dir=""
 ```
 
 Use `mc admin config set` command to update the configuration for the deployment. Here the endpoint is the server listening for webhook notifications. Save the settings and restart the MinIO server for changes to take effect. Note that the endpoint needs to be live and reachable when you restart your MinIO server.

docs/kms/README.md

@@ -401,7 +401,7 @@ export MINIO_KMS_KES_ENDPOINT=https://localhost:7373
 export MINIO_KMS_KES_KEY_FILE=minio.key
 export MINIO_KMS_KES_CERT_FILE=minio.cert
 export MINIO_KMS_KES_KEY_NAME=minio-key-1
-export MINIO_KMS_KES_CAPATH=kes-tls.crt
+export MINIO_KMS_KES_CA_PATH=kes-tls.crt
 ```
 > The `MINIO_KMS_KES_CAPATH` is only required since we use self-signed certificates.
 

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

@@ -5,7 +5,7 @@ version: '3.7'
 # 9001 through 9004.
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio4
     volumes:
       - minio4-data:/export

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+    image: minio/minio:RELEASE.2020-03-09T18-26-53Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/kubernetes/minio-distributed-daemonset.yaml

@@ -30,7 +30,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+        image: minio/minio:RELEASE.2020-03-09T18-26-53Z
         # Unfortunately you must manually define each server. Perhaps autodiscovery via DNS can be implemented in the future.
         args:
         - server

docs/orchestration/kubernetes/minio-distributed-statefulset.yaml

@@ -22,7 +22,7 @@ spec:
           value: "minio"
         - name: MINIO_SECRET_KEY
           value: "minio123"
-        image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+        image: minio/minio:RELEASE.2020-03-09T18-26-53Z
         args:
         - server
         - http://minio-{0...3}.minio.default.svc.cluster.local/data

docs/orchestration/kubernetes/minio-gcs-gateway-deployment.yaml

@@ -24,7 +24,7 @@ spec:
       containers:
       - name: minio
         # Pulls the default Minio image from Docker Hub
-        image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+        image: minio/minio:RELEASE.2020-03-09T18-26-53Z
         args:
         - gateway
         - gcs

docs/orchestration/kubernetes/minio-standalone-deployment.yaml

@@ -32,7 +32,7 @@ spec:
         - name: data 
           mountPath: "/data"
         # Pulls the lastest Minio image from Docker Hub
-        image: minio/minio:RELEASE.2020-02-27T00-23-05Z
+        image: minio/minio:RELEASE.2020-03-09T18-26-53Z
         args:
         - server
         - /data

go.mod

@@ -50,7 +50,7 @@ require (
 	github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf
 	github.com/jonboulle/clockwork v0.1.0 // indirect
 	github.com/json-iterator/go v1.1.9
-	github.com/klauspost/compress v1.10.1
+	github.com/klauspost/compress v1.10.3
 	github.com/klauspost/cpuid v1.2.2
 	github.com/klauspost/pgzip v1.2.1
 	github.com/klauspost/readahead v1.3.1
@@ -67,7 +67,7 @@ require (
 	github.com/minio/hdfs/v3 v3.0.1
 	github.com/minio/highwayhash v1.0.0
 	github.com/minio/lsync v1.0.1
-	github.com/minio/minio-go/v6 v6.0.45
+	github.com/minio/minio-go/v6 v6.0.50-0.20200306231101-b882ba63d570
 	github.com/minio/parquet-go v0.0.0-20200125064549-a1e49702e174
 	github.com/minio/sha256-simd v0.1.1
 	github.com/minio/simdjson-go v0.1.5-0.20200303142138-b17fe061ea37

go.sum

@@ -224,6 +224,8 @@ github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
 github.com/klauspost/compress v1.9.4/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
 github.com/klauspost/compress v1.10.1 h1:a/QY0o9S6wCi0XhxaMX/QmusicNUqCqFugR6WKPOSoQ=
 github.com/klauspost/compress v1.10.1/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.10.3 h1:OP96hzwJVBIHYU52pVTI6CczrxPvrGfgqF9N5eTO0Q8=
+github.com/klauspost/compress v1.10.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/cpuid v1.2.2 h1:1xAgYebNnsb9LKCdLOvFWtAxGU/33mjJtyOVbmUa0Us=
 github.com/klauspost/cpuid v1.2.2/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/pgzip v1.2.1 h1:oIPZROsWuPHpOdMVWLuJZXwgjhrW8r1yEX8UqMyeNHM=
@@ -274,12 +276,12 @@ github.com/minio/lsync v1.0.1 h1:AVvILxA976xc27hstd1oR+X9PQG0sPSom1MNb1ImfUs=
 github.com/minio/lsync v1.0.1/go.mod h1:tCFzfo0dlvdGl70IT4IAK/5Wtgb0/BrTmo/jE8pArKA=
 github.com/minio/minio-go/v6 v6.0.45 h1:aY4NI/DOgSbZiwGN3fEF4NAkC9An4bhaIWuJrQrRYew=
 github.com/minio/minio-go/v6 v6.0.45/go.mod h1:qD0lajrGW49lKZLtXKtCB4X/qkMf0a5tBvN2PaZg7Gg=
+github.com/minio/minio-go/v6 v6.0.50-0.20200306231101-b882ba63d570 h1:GLTZoRC6rhCTucnkJAQ63LhMU2S4CM71MRc9gfX7ohE=
+github.com/minio/minio-go/v6 v6.0.50-0.20200306231101-b882ba63d570/go.mod h1:qD0lajrGW49lKZLtXKtCB4X/qkMf0a5tBvN2PaZg7Gg=
 github.com/minio/parquet-go v0.0.0-20200125064549-a1e49702e174 h1:WYFHZIJ5LTWd4C3CW26jguaBLLDdX7l1/Xa3QSKGkIc=
 github.com/minio/parquet-go v0.0.0-20200125064549-a1e49702e174/go.mod h1:PXYM9yI2l0YPmxHUXe6mFTmkQcyaVasDshAPTbGpDoo=
 github.com/minio/sha256-simd v0.1.1 h1:5QHSlgo3nt5yKOJrC7W8w7X+NFl8cMPZm96iu8kKUJU=
 github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
-github.com/minio/simdjson-go v0.1.4 h1:aG95y9ASXS2PRhcL7NOH9SIioNEJ5vMgvONFAon0snE=
-github.com/minio/simdjson-go v0.1.4/go.mod h1:oKURrZZEBtqObgJrSjN1Ln2n9MJj2icuBTkeJzZnvSI=
 github.com/minio/simdjson-go v0.1.5-0.20200303142138-b17fe061ea37 h1:pDeao6M5AEd8hwTtGmE0pVKomlL56JFRa5SiXDZAuJE=
 github.com/minio/simdjson-go v0.1.5-0.20200303142138-b17fe061ea37/go.mod h1:oKURrZZEBtqObgJrSjN1Ln2n9MJj2icuBTkeJzZnvSI=
 github.com/minio/sio v0.2.0 h1:NCRCFLx0r5pRbXf65LVNjxbCGZgNQvNFQkgX3XF4BoA=

pkg/bucket/policy/condition/jwt.go

@@ -0,0 +1,75 @@
+/*
+ * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package condition
+
+// JWT claims supported substitutions.
+// https://www.iana.org/assignments/jwt/jwt.xhtml#claims
+const (
+	// JWTSub - JWT subject claim substitution.
+	JWTSub Key = "jwt:sub"
+
+	// JWTIss issuer claim substitution.
+	JWTIss Key = "jwt:iss"
+
+	// JWTAud audience claim substitution.
+	JWTAud Key = "jwt:aud"
+
+	// JWTJti JWT unique identifier claim substitution.
+	JWTJti Key = "jwt:jti"
+
+	JWTName         Key = "jwt:name"
+	JWTGivenName    Key = "jwt:given_name"
+	JWTFamilyName   Key = "jwt:family_name"
+	JWTMiddleName   Key = "jwt:middle_name"
+	JWTNickName     Key = "jwt:nickname"
+	JWTPrefUsername Key = "jwt:preferred_username"
+	JWTProfile      Key = "jwt:profile"
+	JWTPicture      Key = "jwt:picture"
+	JWTWebsite      Key = "jwt:website"
+	JWTEmail        Key = "jwt:email"
+	JWTGender       Key = "jwt:gender"
+	JWTBirthdate    Key = "jwt:birthdate"
+	JWTPhoneNumber  Key = "jwt:phone_number"
+	JWTAddress      Key = "jwt:address"
+	JWTScope        Key = "jwt:scope"
+	JWTClientID     Key = "jwt:client_id"
+)
+
+// JWTKeys - Supported JWT keys, non-exhaustive list please
+// expand as new claims are standardized.
+var JWTKeys = []Key{
+	JWTSub,
+	JWTIss,
+	JWTAud,
+	JWTJti,
+	JWTName,
+	JWTGivenName,
+	JWTFamilyName,
+	JWTMiddleName,
+	JWTNickName,
+	JWTPrefUsername,
+	JWTProfile,
+	JWTPicture,
+	JWTWebsite,
+	JWTEmail,
+	JWTGender,
+	JWTBirthdate,
+	JWTPhoneNumber,
+	JWTAddress,
+	JWTScope,
+	JWTClientID,
+}

pkg/bucket/policy/condition/key.go

@@ -85,22 +85,10 @@ const (
 
 	// AWSUsername - user friendly name, in MinIO this value is same as your user Access Key.
 	AWSUsername Key = "aws:username"
-
-	// JWTSub - JWT subject claim substitution.
-	JWTSub Key = "jwt:sub"
-
-	// JWTIss issuer claim substitution.
-	JWTIss Key = "jwt:iss"
-
-	// JWTAud audience claim substitution.
-	JWTAud Key = "jwt:aud"
-
-	// JWTJti JWT unique identifier claim substitution.
-	JWTJti Key = "jwt:jti"
 )
 
 // AllSupportedKeys - is list of all all supported keys.
-var AllSupportedKeys = []Key{
+var AllSupportedKeys = append([]Key{
 	S3XAmzCopySource,
 	S3XAmzServerSideEncryption,
 	S3XAmzServerSideEncryptionCustomerAlgorithm,
@@ -119,15 +107,11 @@ var AllSupportedKeys = []Key{
 	AWSPrincipalType,
 	AWSUserID,
 	AWSUsername,
-	JWTSub,
-	JWTIss,
-	JWTAud,
-	JWTJti,
 	// Add new supported condition keys.
-}
+}, JWTKeys...)
 
 // CommonKeys - is list of all common condition keys.
-var CommonKeys = []Key{
+var CommonKeys = append([]Key{
 	AWSReferer,
 	AWSSourceIP,
 	AWSUserAgent,
@@ -137,11 +121,7 @@ var CommonKeys = []Key{
 	AWSPrincipalType,
 	AWSUserID,
 	AWSUsername,
-	JWTSub,
-	JWTIss,
-	JWTAud,
-	JWTJti,
-}
+}, JWTKeys...)
 
 func substFuncFromValues(values map[string][]string) func(string) string {
 	return func(v string) string {

pkg/disk/directio_darwin.go

@@ -1,5 +1,5 @@
 /*
- * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ * Minio Cloud Storage, (C) 2019-2020 Minio, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,3 +34,8 @@ func DisableDirectIO(f *os.File) error {
 	_, err := unix.FcntlInt(fd, unix.F_NOCACHE, 0)
 	return err
 }
+
+// AlignedBlock - pass through to directio implementation.
+func AlignedBlock(BlockSize int) []byte {
+	return directio.AlignedBlock(BlockSize)
+}

pkg/disk/directio_unix.go

@@ -1,7 +1,7 @@
 // +build linux netbsd freebsd
 
 /*
- * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ * Minio Cloud Storage, (C) 2019-2020 Minio, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -42,3 +42,8 @@ func DisableDirectIO(f *os.File) error {
 	_, err = unix.FcntlInt(fd, unix.F_SETFL, flag)
 	return err
 }
+
+// AlignedBlock - pass through to directio implementation.
+func AlignedBlock(BlockSize int) []byte {
+	return directio.AlignedBlock(BlockSize)
+}

pkg/disk/directio_unsupported.go

@@ -1,7 +1,7 @@
 // +build !linux,!netbsd,!freebsd,!darwin
 
 /*
- * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ * Minio Cloud Storage, (C) 2019-2020 Minio, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,9 +22,30 @@ import (
 	"os"
 )
 
-// OpenBSD and Windows not supported.
-// On OpenBSD O_DIRECT is not supported
-// On Windows there is no documentation on disabling O_DIRECT
+// OpenBSD, Windows, and illumos do not support O_DIRECT.
+// On Windows there is no documentation on disabling O_DIRECT.
+// For these systems we do not attempt to build the 'directio' dependency since
+// the O_DIRECT symbol may not be exposed resulting in a failed build.
+//
+//
+// On illumos an explicit O_DIRECT flag is not necessary for two primary
+// reasons. Note that ZFS is effectively the default filesystem on illumos
+// systems.
+//
+// One benefit of using DirectIO on Linux is that the page cache will not be
+// polluted with single-access data. The ZFS read cache (ARC) is scan-resistant
+// so there is no risk of polluting the entire cache with data accessed once.
+// Another goal of DirectIO is to minimize the mutation of data by the kernel
+// before issuing IO to underlying devices. ZFS users often enable features like
+// compression and checksumming which currently necessitates mutating data in
+// the kernel.
+//
+// DirectIO semantics for a filesystem like ZFS would be quite different than
+// the semantics on filesystems like XFS, and these semantics are not
+// implemented at this time.
+// For more information on why typical DirectIO semantics do not apply to ZFS
+// see this ZFS-on-Linux commit message:
+// https://github.com/openzfs/zfs/commit/a584ef26053065f486d46a7335bea222cb03eeea
 
 func OpenFileDirectIO(filePath string, flag int, perm os.FileMode) (*os.File, error) {
 	return os.OpenFile(filePath, flag, perm)
@@ -33,3 +54,9 @@ func OpenFileDirectIO(filePath string, flag int, perm os.FileMode) (*os.File, er
 func DisableDirectIO(f *os.File) error {
 	return nil
 }
+
+// AlignedBlock simply returns an unaligned buffer for systems that do not
+// support DirectIO.
+func AlignedBlock(BlockSize int) []byte {
+	return make([]byte, BlockSize)
+}

pkg/disk/disk.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2018-2019 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2018-2020 MinIO, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,8 +18,6 @@ package disk
 
 import (
 	"os"
-
-	"github.com/ncw/directio"
 )
 
 // Info stat fs struct is container which holds following values
@@ -73,7 +71,7 @@ func doPerfMeasure(fsPath string, size int64) (writeSpeed, readSpeed float64, er
 	}
 
 	// Fetch aligned buf for direct-io
-	buf := directio.AlignedBlock(speedTestBlockSize)
+	buf := AlignedBlock(speedTestBlockSize)
 
 	writeSpeed, err = speedTestWrite(w, buf, size)
 	w.Close()

pkg/iam/policy/constants.go

@@ -72,7 +72,7 @@ var AdminDiagnostics = Policy{
 		{
 			SID:       policy.ID(""),
 			Effect:    policy.Allow,
-			Actions:   NewActionSet(PerfInfoAdminAction, ProfilingAdminAction, TraceAdminAction, ConsoleLogAdminAction, ServerInfoAdminAction, ServerHardwareInfoAdminAction),
+			Actions:   NewActionSet(PerfInfoAdminAction, ProfilingAdminAction, TraceAdminAction, ConsoleLogAdminAction, ServerInfoAdminAction, ServerHardwareInfoAdminAction, TopLocksAdminAction),
 			Resources: NewResourceSet(NewResource("*", "")),
 		},
 	},

pkg/madmin/heal-commands.go

@@ -274,6 +274,7 @@ func (adm *AdminClient) Heal(bucket, prefix string, healOpts HealOpts,
 type BgHealState struct {
 	ScannedItemsCount int64
 	LastHealActivity  time.Time
+	NextHealRound     time.Time
 }
 
 // BackgroundHealStatus returns the background heal status of the

pkg/s3select/errors.go

@@ -71,6 +71,24 @@ func errInvalidCompressionFormat(err error) *s3Error {
 	}
 }
 
+func errInvalidBZIP2CompressionFormat(err error) *s3Error {
+	return &s3Error{
+		code:       "InvalidCompressionFormat",
+		message:    "BZIP2 is not applicable to the queried object. Please correct the request and try again.",
+		statusCode: 400,
+		cause:      err,
+	}
+}
+
+func errInvalidGZIPCompressionFormat(err error) *s3Error {
+	return &s3Error{
+		code:       "InvalidCompressionFormat",
+		message:    "GZIP is not applicable to the queried object. Please correct the request and try again.",
+		statusCode: 400,
+		cause:      err,
+	}
+}
+
 func errInvalidDataSource(err error) *s3Error {
 	return &s3Error{
 		code:       "InvalidDataSource",

pkg/s3select/progress.go

@@ -100,7 +100,11 @@ func newProgressReader(rc io.ReadCloser, compType CompressionType) (*progressRea
 	case noneType:
 		r = scannedReader
 	case gzipType:
-		if r, err = gzip.NewReader(scannedReader); err != nil {
+		r, err = gzip.NewReader(scannedReader)
+		if err != nil {
+			if errors.Is(err, gzip.ErrHeader) || errors.Is(err, gzip.ErrChecksum) {
+				return nil, errInvalidGZIPCompressionFormat(err)
+			}
 			return nil, errTruncatedInput(err)
 		}
 	case bzip2Type:

pkg/s3select/select.go

@@ -19,7 +19,9 @@ package s3select
 import (
 	"bufio"
 	"bytes"
+	"compress/bzip2"
 	"encoding/xml"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -302,9 +304,12 @@ func (s3Select *S3Select) Open(getReader func(offset, length int64) (io.ReadClos
 		s3Select.recordReader, err = csv.NewReader(s3Select.progressReader, &s3Select.Input.CSVArgs)
 		if err != nil {
 			rc.Close()
+			var stErr bzip2.StructuralError
+			if errors.As(err, &stErr) {
+				return errInvalidBZIP2CompressionFormat(err)
+			}
 			return err
 		}
-
 		return nil
 	case jsonFormat:
 		rc, err := getReader(0, -1)
@@ -387,7 +392,6 @@ func (s3Select *S3Select) Evaluate(w http.ResponseWriter) {
 	}
 	writer := newMessageWriter(w, getProgressFunc)
 
-	var inputRecord sql.Record
 	var outputQueue []sql.Record
 
 	// Create queue based on the type.
@@ -428,6 +432,7 @@ func (s3Select *S3Select) Evaluate(w http.ResponseWriter) {
 	}
 
 	var rec sql.Record
+OuterLoop:
 	for {
 		if s3Select.statement.LimitReached() {
 			if !sendRecord() {
@@ -464,47 +469,50 @@ func (s3Select *S3Select) Evaluate(w http.ResponseWriter) {
 			break
 		}
 
-		if inputRecord, err = s3Select.statement.EvalFrom(s3Select.Input.format, rec); err != nil {
+		var inputRecords []*sql.Record
+		if inputRecords, err = s3Select.statement.EvalFrom(s3Select.Input.format, rec); err != nil {
 			break
 		}
 
-		if s3Select.statement.IsAggregated() {
-			if err = s3Select.statement.AggregateRow(inputRecord); err != nil {
-				break
-			}
-		} else {
-			var outputRecord sql.Record
-			// We will attempt to reuse the records in the table.
-			// The type of these should not change.
-			// The queue should always have at least one entry left for this to work.
-			outputQueue = outputQueue[:len(outputQueue)+1]
-			if t := outputQueue[len(outputQueue)-1]; t != nil {
-				// If the output record is already set, we reuse it.
-				outputRecord = t
-				outputRecord.Reset()
-			} else {
-				// Create new one
-				outputRecord = s3Select.outputRecord()
-				outputQueue[len(outputQueue)-1] = outputRecord
-			}
-			outputRecord, err = s3Select.statement.Eval(inputRecord, outputRecord)
-			if outputRecord == nil || err != nil {
-				// This should not be written.
-				// Remove it from the queue.
-				outputQueue = outputQueue[:len(outputQueue)-1]
-				if err != nil {
-					break
+		for _, inputRecord := range inputRecords {
+			if s3Select.statement.IsAggregated() {
+				if err = s3Select.statement.AggregateRow(*inputRecord); err != nil {
+					break OuterLoop
+				}
+			} else {
+				var outputRecord sql.Record
+				// We will attempt to reuse the records in the table.
+				// The type of these should not change.
+				// The queue should always have at least one entry left for this to work.
+				outputQueue = outputQueue[:len(outputQueue)+1]
+				if t := outputQueue[len(outputQueue)-1]; t != nil {
+					// If the output record is already set, we reuse it.
+					outputRecord = t
+					outputRecord.Reset()
+				} else {
+					// Create new one
+					outputRecord = s3Select.outputRecord()
+					outputQueue[len(outputQueue)-1] = outputRecord
+				}
+				outputRecord, err = s3Select.statement.Eval(*inputRecord, outputRecord)
+				if outputRecord == nil || err != nil {
+					// This should not be written.
+					// Remove it from the queue.
+					outputQueue = outputQueue[:len(outputQueue)-1]
+					if err != nil {
+						break OuterLoop
+					}
+					continue
 				}
-				continue
-			}
 
-			outputQueue[len(outputQueue)-1] = outputRecord
-			if len(outputQueue) < cap(outputQueue) {
-				continue
-			}
+				outputQueue[len(outputQueue)-1] = outputRecord
+				if len(outputQueue) < cap(outputQueue) {
+					continue
+				}
 
-			if !sendRecord() {
-				break
+				if !sendRecord() {
+					break OuterLoop
+				}
 			}
 		}
 	}

pkg/s3select/select_test.go

@@ -260,6 +260,68 @@ func TestJSONQueries(t *testing.T) {
 </SelectObjectContentRequest>`),
 			wantResult: `"[""foo"",""bar"",""whatever""]"`,
 		},
+		{
+			name:  "document",
+			query: "",
+			requestXML: []byte(`
+<?xml version="1.0" encoding="UTF-8"?>
+<SelectObjectContentRequest>
+    <Expression>select * from s3object[*].elements[*] s where s.element_type = '__elem__merfu'</Expression>
+    <ExpressionType>SQL</ExpressionType>
+    <InputSerialization>
+        <CompressionType>NONE</CompressionType>
+        <JSON>
+            <Type>DOCUMENT</Type>
+        </JSON>
+    </InputSerialization>
+    <OutputSerialization>
+        <JSON>
+        </JSON>
+    </OutputSerialization>
+    <RequestProgress>
+        <Enabled>FALSE</Enabled>
+    </RequestProgress>
+</SelectObjectContentRequest>`),
+			withJSON: `
+{
+  "name": "small_pdf1.pdf",
+  "lume_id": "9507193e-572d-4f95-bcf1-e9226d96be65",
+  "elements": [
+    {
+      "element_type": "__elem__image",
+      "element_id": "859d09c4-7cf1-4a37-9674-3a7de8b56abc",
+      "attributes": {
+        "__attr__image_dpi": 300,
+        "__attr__image_size": [
+          2550,
+          3299
+        ],
+        "__attr__image_index": 1,
+        "__attr__image_format": "JPEG",
+        "__attr__file_extension": "jpg",
+        "__attr__data": null
+      }
+    },
+    {
+      "element_type": "__elem__merfu",
+      "element_id": "d868aefe-ef9a-4be2-b9b2-c9fd89cc43eb",
+      "attributes": {
+        "__attr__image_dpi": 300,
+        "__attr__image_size": [
+          2550,
+          3299
+        ],
+        "__attr__image_index": 2,
+        "__attr__image_format": "JPEG",
+        "__attr__file_extension": "jpg",
+        "__attr__data": null
+      }
+    }
+  ],
+  "data": "asdascasdc1234e123erdasdas"
+}`,
+			wantResult: `{"element_type":"__elem__merfu","element_id":"d868aefe-ef9a-4be2-b9b2-c9fd89cc43eb","attributes":{"__attr__image_dpi":300,"__attr__image_size":[2550,3299],"__attr__image_index":2,"__attr__image_format":"JPEG","__attr__file_extension":"jpg","__attr__data":null}}`,
+		},
 	}
 
 	defRequest := `<?xml version="1.0" encoding="UTF-8"?>

pkg/s3select/sql/statement.go

@@ -140,9 +140,9 @@ func parseLimit(v *LitValue) (int64, error) {
 
 // EvalFrom evaluates the From clause on the input record. It only
 // applies to JSON input data format (currently).
-func (e *SelectStatement) EvalFrom(format string, input Record) (Record, error) {
+func (e *SelectStatement) EvalFrom(format string, input Record) ([]*Record, error) {
 	if !e.selectAST.From.HasKeypath() {
-		return input, nil
+		return []*Record{&input}, nil
 	}
 	_, rawVal := input.Raw()
 
@@ -160,6 +160,18 @@ func (e *SelectStatement) EvalFrom(format string, input Record) (Record, error)
 		switch v := txedRec.(type) {
 		case jstream.KVS:
 			kvs = v
+
+		case []interface{}:
+			recs := make([]*Record, len(v))
+			for i, val := range v {
+				tmpRec := input.Clone(nil)
+				if err = tmpRec.Replace(val); err != nil {
+					return nil, err
+				}
+				recs[i] = &tmpRec
+			}
+			return recs, nil
+
 		default:
 			kvs = jstream.KVS{jstream.KV{Key: "_1", Value: v}}
 		}
@@ -168,7 +180,7 @@ func (e *SelectStatement) EvalFrom(format string, input Record) (Record, error)
 			return nil, err
 		}
 
-		return input, nil
+		return []*Record{&input}, nil
 	case simdjson.Object:
 		txedRec, _, err := jsonpathEval(e.selectAST.From.Table.PathExpr[1:], rec)
 		if err != nil {
@@ -181,6 +193,18 @@ func (e *SelectStatement) EvalFrom(format string, input Record) (Record, error)
 			if err != nil {
 				return nil, err
 			}
+
+		case []interface{}:
+			recs := make([]*Record, len(v))
+			for i, val := range v {
+				tmpRec := input.Clone(nil)
+				if err = tmpRec.Replace(val); err != nil {
+					return nil, err
+				}
+				recs[i] = &tmpRec
+			}
+			return recs, nil
+
 		default:
 			input.Reset()
 			input, err = input.Set("_1", &Value{value: v})
@@ -188,7 +212,7 @@ func (e *SelectStatement) EvalFrom(format string, input Record) (Record, error)
 				return nil, err
 			}
 		}
-		return input, nil
+		return []*Record{&input}, nil
 	}
 	return nil, errDataSource(errors.New("unexpected non JSON input"))
 }