add validation logs for configured Logger/Audit HTTP targets (#10274)

extra logs in-case of misconfiguration of audit/logger targets

cmd/bucket-listobjects-handlers.go

@@ -24,12 +24,36 @@ import (
 	"strings"
 
 	"github.com/gorilla/mux"
-	"github.com/minio/minio/cmd/crypto"
 	"github.com/minio/minio/cmd/logger"
 
 	"github.com/minio/minio/pkg/bucket/policy"
+	"github.com/minio/minio/pkg/sync/errgroup"
 )
 
+func concurrentDecryptETag(ctx context.Context, objects []ObjectInfo) {
+	inParallel := func(objects []ObjectInfo) {
+		g := errgroup.WithNErrs(len(objects))
+		for index := range objects {
+			index := index
+			g.Go(func() error {
+				objects[index].ETag = objects[index].GetActualETag(nil)
+				objects[index].Size, _ = objects[index].GetActualSize()
+				return nil
+			}, index)
+		}
+		g.Wait()
+	}
+	const maxConcurrent = 500
+	for {
+		if len(objects) < maxConcurrent {
+			inParallel(objects)
+			return
+		}
+		inParallel(objects[:maxConcurrent])
+		objects = objects[maxConcurrent:]
+	}
+}
+
 // Validate all the ListObjects query arguments, returns an APIErrorCode
 // if one of the args do not meet the required conditions.
 // Special conditions required by MinIO server are as below
@@ -89,6 +113,10 @@ func (api objectAPIHandlers) ListObjectVersionsHandler(w http.ResponseWriter, r
 		return
 	}
 
+	if proxyRequestByBucket(ctx, w, r, bucket) {
+		return
+	}
+
 	listObjectVersions := objectAPI.ListObjectVersions
 
 	// Inititate a list object versions operation based on the input params.
@@ -100,16 +128,7 @@ func (api objectAPIHandlers) ListObjectVersionsHandler(w http.ResponseWriter, r
 		return
 	}
 
-	for i := range listObjectVersionsInfo.Objects {
-		if crypto.IsEncrypted(listObjectVersionsInfo.Objects[i].UserDefined) {
-			listObjectVersionsInfo.Objects[i].ETag = getDecryptedETag(r.Header, listObjectVersionsInfo.Objects[i], false)
-		}
-		listObjectVersionsInfo.Objects[i].Size, err = listObjectVersionsInfo.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectVersionsInfo.Objects)
 
 	response := generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delimiter, encodingType, maxkeys, listObjectVersionsInfo)
 
@@ -178,16 +197,7 @@ func (api objectAPIHandlers) ListObjectsV2MHandler(w http.ResponseWriter, r *htt
 		return
 	}
 
-	for i := range listObjectsV2Info.Objects {
-		if crypto.IsEncrypted(listObjectsV2Info.Objects[i].UserDefined) {
-			listObjectsV2Info.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsV2Info.Objects[i], false)
-		}
-		listObjectsV2Info.Objects[i].Size, err = listObjectsV2Info.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsV2Info.Objects)
 
 	// The next continuation token has id@node_index format to optimize paginated listing
 	nextContinuationToken := listObjectsV2Info.NextContinuationToken
@@ -264,16 +274,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 		return
 	}
 
-	for i := range listObjectsV2Info.Objects {
-		if crypto.IsEncrypted(listObjectsV2Info.Objects[i].UserDefined) {
-			listObjectsV2Info.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsV2Info.Objects[i], false)
-		}
-		listObjectsV2Info.Objects[i].Size, err = listObjectsV2Info.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsV2Info.Objects)
 
 	// The next continuation token has id@node_index format to optimize paginated listing
 	nextContinuationToken := listObjectsV2Info.NextContinuationToken
@@ -396,16 +397,7 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 		return
 	}
 
-	for i := range listObjectsInfo.Objects {
-		if crypto.IsEncrypted(listObjectsInfo.Objects[i].UserDefined) {
-			listObjectsInfo.Objects[i].ETag = getDecryptedETag(r.Header, listObjectsInfo.Objects[i], false)
-		}
-		listObjectsInfo.Objects[i].Size, err = listObjectsInfo.Objects[i].GetActualSize()
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
-			return
-		}
-	}
+	concurrentDecryptETag(ctx, listObjectsInfo.Objects)
 
 	response := generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType, maxKeys, listObjectsInfo)
 

cmd/config-current.go

@@ -455,28 +455,32 @@ func lookupConfigs(s config.Config, setDriveCount int) {
 	for _, l := range loggerCfg.HTTP {
 		if l.Enabled {
 			// Enable http logging
-			logger.AddTarget(
+			if err = logger.AddTarget(
 				http.New(http.WithEndpoint(l.Endpoint),
 					http.WithAuthToken(l.AuthToken),
 					http.WithUserAgent(loggerUserAgent),
 					http.WithLogKind(string(logger.All)),
 					http.WithTransport(NewGatewayHTTPTransport()),
 				),
-			)
+			); err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize console HTTP target: %w", err))
+			}
 		}
 	}
 
 	for _, l := range loggerCfg.Audit {
 		if l.Enabled {
 			// Enable http audit logging
-			logger.AddAuditTarget(
+			if err = logger.AddAuditTarget(
 				http.New(http.WithEndpoint(l.Endpoint),
 					http.WithAuthToken(l.AuthToken),
 					http.WithUserAgent(loggerUserAgent),
 					http.WithLogKind(string(logger.All)),
 					http.WithTransport(NewGatewayHTTPTransport()),
 				),
-			)
+			); err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Unable to initialize audit HTTP target: %w", err))
+			}
 		}
 	}
 

cmd/config/storageclass/storage-class.go

@@ -226,7 +226,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	cfg.RRS.Parity = defaultRRSParity
 
 	if err = config.CheckValidKeys(config.StorageClassSubSys, kvs, DefaultKVS); err != nil {
-		return cfg, err
+		return Config{}, err
 	}
 
 	ssc := env.Get(StandardEnv, kvs.Get(ClassStandard))
@@ -235,7 +235,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	if ssc != "" {
 		cfg.Standard, err = parseStorageClass(ssc)
 		if err != nil {
-			return cfg, err
+			return Config{}, err
 		}
 	}
 	if cfg.Standard.Parity == 0 {
@@ -245,7 +245,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	if rrsc != "" {
 		cfg.RRS, err = parseStorageClass(rrsc)
 		if err != nil {
-			return cfg, err
+			return Config{}, err
 		}
 	}
 	if cfg.RRS.Parity == 0 {
@@ -255,7 +255,7 @@ func LookupConfig(kvs config.KVS, drivesPerSet int) (cfg Config, err error) {
 	// Validation is done after parsing both the storage classes. This is needed because we need one
 	// storage class value to deduce the correct value of the other storage class.
 	if err = validateParity(cfg.Standard.Parity, cfg.RRS.Parity, drivesPerSet); err != nil {
-		return cfg, err
+		return Config{}, err
 	}
 
 	return cfg, nil

cmd/consolelogger.go

@@ -117,6 +117,11 @@ func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan interface{}, doneCh <-chan
 	sys.pubsub.Subscribe(subCh, doneCh, filter)
 }
 
+// Validate if HTTPConsoleLoggerSys is valid, always returns nil right now
+func (sys *HTTPConsoleLoggerSys) Validate() error {
+	return nil
+}
+
 // Send log message 'e' to console and publish to console
 // log pubsub system
 func (sys *HTTPConsoleLoggerSys) Send(e interface{}, logKind string) error {

cmd/crypto/header.go

@@ -18,10 +18,10 @@ import (
 	"bytes"
 	"crypto/md5"
 	"encoding/base64"
-	"encoding/json"
 	"net/http"
 	"strings"
 
+	jsoniter "github.com/json-iterator/go"
 	xhttp "github.com/minio/minio/cmd/http"
 )
 
@@ -148,6 +148,7 @@ func (s3KMS) ParseHTTP(h http.Header) (string, interface{}, error) {
 	contextStr, ok := h[SSEKmsContext]
 	if ok {
 		var context map[string]interface{}
+		var json = jsoniter.ConfigCompatibleWithStandardLibrary
 		if err := json.Unmarshal([]byte(contextStr[0]), &context); err != nil {
 			return "", nil, err
 		}

cmd/crypto/kes.go

@@ -18,7 +18,6 @@ import (
 	"bytes"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -30,10 +29,13 @@ import (
 	"strings"
 	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	xhttp "github.com/minio/minio/cmd/http"
 	xnet "github.com/minio/minio/pkg/net"
 )
 
+var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
 // ErrKESKeyExists is the error returned a KES server
 // when a master key does exist.
 var ErrKESKeyExists = NewKESError(http.StatusBadRequest, "key does already exist")

cmd/data-update-tracker_test.go

@@ -42,6 +42,10 @@ type testingLogger struct {
 	t  testLoggerI
 }
 
+func (t *testingLogger) Validate() error {
+	return nil
+}
+
 func (t *testingLogger) Send(entry interface{}, errKind string) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()

cmd/disk-cache.go

@@ -326,7 +326,9 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 			// avoid cache overwrite if another background routine filled cache
 			if err != nil || oi.ETag != bReader.ObjInfo.ETag {
 				// use a new context to avoid locker prematurely timing out operation when the GetObjectNInfo returns.
-				dcache.Put(context.Background(), bucket, object, bReader, bReader.ObjInfo.Size, rs, ObjectOptions{UserDefined: getMetadata(bReader.ObjInfo)}, false)
+				dcache.Put(GlobalContext, bucket, object, bReader, bReader.ObjInfo.Size, rs, ObjectOptions{
+					UserDefined: getMetadata(bReader.ObjInfo),
+				}, false)
 				return
 			}
 		}()
@@ -337,7 +339,11 @@ func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 	pipeReader, pipeWriter := io.Pipe()
 	teeReader := io.TeeReader(bkReader, pipeWriter)
 	go func() {
-		putErr := dcache.Put(ctx, bucket, object, io.LimitReader(pipeReader, bkReader.ObjInfo.Size), bkReader.ObjInfo.Size, nil, ObjectOptions{UserDefined: getMetadata(bkReader.ObjInfo)}, false)
+		putErr := dcache.Put(ctx, bucket, object,
+			io.LimitReader(pipeReader, bkReader.ObjInfo.Size),
+			bkReader.ObjInfo.Size, nil, ObjectOptions{
+				UserDefined: getMetadata(bkReader.ObjInfo),
+			}, false)
 		// close the write end of the pipe, so the error gets
 		// propagated to getObjReader
 		pipeWriter.CloseWithError(putErr)

cmd/encryption-v1.go

@@ -605,12 +605,14 @@ func getDecryptedETag(headers http.Header, objInfo ObjectInfo, copySource bool)
 	if crypto.IsMultiPart(objInfo.UserDefined) {
 		return objInfo.ETag
 	}
+
 	if crypto.SSECopy.IsRequested(headers) {
 		key, err = crypto.SSECopy.ParseHTTP(headers)
 		if err != nil {
 			return objInfo.ETag
 		}
 	}
+
 	// As per AWS S3 Spec, ETag for SSE-C encrypted objects need not be MD5Sum of the data.
 	// Since server side copy with same source and dest just replaces the ETag, we save
 	// encrypted content MD5Sum as ETag for both SSE-C and SSE-S3, we standardize the ETag

cmd/erasure-zones.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"math/rand"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -2063,6 +2064,8 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 		}
 	}
 
+	reqInfo := (&logger.ReqInfo{}).AppendTags("maintenance", strconv.FormatBool(opts.Maintenance))
+
 	for zoneIdx := range erasureSetUpCount {
 		parityDrives := globalStorageClass.GetParityForSC(storageclass.STANDARD)
 		diskCount := z.zones[zoneIdx].drivesPerSet
@@ -2076,8 +2079,9 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 		}
 		for setIdx := range erasureSetUpCount[zoneIdx] {
 			if erasureSetUpCount[zoneIdx][setIdx] < writeQuorum {
-				logger.LogIf(ctx, fmt.Errorf("Write quorum lost on zone: %d, set: %d, expected write quorum: %d",
-					zoneIdx, setIdx, writeQuorum))
+				logger.LogIf(logger.SetReqInfo(ctx, reqInfo),
+					fmt.Errorf("Write quorum may be lost on zone: %d, set: %d, expected write quorum: %d",
+						zoneIdx, setIdx, writeQuorum))
 				return HealthResult{
 					Healthy:     false,
 					ZoneID:      zoneIdx,
@@ -2101,14 +2105,14 @@ func (z *erasureZones) Health(ctx context.Context, opts HealthOptions) HealthRes
 	// is not taken down for maintenance
 	aggHealStateResult, err := getAggregatedBackgroundHealState(ctx, true)
 	if err != nil {
-		logger.LogIf(ctx, fmt.Errorf("Unable to verify global heal status: %w", err))
+		logger.LogIf(logger.SetReqInfo(ctx, reqInfo), fmt.Errorf("Unable to verify global heal status: %w", err))
 		return HealthResult{
 			Healthy: false,
 		}
 	}
 
 	if len(aggHealStateResult.HealDisks) > 0 {
-		logger.LogIf(ctx, fmt.Errorf("Total drives to be healed %d", len(aggHealStateResult.HealDisks)))
+		logger.LogIf(logger.SetReqInfo(ctx, reqInfo), fmt.Errorf("Total drives to be healed %d", len(aggHealStateResult.HealDisks)))
 	}
 
 	healthy := len(aggHealStateResult.HealDisks) == 0

cmd/gateway-main.go

@@ -180,6 +180,11 @@ func StartGateway(ctx *cli.Context, gw Gateway) {
 	globalRootCAs, err = certs.GetRootCAs(globalCertsCADir.Get())
 	logger.FatalIf(err, "Failed to read root CAs (%v)", err)
 
+	// Add the global public crts as part of global root CAs
+	for _, publicCrt := range globalPublicCerts {
+		globalRootCAs.AddCert(publicCrt)
+	}
+
 	// Register root CAs for remote ENVs
 	env.RegisterGlobalCAs(globalRootCAs)
 

cmd/iam-object-store.go

@@ -151,7 +151,7 @@ func (iamOS *IAMObjectStore) migrateUsersConfigToV1(ctx context.Context, isSTS b
 		cred.AccessKey = user
 		u := newUserIdentity(cred)
 		if err := iamOS.saveIAMConfig(u, identityPath); err != nil {
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(ctx, err)
 			return err
 		}
 
@@ -216,7 +216,7 @@ func (iamOS *IAMObjectStore) saveIAMConfig(item interface{}, path string) error
 			return err
 		}
 	}
-	return saveConfig(context.Background(), iamOS.objAPI, path, data)
+	return saveConfig(GlobalContext, iamOS.objAPI, path, data)
 }
 
 func (iamOS *IAMObjectStore) loadIAMConfig(item interface{}, path string) error {
@@ -586,7 +586,7 @@ func listIAMConfigItems(ctx context.Context, objAPI ObjectLayer, pathPrefix stri
 
 		marker := ""
 		for {
-			lo, err := objAPI.ListObjects(context.Background(),
+			lo, err := objAPI.ListObjects(ctx,
 				minioMetaBucket, pathPrefix, marker, SlashSeparator, maxObjectList)
 			if err != nil {
 				select {

cmd/iam.go

@@ -1640,7 +1640,7 @@ func (sys *IAMSys) IsAllowedServiceAccount(args iampolicy.Args, parent string) b
 	subPolicy, err := iampolicy.ParseConfig(bytes.NewReader([]byte(spolicyStr)))
 	if err != nil {
 		// Log any error in input session policy config.
-		logger.LogIf(context.Background(), err)
+		logger.LogIf(GlobalContext, err)
 		return false
 	}
 
@@ -1783,7 +1783,7 @@ func (sys *IAMSys) IsAllowedSTS(args iampolicy.Args) bool {
 		subPolicy, err := iampolicy.ParseConfig(bytes.NewReader([]byte(spolicyStr)))
 		if err != nil {
 			// Log any error in input session policy config.
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(GlobalContext, err)
 			return false
 		}
 
@@ -1807,7 +1807,7 @@ func (sys *IAMSys) IsAllowed(args iampolicy.Args) bool {
 	if globalPolicyOPA != nil {
 		ok, err := globalPolicyOPA.IsAllowed(args)
 		if err != nil {
-			logger.LogIf(context.Background(), err)
+			logger.LogIf(GlobalContext, err)
 		}
 		return ok
 	}

cmd/logger/audit.go

@@ -125,15 +125,6 @@ func (lrw *ResponseWriter) Size() int {
 	return lrw.bytesWritten
 }
 
-// AuditTargets is the list of enabled audit loggers
-var AuditTargets = []Target{}
-
-// AddAuditTarget adds a new audit logger target to the
-// list of enabled loggers
-func AddAuditTarget(t Target) {
-	AuditTargets = append(AuditTargets, t)
-}
-
 // AuditLog - logs audit logs to all audit targets.
 func AuditLog(w http.ResponseWriter, r *http.Request, api string, reqClaims map[string]interface{}, filterKeys ...string) {
 	// Fast exit if there is not audit target configured

cmd/logger/target/console/console.go

@@ -32,6 +32,11 @@ import (
 // in plain or json format to the standard output.
 type Target struct{}
 
+// Validate - validate if the tty can be written to
+func (c *Target) Validate() error {
+	return nil
+}
+
 // Send log message 'e' to console
 func (c *Target) Send(e interface{}, logKind string) error {
 	entry, ok := e.(log.Entry)

cmd/logger/target/http/http.go

@@ -18,12 +18,16 @@ package http
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"strings"
+	"time"
 
 	xhttp "github.com/minio/minio/cmd/http"
+	"github.com/minio/minio/cmd/logger"
 )
 
 // Target implements logger.Target and sends the json
@@ -45,6 +49,47 @@ type Target struct {
 	client    http.Client
 }
 
+// Validate validate the http target
+func (h *Target) Validate() error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.endpoint, strings.NewReader(`{}`))
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set(xhttp.ContentType, "application/json")
+
+	// Set user-agent to indicate MinIO release
+	// version to the configured log endpoint
+	req.Header.Set("User-Agent", h.userAgent)
+
+	if h.authToken != "" {
+		req.Header.Set("Authorization", h.authToken)
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return err
+	}
+
+	// Drain any response.
+	xhttp.DrainBody(resp.Body)
+
+	if resp.StatusCode != http.StatusOK {
+		switch resp.StatusCode {
+		case http.StatusForbidden:
+			return fmt.Errorf("%s returned '%s', please check if your auth token is correctly set",
+				h.endpoint, resp.Status)
+		}
+		return fmt.Errorf("%s returned '%s', please check your endpoint configuration",
+			h.endpoint, resp.Status)
+	}
+
+	return nil
+}
+
 func (h *Target) startHTTPLogger() {
 	// Create a routine which sends json logs received
 	// from an internal channel.
@@ -55,8 +100,11 @@ func (h *Target) startHTTPLogger() {
 				continue
 			}
 
-			req, err := http.NewRequest(http.MethodPost, h.endpoint, bytes.NewReader(logJSON))
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+				h.endpoint, bytes.NewReader(logJSON))
 			if err != nil {
+				cancel()
 				continue
 			}
 			req.Header.Set(xhttp.ContentType, "application/json")
@@ -70,13 +118,26 @@ func (h *Target) startHTTPLogger() {
 			}
 
 			resp, err := h.client.Do(req)
+			cancel()
 			if err != nil {
-				h.client.CloseIdleConnections()
+				logger.LogIf(ctx, fmt.Errorf("%s returned '%w', please check your endpoint configuration\n",
+					h.endpoint, err))
 				continue
 			}
 
 			// Drain any response.
 			xhttp.DrainBody(resp.Body)
+
+			if resp.StatusCode != http.StatusOK {
+				switch resp.StatusCode {
+				case http.StatusForbidden:
+					logger.LogIf(ctx, fmt.Errorf("%s returned '%s', please check if your auth token is correctly set",
+						h.endpoint, resp.Status))
+				default:
+					logger.LogIf(ctx, fmt.Errorf("%s returned '%s', please check your endpoint configuration",
+						h.endpoint, resp.Status))
+				}
+			}
 		}
 	}()
 }

cmd/logger/targets.go

@@ -20,14 +20,33 @@ package logger
 // a single log entry and Send it to the log target
 //   e.g. Send the log to a http server
 type Target interface {
+	Validate() error
 	Send(entry interface{}, errKind string) error
 }
 
 // Targets is the set of enabled loggers
 var Targets = []Target{}
 
+// AuditTargets is the list of enabled audit loggers
+var AuditTargets = []Target{}
+
+// AddAuditTarget adds a new audit logger target to the
+// list of enabled loggers
+func AddAuditTarget(t Target) error {
+	if err := t.Validate(); err != nil {
+		return err
+	}
+
+	AuditTargets = append(AuditTargets, t)
+	return nil
+}
+
 // AddTarget adds a new logger target to the
 // list of enabled loggers
-func AddTarget(t Target) {
+func AddTarget(t Target) error {
+	if err := t.Validate(); err != nil {
+		return err
+	}
 	Targets = append(Targets, t)
+	return nil
 }

cmd/namespace-lock.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/minio/minio/cmd/config/storageclass"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/dsync"
 	"github.com/minio/minio/pkg/lsync"
@@ -147,7 +148,18 @@ func (di *distLockInstance) GetLock(timeout *dynamicTimeout) (timedOutErr error)
 	lockSource := getSource(2)
 	start := UTCNow()
 
-	if !di.rwMutex.GetLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
+	// Lockers default to standard storage class always, why because
+	// we always dictate storage tolerance in terms of standard
+	// storage class be it number of drives or a multiplicative
+	// of number of nodes, defaulting lockers to this value
+	// simply means that locking is always similar in behavior
+	// and effect with erasure coded drive tolerance.
+	tolerance := globalStorageClass.GetParityForSC(storageclass.STANDARD)
+
+	if !di.rwMutex.GetLock(di.ctx, di.opsID, lockSource, dsync.Options{
+		Timeout:   timeout.Timeout(),
+		Tolerance: tolerance,
+	}) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}
@@ -164,7 +176,14 @@ func (di *distLockInstance) Unlock() {
 func (di *distLockInstance) GetRLock(timeout *dynamicTimeout) (timedOutErr error) {
 	lockSource := getSource(2)
 	start := UTCNow()
-	if !di.rwMutex.GetRLock(di.ctx, di.opsID, lockSource, timeout.Timeout()) {
+
+	// Lockers default to standard storage class always.
+	tolerance := globalStorageClass.GetParityForSC(storageclass.STANDARD)
+
+	if !di.rwMutex.GetRLock(di.ctx, di.opsID, lockSource, dsync.Options{
+		Timeout:   timeout.Timeout(),
+		Tolerance: tolerance,
+	}) {
 		timeout.LogFailure()
 		return OperationTimedOut{}
 	}

cmd/object-api-listobjects_test.go

@@ -1,5 +1,5 @@
 /*
- * MinIO Cloud Storage, (C) 2015-2016, 2017 MinIO, Inc.
+ * MinIO Cloud Storage, (C) 2015-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.
@@ -448,7 +448,9 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 				{Name: "temporary/0/"},
 			},
 		},
-		// ListObjectsResult-34 Listing with marker > last object should return empty
+		// ListObjectsResult-34:
+		//    * Listing with marker > last object should return empty
+		//    * Listing an object with a trailing slash and '/' delimiter
 		{
 			IsTruncated: false,
 			Objects:     []ObjectInfo{},
@@ -569,6 +571,8 @@ func testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler) {
 		{"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},
+		// Test listing an object with a trailing slash and a slash delimiter (66)
+		{"test-bucket-list-object", "Asia-maps.png/", "", "/", 1000, resultCases[34], nil, true},
 	}
 
 	for i, testCase := range testCases {

cmd/object-api-utils.go

@@ -400,6 +400,15 @@ func (o ObjectInfo) IsCompressedOK() (bool, error) {
 	return true, fmt.Errorf("unknown compression scheme: %s", scheme)
 }
 
+// GetActualETag - returns the actual etag of the stored object
+// decrypts SSE objects.
+func (o ObjectInfo) GetActualETag(h http.Header) string {
+	if !crypto.IsEncrypted(o.UserDefined) {
+		return o.ETag
+	}
+	return getDecryptedETag(h, o, false)
+}
+
 // GetActualSize - returns the actual size of the stored object
 func (o ObjectInfo) GetActualSize() (int64, error) {
 	if crypto.IsEncrypted(o.UserDefined) {

cmd/object-handlers.go

@@ -1261,7 +1261,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
 	encodedSuccessResponse := encodeResponse(response)
 	if mustReplicate(ctx, r, dstBucket, dstObject, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		defer replicateObject(context.Background(), dstBucket, dstObject, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, dstBucket, dstObject, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedCopy,
 			BucketName:   dstBucket,
 			Object:       objInfo,
@@ -1577,7 +1577,7 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		}
 	}
 	if mustReplicate(ctx, r, bucket, object, metadata, "") {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedPut,
 			BucketName:   bucket,
 			Object:       objInfo,
@@ -2652,7 +2652,7 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 
 	setPutObjHeaders(w, objInfo, false)
 	if mustReplicate(ctx, r, bucket, object, objInfo.UserDefined, objInfo.ReplicationStatus.String()) {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedCompleteMultipartUpload,
 			BucketName:   bucket,
 			Object:       objInfo,

cmd/peer-rest-server.go

@@ -182,7 +182,7 @@ func (s *peerRESTServer) DeleteServiceAccountHandler(w http.ResponseWriter, r *h
 		return
 	}
 
-	if err := globalIAMSys.DeleteServiceAccount(context.Background(), accessKey); err != nil {
+	if err := globalIAMSys.DeleteServiceAccount(r.Context(), accessKey); err != nil {
 		s.writeErrorResponse(w, err)
 		return
 	}

cmd/server-main.go

@@ -122,6 +122,11 @@ func serverHandleCmdArgs(ctx *cli.Context) {
 	globalRootCAs, err = certs.GetRootCAs(globalCertsCADir.Get())
 	logger.FatalIf(err, "Failed to read root CAs (%v)", err)
 
+	// Add the global public crts as part of global root CAs
+	for _, publicCrt := range globalPublicCerts {
+		globalRootCAs.AddCert(publicCrt)
+	}
+
 	// Register root CAs for remote ENVs
 	env.RegisterGlobalCAs(globalRootCAs)
 
@@ -224,6 +229,8 @@ func initSafeMode(ctx context.Context, newObject ObjectLayer) (err error) {
 		initAutoHeal(ctx, newObject)
 	}
 
+	timeout := newDynamicTimeout(3*time.Second, 3*time.Second)
+
 	// ****  WARNING ****
 	// Migrating to encrypted backend should happen before initialization of any
 	// sub-systems, make sure that we do not move the above codeblock elsewhere.
@@ -239,7 +246,7 @@ func initSafeMode(ctx context.Context, newObject ObjectLayer) (err error) {
 	for range retry.NewTimer(retryCtx) {
 		// let one of the server acquire the lock, if not let them timeout.
 		// which shall be retried again by this loop.
-		if err = txnLk.GetLock(newDynamicTimeout(3*time.Second, 3*time.Second)); err != nil {
+		if err = txnLk.GetLock(timeout); err != nil {
 			logger.Info("Waiting for all MinIO sub-systems to be initialized.. trying to acquire lock")
 			continue
 		}

cmd/storage-rest-client.go

@@ -478,7 +478,7 @@ func (client *storageRESTClient) WalkVersions(volume, dirPath, marker string, re
 			if gerr := decoder.Decode(&fi); gerr != nil {
 				// Upon error return
 				if gerr != io.EOF {
-					logger.LogIf(context.Background(), gerr)
+					logger.LogIf(GlobalContext, gerr)
 				}
 				return
 			}

cmd/web-handlers.go

@@ -1173,7 +1173,7 @@ func (web *webAPIHandlers) Upload(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	if mustReplicate {
-		defer replicateObject(context.Background(), bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
+		defer replicateObject(GlobalContext, bucket, object, objInfo.VersionID, objectAPI, &eventArgs{
 			EventName:    event.ObjectCreatedPut,
 			BucketName:   bucket,
 			Object:       objInfo,

cmd/xl-storage.go

@@ -949,6 +949,14 @@ func (s *xlStorage) Walk(volume, dirPath, marker string, recursive bool, endWalk
 		return nil, err
 	}
 
+	// Fast exit track to check if we are listing an object with
+	// a trailing slash, this will avoid to list the object content.
+	if HasSuffix(dirPath, SlashSeparator) {
+		if st, err := os.Stat(pathJoin(volumeDir, dirPath, xlStorageFormatFile)); err == nil && st.Mode().IsRegular() {
+			return nil, errFileNotFound
+		}
+	}
+
 	// buffer channel matches the S3 ListObjects implementation
 	ch = make(chan FileInfo, maxObjectList)
 	go func() {

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

@@ -5,7 +5,7 @@ version: '3.7'
 # 9001 through 9004.
 services:
   minio1:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     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-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     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-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2020-08-08T04-50-06Z
+    image: minio/minio:RELEASE.2020-08-13T02-39-50Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/zh_CN/bucket/retention/README.md

@@ -0,0 +1,60 @@
+# 对象锁定指南 [![Slack](https://slack.min.io/slack?type=svg)](https://slack.min.io)
+
+MinIO服务器允许针对特定的对象一次写入，多次读取 (WORM),或者通过使用默认的对象锁定配置设置存储桶，该配置将默认的保留模式和保留期限应用于所有对象。这将使存储桶中的对象不可改变，也就是说，直到存储桶的对象锁定配置或对象保留中指定的时间到期后，才允许删除版本。
+
+对象锁定要求在创建存储桶时启用锁定，而且还会自动在存储桶上启用版本控制。此外，可以在存储桶上配置默认保留期限和保留模式，以应用于该存储桶中创建的对象。
+
+和保留期限无关，一个对象也可以被依法保留。依法保留的对象将一直有效（不可变），直到通过API调用删除依法保留为止。
+
+## 开始使用
+
+### 1. 前置条件
+
+- 安装 MinIO - [MinIO快速入门指南](https://docs.min.io/cn/minio-quickstart-guide)
+- 安装 `awscli` - [Installing AWS Command Line Interface](https://docs.aws.amazon.com/zh_cn/cli/latest/userguide/cli-chap-install.html)
+
+### 2. 设置存储桶WORM配置
+
+通过设置对象锁定配置可以启用存储桶的一次写入，多次读取 (WORM)模式。此配置将应用于存储桶中的现有对象和新对象。下面的例子设置了`Governance`模式，并且`mybucket`里的所有对象自创建时起，保留一天的时间。
+
+```sh
+$ awscli s3api put-object-lock-configuration --bucket mybucket --object-lock-configuration 'ObjectLockEnabled=\"Enabled\",Rule={DefaultRetention={Mode=\"GOVERNANCE\",Days=1}}'
+```
+
+### 设置对象锁定
+
+PutObject API 可以通过`x-amz-object-lock-mode`和`x-amz-object-lock-retain-until-date`请求头设置每一个对象的保留模式和保留期限。这要比存储桶上设置的对象锁定配置优先级高。
+
+```sh
+aws s3api put-object --bucket testbucket --key lockme --object-lock-mode GOVERNANCE --object-lock-retain-until-date "2019-11-20"  --body /etc/issue
+```
+
+请参阅 https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/dev/object-lock-overview.html 以获取有关对象锁定、governance bypass所需权限的AWS S3规范。
+
+### 设置依法保留对象
+
+PutObject API 可以通过 `x-amz-object-lock-legal-hold` 请求头设置依法保留.
+
+```sh
+aws s3api put-object --bucket testbucket --key legalhold --object-lock-legal-hold-status ON --body /etc/issue
+```
+
+请参阅 https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/dev/object-lock-overview.html 以获取有关对象锁定、指定依法保留所需的权限。
+
+## 概念
+- 如果一个对象被依法保留,除非明确删除各个版本ID的依法保留，否则无法删除该对象。 DeleteObjectVersion() 会失败。
+- 在 `Compliance` 模式中, 在各个版本ID的保留期限到期之前，任何人都无法删除对象。如果用户具有必需的governance bypass权限，则可以在`Compliance`模式下延长对象的保留期限。
+- 一旦在一个存储桶上设置对象锁定设置后
+  - 新对象会自动继承存储桶对象锁定配置的保留设置
+  - 上传对象时可以选择是否设置保留头信息
+  - 会对对象调用PutObjectRetention API
+- 如果不需要系统时间来设置保留日期，则可以将*MINIO_NTP_SERVER*环境变量设置为远程NTP server endpoint。
+- **对象锁定功能仅在纠删码和分布式纠删码模式下可用**。
+
+## 进一步探索
+
+- [使用`mc`](https://docs.min.io/cn/minio-client-quickstart-guide)
+- [使用`aws-cli`](https://docs.min.io/cn/aws-cli-with-minio)
+- [使用`s3cmd`](https://docs.min.io/cn/s3cmd-with-minio)
+- [使用`minio-go`](https://docs.min.io/cn/golang-client-quickstart-guide)
+- [MinIO文档](https://docs.min.io/cn)

go.mod

@@ -83,6 +83,7 @@ require (
 	golang.org/x/crypto v0.0.0-20200709230013-948cd5f35899
 	golang.org/x/net v0.0.0-20200707034311-ab3426394381
 	golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae
+	golang.org/x/tools v0.0.0-20200814172026-c4923e618c08 // indirect
 	google.golang.org/api v0.5.0
 	gopkg.in/jcmturner/gokrb5.v7 v7.3.0
 	gopkg.in/ldap.v3 v3.0.3

go.sum

@@ -447,6 +447,7 @@ github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.etcd.io/bbolt v1.3.5 h1:XAzx9gjCb0Rxj7EoqcClPD1d5ZBxZJk0jbuoPHenBt0=
 go.etcd.io/bbolt v1.3.5/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
 go.etcd.io/etcd/v3 v3.3.0-rc.0.0.20200707003333-58bb8ae09f8e h1:HZQLoe71Q24wVyDrGBRcVuogx32U+cPlcm/WoSLUI6c=
@@ -487,6 +488,8 @@ golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.2.0 h1:KU7oHjnv3XNWfa5COkzUifxZmxp1TyI7ImMXqFxLwvQ=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0 h1:RM4zey1++hCTbCVQfnWeKs9/IEsaBLA8vTkd0WVtmH4=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -500,6 +503,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20191112182307-2180aed22343/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -512,6 +516,8 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208 h1:qwRHBd0NqMbJxfbotnDhm2ByMI1Shq4Y6oRJo21SGJA=
+golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -560,6 +566,8 @@ golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200425043458-8463f397d07c h1:iHhCR0b26amDCiiO+kBguKZom9aMF+NrFxh9zeKR/XU=
 golang.org/x/tools v0.0.0-20200425043458-8463f397d07c/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20200814172026-c4923e618c08 h1:sfBQLM20fzeXhOixVQirwEbuW4PGStP773EXQpsBB6E=
+golang.org/x/tools v0.0.0-20200814172026-c4923e618c08/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=

pkg/disk/root_disk_unix.go

@@ -31,7 +31,7 @@ func IsRootDisk(diskPath string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-	rootInfo, err := os.Stat("/")
+	rootInfo, err := os.Stat("/etc/hosts")
 	if err != nil {
 		return false, err
 	}

pkg/dsync/drwmutex.go

@@ -86,7 +86,15 @@ func NewDRWMutex(clnt *Dsync, names ...string) *DRWMutex {
 func (dm *DRWMutex) Lock(id, source string) {
 
 	isReadLock := false
-	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), id, source, isReadLock, Options{
+		Timeout: drwMutexInfinite,
+	})
+}
+
+// Options lock options.
+type Options struct {
+	Timeout   time.Duration
+	Tolerance int
 }
 
 // GetLock tries to get a write lock on dm before the timeout elapses.
@@ -94,10 +102,10 @@ func (dm *DRWMutex) Lock(id, source string) {
 // If the lock is already in use, the calling go routine
 // blocks until either the mutex becomes available and return success or
 // more time has passed than the timeout value and return false.
-func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, opts Options) (locked bool) {
 
 	isReadLock := false
-	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, id, source, isReadLock, opts)
 }
 
 // RLock holds a read lock on dm.
@@ -107,7 +115,9 @@ func (dm *DRWMutex) GetLock(ctx context.Context, id, source string, timeout time
 func (dm *DRWMutex) RLock(id, source string) {
 
 	isReadLock := true
-	dm.lockBlocking(context.Background(), drwMutexInfinite, id, source, isReadLock)
+	dm.lockBlocking(context.Background(), id, source, isReadLock, Options{
+		Timeout: drwMutexInfinite,
+	})
 }
 
 // GetRLock tries to get a read lock on dm before the timeout elapses.
@@ -116,10 +126,10 @@ func (dm *DRWMutex) RLock(id, source string) {
 // Otherwise the calling go routine blocks until either the mutex becomes
 // available and return success or more time has passed than the timeout
 // value and return false.
-func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, timeout time.Duration) (locked bool) {
+func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, opts Options) (locked bool) {
 
 	isReadLock := true
-	return dm.lockBlocking(ctx, timeout, id, source, isReadLock)
+	return dm.lockBlocking(ctx, id, source, isReadLock, opts)
 }
 
 // lockBlocking will try to acquire either a read or a write lock
@@ -127,10 +137,10 @@ func (dm *DRWMutex) GetRLock(ctx context.Context, id, source string, timeout tim
 // The function will loop using a built-in timing randomized back-off
 // algorithm until either the lock is acquired successfully or more
 // time has elapsed than the timeout value.
-func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id, source string, isReadLock bool) (locked bool) {
+func (dm *DRWMutex) lockBlocking(ctx context.Context, id, source string, isReadLock bool, opts Options) (locked bool) {
 	restClnts := dm.clnt.GetLockersFn()
 
-	retryCtx, cancel := context.WithTimeout(ctx, timeout)
+	retryCtx, cancel := context.WithTimeout(ctx, opts.Timeout)
 
 	defer cancel()
 
@@ -140,7 +150,7 @@ func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id,
 		locks := make([]string, len(restClnts))
 
 		// Try to acquire the lock.
-		locked = lock(retryCtx, dm.clnt, &locks, id, source, isReadLock, dm.Names...)
+		locked = lock(retryCtx, dm.clnt, &locks, id, source, isReadLock, opts.Tolerance, dm.Names...)
 		if !locked {
 			continue
 		}
@@ -167,10 +177,29 @@ func (dm *DRWMutex) lockBlocking(ctx context.Context, timeout time.Duration, id,
 }
 
 // lock tries to acquire the distributed lock, returning true or false.
-func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, isReadLock bool, lockNames ...string) bool {
+func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, isReadLock bool, tolerance int, lockNames ...string) bool {
 
 	restClnts := ds.GetLockersFn()
 
+	// Tolerance is not set, defaults to half of the locker clients.
+	if tolerance == 0 {
+		tolerance = len(restClnts) / 2
+	}
+
+	// Quorum is effectively = total clients subtracted with tolerance limit
+	quorum := len(restClnts) - tolerance
+	if !isReadLock {
+		// In situations for write locks, as a special case
+		// to avoid split brains we make sure to acquire
+		// quorum + 1 when tolerance is exactly half of the
+		// total locker clients.
+		if quorum == tolerance {
+			quorum++
+		}
+	}
+
+	tolerance = len(restClnts) - quorum
+
 	// Create buffered channel of size equal to total number of nodes.
 	ch := make(chan Granted, len(restClnts))
 	defer close(ch)
@@ -217,7 +246,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		}(index, isReadLock, c)
 	}
 
-	quorum := false
+	quorumMet := false
 
 	wg.Add(1)
 	go func(isReadLock bool) {
@@ -232,9 +261,6 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		done := false
 		timeout := time.After(DRWMutexAcquireTimeout)
 
-		dquorumReads := (len(restClnts) + 1) / 2
-		dquorum := dquorumReads + 1
-
 		for ; i < len(restClnts); i++ { // Loop until we acquired all locks
 
 			select {
@@ -244,8 +270,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 					(*locks)[grant.index] = grant.lockUID
 				} else {
 					locksFailed++
-					if !isReadLock && locksFailed > len(restClnts)-dquorum ||
-						isReadLock && locksFailed > len(restClnts)-dquorumReads {
+					if locksFailed > tolerance {
 						// We know that we are not going to get the lock anymore,
 						// so exit out and release any locks that did get acquired
 						done = true
@@ -258,7 +283,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 				done = true
 				// timeout happened, maybe one of the nodes is slow, count
 				// number of locks to check whether we have quorum or not
-				if !quorumMet(locks, isReadLock, dquorum, dquorumReads) {
+				if !checkQuorumMet(locks, quorum) {
 					log("Quorum not met after timeout\n")
 					releaseAll(ds, locks, isReadLock, restClnts, lockNames...)
 				} else {
@@ -272,7 +297,7 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 		}
 
 		// Count locks in order to determine whether we have quorum or not
-		quorum = quorumMet(locks, isReadLock, dquorum, dquorumReads)
+		quorumMet = checkQuorumMet(locks, quorum)
 
 		// Signal that we have the quorum
 		wg.Done()
@@ -292,12 +317,11 @@ func lock(ctx context.Context, ds *Dsync, locks *[]string, id, source string, is
 
 	wg.Wait()
 
-	return quorum
+	return quorumMet
 }
 
-// quorumMet determines whether we have acquired the required quorum of underlying locks or not
-func quorumMet(locks *[]string, isReadLock bool, quorum, quorumReads int) bool {
-
+// checkQuorumMet determines whether we have acquired the required quorum of underlying locks or not
+func checkQuorumMet(locks *[]string, quorum int) bool {
 	count := 0
 	for _, uid := range *locks {
 		if isLocked(uid) {
@@ -305,14 +329,7 @@ func quorumMet(locks *[]string, isReadLock bool, quorum, quorumReads int) bool {
 		}
 	}
 
-	var metQuorum bool
-	if isReadLock {
-		metQuorum = count >= quorumReads
-	} else {
-		metQuorum = count >= quorum
-	}
-
-	return metQuorum
+	return count >= quorum
 }
 
 // releaseAll releases all locks that are marked as locked

pkg/dsync/drwmutex_test.go

@@ -36,12 +36,12 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 
 	drwm := NewDRWMutex(ds, "simplelock")
 
-	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("1st read lock acquired, waiting...")
 
-	if !drwm.GetRLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire read lock")
 	}
 	// fmt.Println("2nd read lock acquired, waiting...")
@@ -59,7 +59,7 @@ func testSimpleWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire write lock, waiting...")
-	locked = drwm.GetLock(context.Background(), id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, Options{Timeout: duration})
 	if locked {
 		// fmt.Println("Write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -93,7 +93,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	drwm := NewDRWMutex(ds, "duallock")
 
 	// fmt.Println("Getting initial write lock")
-	if !drwm.GetLock(context.Background(), id, source, time.Second) {
+	if !drwm.GetLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 		panic("Failed to acquire initial write lock")
 	}
 
@@ -104,7 +104,7 @@ func testDualWriteLock(t *testing.T, duration time.Duration) (locked bool) {
 	}()
 
 	// fmt.Println("Trying to acquire 2nd write lock, waiting...")
-	locked = drwm.GetLock(context.Background(), id, source, duration)
+	locked = drwm.GetLock(context.Background(), id, source, Options{Timeout: duration})
 	if locked {
 		// fmt.Println("2nd write lock acquired, waiting...")
 		time.Sleep(time.Second)
@@ -139,7 +139,7 @@ func TestDualWriteLockTimedOut(t *testing.T) {
 
 // Borrowed from rwmutex_test.go
 func parallelReader(ctx context.Context, m *DRWMutex, clocked, cunlock, cdone chan bool) {
-	if m.GetRLock(ctx, id, source, time.Second) {
+	if m.GetRLock(ctx, id, source, Options{Timeout: time.Second}) {
 		clocked <- true
 		<-cunlock
 		m.RUnlock()
@@ -182,7 +182,7 @@ func TestParallelReaders(t *testing.T) {
 // Borrowed from rwmutex_test.go
 func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetRLock(context.Background(), id, source, time.Second) {
+		if rwm.GetRLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 			n := atomic.AddInt32(activity, 1)
 			if n < 1 || n >= 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))
@@ -199,7 +199,7 @@ func reader(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool)
 // Borrowed from rwmutex_test.go
 func writer(rwm *DRWMutex, numIterations int, activity *int32, cdone chan bool) {
 	for i := 0; i < numIterations; i++ {
-		if rwm.GetLock(context.Background(), id, source, time.Second) {
+		if rwm.GetLock(context.Background(), id, source, Options{Timeout: time.Second}) {
 			n := atomic.AddInt32(activity, 10000)
 			if n != 10000 {
 				panic(fmt.Sprintf("wlock(%d)\n", n))

pkg/dsync/dsync_test.go

@@ -63,7 +63,7 @@ func TestMain(m *testing.M) {
 
 	rand.Seed(time.Now().UTC().UnixNano())
 
-	nodes := make([]string, 4) // list of node IP addrs or hostname with ports.
+	nodes := make([]string, 5) // list of node IP addrs or hostname with ports.
 	for i := range nodes {
 		nodes[i] = fmt.Sprintf("127.0.0.1:%d", i+12345)
 	}

pkg/licverifier/verifier.go

@@ -32,20 +32,21 @@ type LicenseVerifier struct {
 
 // LicenseInfo holds customer metadata present in the license key.
 type LicenseInfo struct {
-	Email           string `json:"sub"`         // Email of the license key requestor
-	TeamName        string `json:"teamName"`    // Subnet team name
-	AccountID       int64  `json:"accountId"`   // Subnet account id
-	StorageCapacity int64  `json:"capacity"`    // Storage capacity used in TB
-	ServiceType     string `json:"serviceType"` // Subnet service type
+	Email           string // Email of the license key requestor
+	TeamName        string // Subnet team name
+	AccountID       int64  // Subnet account id
+	StorageCapacity int64  // Storage capacity used in TB
+	ServiceType     string // Subnet service type
 }
 
-// Valid checks if customer metadata from the license key is valid.
-func (li *LicenseInfo) Valid() error {
-	if li.AccountID <= 0 {
-		return errors.New("Invalid accountId in claims")
-	}
-	return nil
-}
+// license key JSON field names
+const (
+	accountID   = "accountId"
+	sub         = "sub"
+	teamName    = "teamName"
+	capacity    = "capacity"
+	serviceType = "serviceType"
+)
 
 // NewLicenseVerifier returns an initialized license verifier with the given
 // ECDSA public key in PEM format.
@@ -59,14 +60,49 @@ func NewLicenseVerifier(pemBytes []byte) (*LicenseVerifier, error) {
 	}, nil
 }
 
+// toLicenseInfo extracts LicenseInfo from claims. It returns an error if any of
+// the claim values are invalid.
+func toLicenseInfo(claims jwt.MapClaims) (LicenseInfo, error) {
+	accID, ok := claims[accountID].(float64)
+	if !ok || ok && accID <= 0 {
+		return LicenseInfo{}, errors.New("Invalid accountId in claims")
+	}
+	email, ok := claims[sub].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid email in claims")
+	}
+	tName, ok := claims[teamName].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid team name in claims")
+	}
+	storageCap, ok := claims[capacity].(float64)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid storage capacity in claims")
+	}
+	sType, ok := claims[serviceType].(string)
+	if !ok {
+		return LicenseInfo{}, errors.New("Invalid service type in claims")
+	}
+	return LicenseInfo{
+		Email:           email,
+		TeamName:        tName,
+		AccountID:       int64(accID),
+		StorageCapacity: int64(storageCap),
+		ServiceType:     sType,
+	}, nil
+
+}
+
 // Verify verifies the license key and validates the claims present in it.
 func (lv *LicenseVerifier) Verify(license string) (LicenseInfo, error) {
-	var licInfo LicenseInfo
-	_, err := jwt.ParseWithClaims(license, &licInfo, func(token *jwt.Token) (interface{}, error) {
+	token, err := jwt.ParseWithClaims(license, &jwt.MapClaims{}, func(token *jwt.Token) (interface{}, error) {
 		return lv.ecPubKey, nil
 	})
 	if err != nil {
 		return LicenseInfo{}, fmt.Errorf("Failed to verify license: %s", err)
 	}
-	return licInfo, nil
+	if claims, ok := token.Claims.(*jwt.MapClaims); ok && token.Valid {
+		return toLicenseInfo(*claims)
+	}
+	return LicenseInfo{}, errors.New("Invalid claims found in license")
 }

pkg/licverifier/verifier_test.go

@@ -19,8 +19,18 @@ package licverifier
 import (
 	"fmt"
 	"testing"
+	"time"
+
+	"github.com/dgrijalva/jwt-go"
 )
 
+// at fixes the jwt.TimeFunc at t and calls f in that context.
+func at(t time.Time, f func()) {
+	jwt.TimeFunc = func() time.Time { return t }
+	f()
+	jwt.TimeFunc = time.Now
+}
+
 // TestLicenseVerify tests the license key verification process with a valid and
 // an invalid key.
 func TestLicenseVerify(t *testing.T) {
@@ -48,18 +58,22 @@ mr/cKCUyBL7rcAvg0zNq1vcSrUSGlAmY3SEDCu3GOKnjG/U4E7+p957ocWSV+mQU
 	}
 
 	for i, tc := range testCases {
-		licInfo, err := lv.Verify(tc.lic)
-		if err != nil && tc.shouldPass {
-			t.Fatalf("%d: Expected license to pass verification but failed with %s", i+1, err)
-		}
-		if err == nil {
-			if !tc.shouldPass {
-				t.Fatalf("%d: Expected license to fail verification but passed", i+1)
+		// Fixing the jwt.TimeFunc at 2020-08-05 22:17:43 +0000 UTC to
+		// ensure that the license JWT doesn't expire ever.
+		at(time.Unix(int64(1596665863), 0), func() {
+			licInfo, err := lv.Verify(tc.lic)
+			if err != nil && tc.shouldPass {
+				t.Fatalf("%d: Expected license to pass verification but failed with %s", i+1, err)
 			}
-			if tc.expectedLicInfo != licInfo {
-				t.Fatalf("%d: Expected license info %v but got %v", i+1, tc.expectedLicInfo, licInfo)
+			if err == nil {
+				if !tc.shouldPass {
+					t.Fatalf("%d: Expected license to fail verification but passed", i+1)
+				}
+				if tc.expectedLicInfo != licInfo {
+					t.Fatalf("%d: Expected license info %v but got %v", i+1, tc.expectedLicInfo, licInfo)
+				}
 			}
-		}
+		})
 	}
 }