audit: Add the drives where the dangling object is removed (#18737)

Add a hidden configuration under the scanner sub section to configure if
the scanner should sleep between two objects scan. The configuration has
only effect when there is no drive activity related to s3 requests or
healing.

By default, the code will keep the current behavior which is doing
sleep between objects.

To forcefully enable the full scan speed in idle mode, you can do this:

   `mc admin config set myminio scanner idle_speed=full`

.github/workflows/replication.yaml

@@ -54,3 +54,8 @@ jobs:
           sudo sysctl net.ipv6.conf.default.disable_ipv6=0
           make test-site-replication-minio
 
+      - name: Test Versioning
+        run: |
+          sudo sysctl net.ipv6.conf.all.disable_ipv6=0
+          sudo sysctl net.ipv6.conf.default.disable_ipv6=0
+          make test-versioning

Makefile

@@ -59,6 +59,10 @@ test-decom: install-race
 	@env bash $(PWD)/docs/distributed/decom-encrypted-sse-s3.sh
 	@env bash $(PWD)/docs/distributed/decom-compressed-sse-s3.sh
 
+test-versioning: install-race
+	@echo "Running minio versioning tests"
+	@env bash $(PWD)/docs/bucket/versioning/versioning-tests.sh
+
 test-configfile: install-race
 	@env bash $(PWD)/docs/distributed/distributed-from-config-file.sh
 

cmd/api-errors.go

@@ -430,6 +430,9 @@ const (
 	ErrLambdaARNInvalid
 	ErrLambdaARNNotFound
 
+	// New Codes for GetObjectAttributes and GetObjectVersionAttributes
+	ErrInvalidAttributeName
+
 	apiErrCodeEnd // This is used only for the testing code
 )
 
@@ -2063,6 +2066,11 @@ var errorCodes = errorCodeMap{
 		Description:    "The specified policy is not found.",
 		HTTPStatusCode: http.StatusNotFound,
 	},
+	ErrInvalidAttributeName: {
+		Code:           "InvalidArgument",
+		Description:    "Invalid attribute name specified.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	// Add your error structure here.
 }
 

cmd/api-router.go

@@ -227,6 +227,9 @@ func registerAPIRouter(router *mux.Router) {
 		// HeadObject
 		router.Methods(http.MethodHead).Path("/{object:.+}").HandlerFunc(
 			collectAPIStats("headobject", maxClients(gz(httpTraceAll(api.HeadObjectHandler)))))
+		// GetObjectAttribytes
+		router.Methods(http.MethodGet).Path("/{object:.+}").HandlerFunc(
+			collectAPIStats("getobjectattributes", maxClients(gz(httpTraceHdrs(api.GetObjectAttributesHandler))))).Queries("attributes", "")
 		// CopyObjectPart
 		router.Methods(http.MethodPut).Path("/{object:.+}").
 			HeadersRegexp(xhttp.AmzCopySource, ".*?(\\/|%2F).*?").

cmd/config-current.go

@@ -600,6 +600,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 			return fmt.Errorf("Unable to apply scanner config: %w", err)
 		}
 		// update dynamic scanner values.
+		scannerIdleMode.Store(scannerCfg.IdleMode)
 		scannerCycle.Store(scannerCfg.Cycle)
 		logger.LogIf(ctx, scannerSleeper.Update(scannerCfg.Delay, scannerCfg.MaxWait))
 	case config.LoggerWebhookSubSys:

cmd/data-scanner.go

@@ -65,8 +65,9 @@ var (
 	globalHealConfig heal.Config
 
 	// Sleeper values are updated when config is loaded.
-	scannerSleeper = newDynamicSleeper(2, time.Second, true) // Keep defaults same as config defaults
-	scannerCycle   = uatomic.NewDuration(dataScannerStartDelay)
+	scannerSleeper  = newDynamicSleeper(2, time.Second, true) // Keep defaults same as config defaults
+	scannerCycle    = uatomic.NewDuration(dataScannerStartDelay)
+	scannerIdleMode = uatomic.NewInt32(0) // default is throttled when idle
 )
 
 // initDataScanner will start the scanner in the background.
@@ -246,6 +247,8 @@ type folderScanner struct {
 	healObjectSelect      uint32 // Do a heal check on an object once every n cycles. Must divide into healFolderInclude
 	scanMode              madmin.HealScanMode
 
+	weSleep func() bool
+
 	disks       []StorageAPI
 	disksQuorum int
 
@@ -299,7 +302,7 @@ type folderScanner struct {
 // The returned cache will always be valid, but may not be updated from the existing.
 // Before each operation sleepDuration is called which can be used to temporarily halt the scanner.
 // If the supplied context is canceled the function will return at the first chance.
-func scanDataFolder(ctx context.Context, disks []StorageAPI, basePath string, cache dataUsageCache, getSize getSizeFn, scanMode madmin.HealScanMode) (dataUsageCache, error) {
+func scanDataFolder(ctx context.Context, disks []StorageAPI, basePath string, cache dataUsageCache, getSize getSizeFn, scanMode madmin.HealScanMode, weSleep func() bool) (dataUsageCache, error) {
 	switch cache.Info.Name {
 	case "", dataUsageRoot:
 		return cache, errors.New("internal error: root scan attempted")
@@ -316,6 +319,7 @@ func scanDataFolder(ctx context.Context, disks []StorageAPI, basePath string, ca
 		dataUsageScannerDebug: false,
 		healObjectSelect:      0,
 		scanMode:              scanMode,
+		weSleep:               weSleep,
 		updates:               cache.Info.updates,
 		updateCurrentPath:     updatePath,
 		disks:                 disks,
@@ -372,6 +376,8 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 	done := ctx.Done()
 	scannerLogPrefix := color.Green("folder-scanner:")
 
+	noWait := func() {}
+
 	thisHash := hashPath(folder.name)
 	// Store initial compaction state.
 	wasCompacted := into.Compacted
@@ -401,8 +407,10 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 		if !f.oldCache.Info.replication.Empty() && f.oldCache.Info.replication.Config.HasActiveRules(prefix, true) {
 			replicationCfg = f.oldCache.Info.replication
 		}
-		// Check if we can skip it due to bloom filter...
-		scannerSleeper.Sleep(ctx, dataScannerSleepPerFolder)
+
+		if f.weSleep() {
+			scannerSleeper.Sleep(ctx, dataScannerSleepPerFolder)
+		}
 
 		var existingFolders, newFolders []cachedFolder
 		var foundObjects bool
@@ -453,8 +461,11 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 				return nil
 			}
 
-			// Dynamic time delay.
-			wait := scannerSleeper.Timer(ctx)
+			wait := noWait
+			if f.weSleep() {
+				// Dynamic time delay.
+				wait = scannerSleeper.Timer(ctx)
+			}
 
 			// Get file size, ignore errors.
 			item := scannerItem{
@@ -704,8 +715,11 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 						// this object might be dangling.
 						entry, _ = entries.firstFound()
 					}
-					// wait timer per object.
-					wait := scannerSleeper.Timer(ctx)
+					wait := noWait
+					if f.weSleep() {
+						// wait timer per object.
+						wait = scannerSleeper.Timer(ctx)
+					}
 					defer wait()
 					f.updateCurrentPath(entry.name)
 					stopFn := globalScannerMetrics.log(scannerMetricHealAbandonedObject, f.root, entry.name)

cmd/data-usage_test.go

@@ -62,7 +62,9 @@ func TestDataUsageUpdate(t *testing.T) {
 		return
 	}
 
-	got, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize, 0)
+	weSleep := func() bool { return false }
+
+	got, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize, 0, weSleep)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -173,7 +175,7 @@ func TestDataUsageUpdate(t *testing.T) {
 	}
 	// Changed dir must be picked up in this many cycles.
 	for i := 0; i < dataUsageUpdateDirCycles; i++ {
-		got, err = scanDataFolder(context.Background(), nil, base, got, getSize, 0)
+		got, err = scanDataFolder(context.Background(), nil, base, got, getSize, 0, weSleep)
 		got.Info.NextCycle++
 		if err != nil {
 			t.Fatal(err)
@@ -280,7 +282,10 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 		}
 		return
 	}
-	got, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket"}}, getSize, 0)
+
+	weSleep := func() bool { return false }
+
+	got, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket"}}, getSize, 0, weSleep)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -414,7 +419,7 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 	}
 	// Changed dir must be picked up in this many cycles.
 	for i := 0; i < dataUsageUpdateDirCycles; i++ {
-		got, err = scanDataFolder(context.Background(), nil, base, got, getSize, 0)
+		got, err = scanDataFolder(context.Background(), nil, base, got, getSize, 0, weSleep)
 		got.Info.NextCycle++
 		if err != nil {
 			t.Fatal(err)
@@ -562,7 +567,8 @@ func TestDataUsageCacheSerialize(t *testing.T) {
 		}
 		return
 	}
-	want, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize, 0)
+	weSleep := func() bool { return false }
+	want, err := scanDataFolder(context.Background(), nil, base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize, 0, weSleep)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/encryption-v1.go

@@ -1082,6 +1082,30 @@ func (o *ObjectInfo) metadataDecrypter() objectMetaDecryptFn {
 	}
 }
 
+// decryptChecksums will attempt to decode checksums and return it/them if set.
+// if part > 0, and we have the checksum for the part that will be returned.
+func (o *ObjectInfo) decryptPartsChecksums() {
+	data := o.Checksum
+	if len(data) == 0 {
+		return
+	}
+	if _, encrypted := crypto.IsEncrypted(o.UserDefined); encrypted {
+		decrypted, err := o.metadataDecrypter()("object-checksum", data)
+		if err != nil {
+			logger.LogIf(GlobalContext, err)
+			return
+		}
+		data = decrypted
+	}
+	cs := hash.ReadPartCheckSums(data)
+	if len(cs) == len(o.Parts) {
+		for i := range o.Parts {
+			o.Parts[i].Checksums = cs[i]
+		}
+	}
+	return
+}
+
 // metadataEncryptFn provides an encryption function for metadata.
 // Will return nil, nil if unencrypted.
 func (o *ObjectInfo) metadataEncryptFn(headers http.Header) (objectMetaEncryptFn, error) {

cmd/erasure-object.go

@@ -550,7 +550,22 @@ func (er erasureObjects) deleteIfDangling(ctx context.Context, bucket, object st
 			}, index)
 		}
 
-		g.Wait()
+		rmDisks := make(map[string]string, len(disks))
+		for index, err := range g.Wait() {
+			var errStr, diskName string
+			if err != nil {
+				errStr = err.Error()
+			} else {
+				errStr = "<nil>"
+			}
+			if disks[index] != nil {
+				diskName = disks[index].String()
+			} else {
+				diskName = fmt.Sprintf("disk-%d", index)
+			}
+			rmDisks[diskName] = errStr
+		}
+		tags["cleanupResult"] = rmDisks
 	}
 	return m, err
 }

cmd/erasure-server-pool.go

@@ -2296,6 +2296,7 @@ type HealthResult struct {
 		PoolID, SetID int
 		HealthyDrives int
 		HealingDrives int
+		ReadQuorum    int
 		WriteQuorum   int
 	}
 	WriteQuorum   int
@@ -2373,8 +2374,10 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 	}
 
 	b := z.BackendInfo()
+	poolReadQuorums := make([]int, len(b.StandardSCData))
 	poolWriteQuorums := make([]int, len(b.StandardSCData))
 	for i, data := range b.StandardSCData {
+		poolReadQuorums[i] = data
 		poolWriteQuorums[i] = data
 		if data == b.StandardSCParity {
 			poolWriteQuorums[i] = data + 1
@@ -2412,15 +2415,17 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 	for poolIdx := range erasureSetUpCount {
 		for setIdx := range erasureSetUpCount[poolIdx] {
 			result.ESHealth = append(result.ESHealth, struct {
-				Maintenance                               bool
-				PoolID, SetID                             int
-				HealthyDrives, HealingDrives, WriteQuorum int
+				Maintenance                  bool
+				PoolID, SetID                int
+				HealthyDrives, HealingDrives int
+				ReadQuorum, WriteQuorum      int
 			}{
 				Maintenance:   opts.Maintenance,
 				SetID:         setIdx,
 				PoolID:        poolIdx,
 				HealthyDrives: erasureSetUpCount[poolIdx][setIdx].online,
 				HealingDrives: erasureSetUpCount[poolIdx][setIdx].healing,
+				ReadQuorum:    poolReadQuorums[poolIdx],
 				WriteQuorum:   poolWriteQuorums[poolIdx],
 			})
 

cmd/erasure.go

@@ -522,7 +522,7 @@ func (er erasureObjects) nsScanner(ctx context.Context, buckets []BucketInfo, wa
 				// Calc usage
 				before := cache.Info.LastUpdate
 				var err error
-				cache, err = disk.NSScanner(ctx, cache, updates, healScanMode)
+				cache, err = disk.NSScanner(ctx, cache, updates, healScanMode, nil)
 				if err != nil {
 					if !cache.Info.LastUpdate.IsZero() && cache.Info.LastUpdate.After(before) {
 						logger.LogIf(ctx, cache.save(ctx, er, cacheName))

cmd/metrics-v2.go

@@ -3236,6 +3236,16 @@ func getClusterHealthStatusMD() MetricDescription {
 	}
 }
 
+func getClusterErasureSetReadQuorumMD() MetricDescription {
+	return MetricDescription{
+		Namespace: clusterMetricNamespace,
+		Subsystem: "health",
+		Name:      "erasure_set_read_quorum",
+		Help:      "Get the read quorum for this erasure set",
+		Type:      gaugeMetric,
+	}
+}
+
 func getClusterErasureSetWriteQuorumMD() MetricDescription {
 	return MetricDescription{
 		Namespace: clusterMetricNamespace,
@@ -3280,7 +3290,7 @@ func getClusterHealthMetrics() *MetricsGroup {
 		opts := HealthOptions{}
 		result := objLayer.Health(ctx, opts)
 
-		metrics = make([]Metric, 0, 2+3*len(result.ESHealth))
+		metrics = make([]Metric, 0, 2+4*len(result.ESHealth))
 
 		metrics = append(metrics, Metric{
 			Description: getClusterWriteQuorumMD(),
@@ -3302,6 +3312,11 @@ func getClusterHealthMetrics() *MetricsGroup {
 				"pool": strconv.Itoa(h.PoolID),
 				"set":  strconv.Itoa(h.SetID),
 			}
+			metrics = append(metrics, Metric{
+				Description:    getClusterErasureSetReadQuorumMD(),
+				VariableLabels: labels,
+				Value:          float64(h.ReadQuorum),
+			})
 			metrics = append(metrics, Metric{
 				Description:    getClusterErasureSetWriteQuorumMD(),
 				VariableLabels: labels,

cmd/naughty-disk_test.go

@@ -112,8 +112,11 @@ func (d *naughtyDisk) SetDiskID(id string) {
 	d.disk.SetDiskID(id)
 }
 
-func (d *naughtyDisk) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (info dataUsageCache, err error) {
-	return d.disk.NSScanner(ctx, cache, updates, scanMode)
+func (d *naughtyDisk) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, weSleep func() bool) (info dataUsageCache, err error) {
+	if err := d.calcError(); err != nil {
+		return info, err
+	}
+	return d.disk.NSScanner(ctx, cache, updates, scanMode, weSleep)
 }
 
 func (d *naughtyDisk) DiskInfo(ctx context.Context, metrics bool) (info DiskInfo, err error) {

cmd/object-api-datatypes.go

@@ -612,3 +612,42 @@ type NewMultipartUploadResult struct {
 	UploadID     string
 	ChecksumAlgo string
 }
+
+type getObjectAttributesResponse struct {
+	ETag         string                    `xml:",omitempty"`
+	Checksum     *objectAttributesChecksum `xml:",omitempty"`
+	ObjectParts  *objectAttributesParts    `xml:",omitempty"`
+	StorageClass string                    `xml:",omitempty"`
+	ObjectSize   int64                     `xml:",omitempty"`
+}
+
+type objectAttributesChecksum struct {
+	ChecksumCRC32  string `xml:",omitempty"`
+	ChecksumCRC32C string `xml:",omitempty"`
+	ChecksumSHA1   string `xml:",omitempty"`
+	ChecksumSHA256 string `xml:",omitempty"`
+}
+
+type objectAttributesParts struct {
+	IsTruncated          bool
+	MaxParts             int
+	NextPartNumberMarker int
+	PartNumberMarker     int
+	PartsCount           int
+	Parts                []*objectAttributesPart `xml:"Part"`
+}
+
+type objectAttributesPart struct {
+	PartNumber     int
+	Size           int64
+	ChecksumCRC32  string `xml:",omitempty"`
+	ChecksumCRC32C string `xml:",omitempty"`
+	ChecksumSHA1   string `xml:",omitempty"`
+	ChecksumSHA256 string `xml:",omitempty"`
+}
+
+type objectAttributesErrorResponse struct {
+	ArgumentValue *string `xml:"ArgumentValue,omitempty"`
+	ArgumentName  *string `xml:"ArgumentName"`
+	APIErrorResponse
+}

cmd/object-api-interface.go

@@ -69,6 +69,9 @@ type ObjectOptions struct {
 	Tagging                 bool // Is only in GET/HEAD operations to return tagging metadata along with regular metadata and body.
 
 	UserDefined         map[string]string   // only set in case of POST/PUT operations
+	ObjectAttributes    map[string]struct{} // Attribute tags defined by the users for the GetObjectAttributes request
+	MaxParts            int                 // used in GetObjectAttributes. Signals how many parts we should return
+	PartNumberMarker    int                 // used in GetObjectAttributes. Signals the part number after which results should be returned
 	PartNumber          int                 // only useful in case of GetObject/HeadObject
 	CheckPrecondFn      CheckPreconditionFn // only set during GetObject/HeadObject/CopyObjectPart preconditional valuation
 	EvalMetadataFn      EvalMetadataFn      // only set for retention settings, meant to be used only when updating metadata in-place.

cmd/object-api-options.go

@@ -133,6 +133,121 @@ func getOpts(ctx context.Context, r *http.Request, bucket, object string) (Objec
 	return opts, nil
 }
 
+func getAndValidateAttributesOpts(ctx context.Context, w http.ResponseWriter, r *http.Request, bucket, object string) (opts ObjectOptions, valid bool) {
+	var argumentName string
+	var argumentValue string
+	var apiErr APIError
+	var err error
+	valid = true
+
+	defer func() {
+		if valid {
+			return
+		}
+
+		errResp := objectAttributesErrorResponse{
+			ArgumentName:  &argumentName,
+			ArgumentValue: &argumentValue,
+			APIErrorResponse: getAPIErrorResponse(
+				ctx,
+				apiErr,
+				r.URL.Path,
+				w.Header().Get(xhttp.AmzRequestID),
+				w.Header().Get(xhttp.AmzRequestHostID),
+			),
+		}
+
+		writeResponse(w, apiErr.HTTPStatusCode, encodeResponse(errResp), mimeXML)
+	}()
+
+	opts, err = getOpts(ctx, r, bucket, object)
+	if err != nil {
+		switch vErr := err.(type) {
+		case InvalidVersionID:
+			apiErr = toAPIError(ctx, vErr)
+			argumentName = strings.ToLower("versionId")
+			argumentValue = vErr.VersionID
+		default:
+			apiErr = toAPIError(ctx, vErr)
+		}
+		valid = false
+		return
+	}
+
+	opts.MaxParts, err = parseIntHeader(bucket, object, r.Header, xhttp.AmzMaxParts)
+	if err != nil {
+		apiErr = toAPIError(ctx, err)
+		argumentName = strings.ToLower(xhttp.AmzMaxParts)
+		valid = false
+		return
+	}
+
+	if opts.MaxParts == 0 {
+		opts.MaxParts = maxPartsList
+	}
+
+	opts.PartNumberMarker, err = parseIntHeader(bucket, object, r.Header, xhttp.AmzPartNumberMarker)
+	if err != nil {
+		apiErr = toAPIError(ctx, err)
+		argumentName = strings.ToLower(xhttp.AmzPartNumberMarker)
+		valid = false
+		return
+	}
+
+	opts.ObjectAttributes = parseObjectAttributes(r.Header)
+	if len(opts.ObjectAttributes) < 1 {
+		apiErr = errorCodes.ToAPIErr(ErrInvalidAttributeName)
+		argumentName = strings.ToLower(xhttp.AmzObjectAttributes)
+		valid = false
+		return
+	}
+
+	for tag := range opts.ObjectAttributes {
+		switch tag {
+		case xhttp.ETag:
+		case xhttp.Checksum:
+		case xhttp.StorageClass:
+		case xhttp.ObjectSize:
+		case xhttp.ObjectParts:
+		default:
+			apiErr = errorCodes.ToAPIErr(ErrInvalidAttributeName)
+			argumentName = strings.ToLower(xhttp.AmzObjectAttributes)
+			argumentValue = tag
+			valid = false
+			return
+		}
+	}
+
+	return
+}
+
+func parseObjectAttributes(h http.Header) (attributes map[string]struct{}) {
+	attributes = make(map[string]struct{})
+	for _, v := range strings.Split(strings.TrimSpace(h.Get(xhttp.AmzObjectAttributes)), ",") {
+		if v != "" {
+			attributes[v] = struct{}{}
+		}
+	}
+
+	return
+}
+
+func parseIntHeader(bucket, object string, h http.Header, headerName string) (value int, err error) {
+	stringInt := strings.TrimSpace(h.Get(headerName))
+	if stringInt == "" {
+		return
+	}
+	value, err = strconv.Atoi(stringInt)
+	if err != nil {
+		return 0, InvalidArgument{
+			Bucket: bucket,
+			Object: object,
+			Err:    fmt.Errorf("Unable to parse %s, value should be an integer", headerName),
+		}
+	}
+	return
+}
+
 func parseBoolHeader(bucket, object string, h http.Header, headerName string) (bool, error) {
 	value := strings.TrimSpace(h.Get(headerName))
 	if value != "" {

cmd/object-handlers.go

@@ -689,6 +689,145 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 	})
 }
 
+// GetObjectAttributes ...
+func (api objectAPIHandlers) getObjectAttributesHandler(ctx context.Context, objectAPI ObjectLayer, bucket, object string, w http.ResponseWriter, r *http.Request) {
+	opts, valid := getAndValidateAttributesOpts(ctx, w, r, bucket, object)
+	if !valid {
+		return
+	}
+
+	var s3Error APIErrorCode
+	if opts.VersionID != "" {
+		s3Error = checkRequestAuthType(ctx, r, policy.GetObjectVersionAttributesAction, bucket, object)
+		if s3Error == ErrNone {
+			s3Error = checkRequestAuthType(ctx, r, policy.GetObjectVersionAction, bucket, object)
+		}
+	} else {
+		s3Error = checkRequestAuthType(ctx, r, policy.GetObjectAttributesAction, bucket, object)
+		if s3Error == ErrNone {
+			s3Error = checkRequestAuthType(ctx, r, policy.GetObjectAction, bucket, object)
+		}
+	}
+
+	if s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+
+	objInfo, err := objectAPI.GetObjectInfo(ctx, bucket, object, opts)
+	if err != nil {
+		s3Error = checkRequestAuthType(ctx, r, policy.ListBucketAction, bucket, object)
+		if s3Error == ErrNone {
+			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+			return
+		}
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		return
+	}
+
+	if _, err = DecryptObjectInfo(&objInfo, r); err != nil {
+		writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
+		return
+	}
+
+	if checkPreconditions(ctx, w, r, objInfo, opts) {
+		return
+	}
+
+	OA := new(getObjectAttributesResponse)
+
+	if opts.Versioned {
+		w.Header().Set(xhttp.AmzVersionID, objInfo.VersionID)
+	}
+
+	lastModified := objInfo.ModTime.UTC().Format(http.TimeFormat)
+	w.Header().Set(xhttp.LastModified, lastModified)
+	w.Header().Del(xhttp.ContentType)
+
+	if _, ok := opts.ObjectAttributes[xhttp.Checksum]; ok {
+		chkSums := objInfo.decryptChecksums(0)
+		// AWS does not appear to append part number on this API call.
+		switch {
+		case chkSums["CRC32"] != "":
+			OA.Checksum = new(objectAttributesChecksum)
+			OA.Checksum.ChecksumCRC32 = strings.Split(chkSums["CRC32"], "-")[0]
+		case chkSums["CRC32C"] != "":
+			OA.Checksum = new(objectAttributesChecksum)
+			OA.Checksum.ChecksumCRC32C = strings.Split(chkSums["CRC32C"], "-")[0]
+		case chkSums["SHA256"] != "":
+			OA.Checksum = new(objectAttributesChecksum)
+			OA.Checksum.ChecksumSHA1 = strings.Split(chkSums["SHA1"], "-")[0]
+		case chkSums["SHA1"] != "":
+			OA.Checksum = new(objectAttributesChecksum)
+			OA.Checksum.ChecksumSHA256 = strings.Split(chkSums["SHA256"], "-")[0]
+		}
+
+	}
+
+	if _, ok := opts.ObjectAttributes[xhttp.ETag]; ok {
+		OA.ETag = objInfo.ETag
+	}
+
+	if _, ok := opts.ObjectAttributes[xhttp.ObjectSize]; ok {
+		OA.ObjectSize, _ = objInfo.GetActualSize()
+	}
+
+	if _, ok := opts.ObjectAttributes[xhttp.StorageClass]; ok {
+		OA.StorageClass = objInfo.StorageClass
+	}
+
+	objInfo.decryptPartsChecksums()
+
+	if _, ok := opts.ObjectAttributes[xhttp.ObjectParts]; ok {
+		OA.ObjectParts = new(objectAttributesParts)
+		OA.ObjectParts.PartNumberMarker = opts.PartNumberMarker
+
+		OA.ObjectParts.MaxParts = opts.MaxParts
+		partsLength := len(objInfo.Parts)
+		OA.ObjectParts.PartsCount = partsLength
+
+		if opts.MaxParts > -1 {
+			for i, v := range objInfo.Parts {
+				if v.Number <= opts.PartNumberMarker {
+					continue
+				}
+
+				if len(OA.ObjectParts.Parts) == opts.MaxParts {
+					break
+				}
+
+				OA.ObjectParts.NextPartNumberMarker = v.Number
+				OA.ObjectParts.Parts = append(OA.ObjectParts.Parts, &objectAttributesPart{
+					ChecksumSHA1:   objInfo.Parts[i].Checksums["SHA1"],
+					ChecksumSHA256: objInfo.Parts[i].Checksums["SHA256"],
+					ChecksumCRC32:  objInfo.Parts[i].Checksums["CRC32"],
+					ChecksumCRC32C: objInfo.Parts[i].Checksums["CRC32C"],
+					PartNumber:     objInfo.Parts[i].Number,
+					Size:           objInfo.Parts[i].Size,
+				})
+			}
+		}
+
+		if OA.ObjectParts.NextPartNumberMarker != partsLength {
+			OA.ObjectParts.IsTruncated = true
+		}
+	}
+
+	writeSuccessResponseXML(w, encodeResponse(OA))
+
+	sendEvent(eventArgs{
+		EventName:    event.ObjectAccessedAttributes,
+		BucketName:   bucket,
+		Object:       objInfo,
+		ReqParams:    extractReqParams(r),
+		RespElements: extractRespElements(w),
+		UserAgent:    r.UserAgent(),
+		Host:         handlers.GetSourceIP(r),
+	})
+
+	return
+}
+
 // GetObjectHandler - GET Object
 // ----------
 // This implementation of the GET operation retrieves object. To use GET,
@@ -1043,6 +1182,30 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 	})
 }
 
+// GetObjectAttributesHandles - GET Object
+// -----------
+// This operation retrieves metadata and part metadata from an object without returning the object itself.
+func (api objectAPIHandlers) GetObjectAttributesHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "GetObjectAttributes")
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(ErrServerNotInitialized))
+		return
+	}
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	object, err := unescapePath(vars["object"])
+	if err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	api.getObjectAttributesHandler(ctx, objectAPI, bucket, object, w, r)
+}
+
 // HeadObjectHandler - HEAD Object
 // -----------
 // The HEAD operation retrieves metadata from an object without returning the object itself.

cmd/storage-interface.go

@@ -66,7 +66,7 @@ type StorageAPI interface {
 	// has never been replaced.
 	Healing() *healingTracker
 	DiskInfo(ctx context.Context, metrics bool) (info DiskInfo, err error)
-	NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (dataUsageCache, error)
+	NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, shouldSleep func() bool) (dataUsageCache, error)
 
 	// Volume operations.
 	MakeVol(ctx context.Context, volume string) (err error)
@@ -147,7 +147,7 @@ func (p *unrecognizedDisk) Healing() *healingTracker {
 	return nil
 }
 
-func (p *unrecognizedDisk) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (dataUsageCache, error) {
+func (p *unrecognizedDisk) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, shouldSleep func() bool) (dataUsageCache, error) {
 	return dataUsageCache{}, errDiskNotFound
 }
 

cmd/storage-rest-client.go

@@ -213,7 +213,7 @@ func (client *storageRESTClient) Healing() *healingTracker {
 	return nil
 }
 
-func (client *storageRESTClient) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (dataUsageCache, error) {
+func (client *storageRESTClient) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
 	atomic.AddInt32(&client.scanning, 1)
 	defer atomic.AddInt32(&client.scanning, -1)
 	defer close(updates)

cmd/storage-rest-server.go

@@ -241,7 +241,7 @@ func (s *storageRESTServer) NSScannerHandler(ctx context.Context, params *nsScan
 			out <- resp
 		}
 	}()
-	ui, err := s.getStorage().NSScanner(ctx, *params.Cache, updates, madmin.HealScanMode(params.ScanMode))
+	ui, err := s.getStorage().NSScanner(ctx, *params.Cache, updates, madmin.HealScanMode(params.ScanMode), nil)
 	wg.Wait()
 	if err != nil {
 		return grid.NewRemoteErr(err)

cmd/xl-storage-disk-id-check.go

@@ -274,7 +274,7 @@ func (p *xlStorageDiskIDCheck) Healing() *healingTracker {
 	return p.storage.Healing()
 }
 
-func (p *xlStorageDiskIDCheck) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (dataUsageCache, error) {
+func (p *xlStorageDiskIDCheck) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, _ func() bool) (dataUsageCache, error) {
 	if contextCanceled(ctx) {
 		close(updates)
 		return dataUsageCache{}, ctx.Err()
@@ -284,7 +284,12 @@ func (p *xlStorageDiskIDCheck) NSScanner(ctx context.Context, cache dataUsageCac
 		close(updates)
 		return dataUsageCache{}, err
 	}
-	return p.storage.NSScanner(ctx, cache, updates, scanMode)
+
+	weSleep := func() bool {
+		return scannerIdleMode.Load() == 0
+	}
+
+	return p.storage.NSScanner(ctx, cache, updates, scanMode, weSleep)
 }
 
 func (p *xlStorageDiskIDCheck) GetDiskLoc() (poolIdx, setIdx, diskIdx int) {

cmd/xl-storage-format-v2.go

@@ -1410,15 +1410,13 @@ func (x *xlMetaV2) DeleteVersion(fi FileInfo) (string, error) {
 				err = x.setIdx(i, *ver)
 				return "", err
 			}
-			var err error
 			x.versions = append(x.versions[:i], x.versions[i+1:]...)
 			if fi.MarkDeleted && (fi.VersionPurgeStatus().Empty() || (fi.VersionPurgeStatus() != Complete)) {
 				err = x.addVersion(ventry)
+			} else if fi.Deleted && uv.String() == emptyUUID {
+				return "", x.addVersion(ventry)
 			}
-			// if we remove null version. we should try to add null version to top layer.
-			if uv.String() != emptyUUID {
-				return "", err
-			}
+			return "", err
 		case ObjectType:
 			if updateVersion && !fi.Deleted {
 				ver, err := x.getIdx(i)

cmd/xl-storage.go

@@ -459,7 +459,7 @@ func (s *xlStorage) readMetadata(ctx context.Context, itemPath string) ([]byte,
 	return buf, err
 }
 
-func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode) (dataUsageCache, error) {
+func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry, scanMode madmin.HealScanMode, weSleep func() bool) (dataUsageCache, error) {
 	atomic.AddInt32(&s.scanning, 1)
 	defer atomic.AddInt32(&s.scanning, -1)
 
@@ -663,7 +663,7 @@ func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache, updates
 			return sizeSummary{}, errIgnoreFileContrib
 		}
 		return sizeS, nil
-	}, scanMode)
+	}, scanMode, weSleep)
 	if err != nil {
 		return dataUsageInfo, err
 	}

docs/bucket/versioning/versioning-tests.sh

@@ -0,0 +1,81 @@
+#!/usr/bin/env bash
+
+if [ -n "$TEST_DEBUG" ]; then
+	set -x
+fi
+
+trap 'catch $LINENO' ERR
+
+# shellcheck disable=SC2120
+catch() {
+	if [ $# -ne 0 ]; then
+		echo "error on line $1"
+		echo "$site server logs ========="
+		cat "/tmp/${site}_1.log"
+		echo "==========================="
+		cat "/tmp/${site}_2.log"
+	fi
+
+	echo "Cleaning up instances of MinIO"
+	pkill minio
+	pkill -9 minio
+	rm -rf /tmp/multisitea
+}
+
+catch
+
+set -e
+export MINIO_CI_CD=1
+export MINIO_BROWSER=off
+export MINIO_ROOT_USER="minio"
+export MINIO_ROOT_PASSWORD="minio123"
+export MINIO_KMS_AUTO_ENCRYPTION=off
+export MINIO_PROMETHEUS_AUTH_TYPE=public
+export MINIO_KMS_SECRET_KEY=my-minio-key:OSMM+vkKUTCvQs9YL/CVMIMt43HFhkUpqJxTmGl6rYw=
+unset MINIO_KMS_KES_CERT_FILE
+unset MINIO_KMS_KES_KEY_FILE
+unset MINIO_KMS_KES_ENDPOINT
+unset MINIO_KMS_KES_KEY_NAME
+
+if [ ! -f ./mc ]; then
+	wget -O mc https://dl.minio.io/client/mc/release/linux-amd64/mc &&
+		chmod +x mc
+fi
+
+minio server --address 127.0.0.1:9001 "http://127.0.0.1:9001/tmp/multisitea/data/disterasure/xl{1...4}" \
+	"http://127.0.0.1:9002/tmp/multisitea/data/disterasure/xl{5...8}" >/tmp/sitea_1.log 2>&1 &
+minio server --address 127.0.0.1:9002 "http://127.0.0.1:9001/tmp/multisitea/data/disterasure/xl{1...4}" \
+	"http://127.0.0.1:9002/tmp/multisitea/data/disterasure/xl{5...8}" >/tmp/sitea_2.log 2>&1 &
+
+export MC_HOST_sitea=http://minio:minio123@127.0.0.1:9001
+
+./mc mb sitea/delissue
+
+./mc version enable sitea/delissue
+
+echo hello | ./mc pipe sitea/delissue/hello
+
+./mc version suspend sitea/delissue
+
+./mc rm sitea/delissue/hello
+
+./mc version enable sitea/delissue
+
+echo hello | ./mc pipe sitea/delissue/hello
+
+./mc version suspend sitea/delissue
+
+./mc rm sitea/delissue/hello
+
+count=$(./mc ls --versions sitea/delissue | wc -l)
+
+if [ ${count} -ne 3 ]; then
+	echo "BUG: expected number of versions to be '3' found ${count}"
+	echo "===== DEBUG ====="
+	./mc ls --versions sitea/delissue
+fi
+
+echo "SUCCESS:"
+./mc ls --versions sitea/delissue
+
+catch

docs/metrics/prometheus/list.md

@@ -192,13 +192,20 @@ For deployments with [bucket](https://min.io/docs/minio/linux/administration/buc
 
 ## Information Lifecycle Management (ILM) Metrics
 
-| Name                                               | Description                                                                 |
-|:---------------------------------------------------|:----------------------------------------------------------------------------|
-| `minio_node_ilm_expiry_pending_tasks`              | Number of pending ILM expiry tasks in the queue.                            |
-| `minio_node_ilm_transition_active_tasks`           | Number of active ILM transition tasks.                                      |
-| `minio_node_ilm_transition_pending_tasks`          | Number of pending ILM transition tasks in the queue.                        |
-| `minio_node_ilm_transition_missed_immediate_tasks` | Number of missed immediate ILM transition tasks.                            |
-| `minio_node_ilm_versions_scanned`                  | Total number of object versions checked for ilm actions since server start. |
+| Name                                                         | Description                                                                                                |
+|:-------------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------|
+| `minio_node_ilm_expiry_pending_tasks`                        | Number of pending ILM expiry tasks in the queue.                                                           |
+| `minio_node_ilm_transition_active_tasks`                     | Number of active ILM transition tasks.                                                                     |
+| `minio_node_ilm_transition_pending_tasks`                    | Number of pending ILM transition tasks in the queue.                                                       |
+| `minio_node_ilm_transition_missed_immediate_tasks`           | Number of missed immediate ILM transition tasks.                                                           |
+| `minio_node_ilm_versions_scanned`                            | Total number of object versions checked for ilm actions since server start.                                |
+| `minio_node_ilm_action_count_delete_action`                  | Total action outcome of lifecycle checks since server start for deleting object                            |
+| `minio_node_ilm_action_count_delete_version_action`          | Total action outcome of lifecycle checks since server start for deleting a version                         |
+| `minio_node_ilm_action_count_transition_action`              | Total action outcome of lifecycle checks since server start for transition of an object                    |
+| `minio_node_ilm_action_count_transition_version_action`      | Total action outcome of lifecycle checks since server start for transition of a particular object version  |
+| `minio_node_ilm_action_count_delete_restored_action`         | Total action outcome of lifecycle checks since server start for deletion of temporarily restored object    |
+| `minio_node_ilm_action_count_delete_restored_version_action` | Total action outcome of lifecycle checks since server start for deletion of a temporarily restored version |
+| `minio_node_ilm_action_count_delete_all_versions_action`     | Total action outcome of lifecycle checks since server start for deletion of all versions                   |
 
 ## Tier Metrics
 

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2023-12-23T07-19-11Z
+  image: quay.io/minio/minio:RELEASE.2024-01-01T16-36-33Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

internal/config/api/api.go

@@ -130,9 +130,9 @@ var (
 			Value: "5m",
 		},
 		config.KV{
-			Key:        apiDisableODirect,
-			Value:      "",
-			Deprecated: true,
+			Key:           apiDisableODirect,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
 			Key:   apiODirect,

internal/config/config.go

@@ -274,7 +274,7 @@ type KV struct {
 	Key   string `json:"key"`
 	Value string `json:"value"`
 
-	Deprecated bool `json:"-"`
+	HiddenIfEmpty bool `json:"-"`
 }
 
 func (kv KV) String() string {
@@ -1447,7 +1447,7 @@ func (cs *SubsysInfo) WriteTo(b *strings.Builder, off bool) {
 			continue
 		}
 		// Ignore empty and deprecated values
-		if dkv.Deprecated && kv.Value == "" {
+		if dkv.HiddenIfEmpty && kv.Value == "" {
 			continue
 		}
 		// Do not need to print if state is on

internal/config/scanner/scanner.go

@@ -31,6 +31,9 @@ const (
 	Speed    = "speed"
 	EnvSpeed = "MINIO_SCANNER_SPEED"
 
+	IdleSpeed    = "idle_speed"
+	EnvIdleSpeed = "MINIO_SCANNER_IDLE_SPEED"
+
 	// All below are deprecated in October 2022 and
 	// replaced them with a single speed parameter
 	Delay            = "delay"
@@ -47,6 +50,8 @@ const (
 type Config struct {
 	// Delay is the sleep multiplier.
 	Delay float64 `json:"delay"`
+	// Behavior of the scanner when there is no other parallel S3 requests
+	IdleMode int32 // 0 => throttling, 1 => full speed
 	// MaxWait is maximum wait time between operations
 	MaxWait time.Duration
 	// Cycle is the time.Duration between each scanner cycles
@@ -59,23 +64,28 @@ var DefaultKVS = config.KVS{
 		Key:   Speed,
 		Value: "default",
 	},
-	// Deprecated Oct 2022
 	config.KV{
-		Key:        Delay,
-		Value:      "",
-		Deprecated: true,
+		Key:           IdleSpeed,
+		Value:         "",
+		HiddenIfEmpty: true,
 	},
 	// Deprecated Oct 2022
 	config.KV{
-		Key:        MaxWait,
-		Value:      "",
-		Deprecated: true,
+		Key:           Delay,
+		Value:         "",
+		HiddenIfEmpty: true,
 	},
 	// Deprecated Oct 2022
 	config.KV{
-		Key:        Cycle,
-		Value:      "",
-		Deprecated: true,
+		Key:           MaxWait,
+		Value:         "",
+		HiddenIfEmpty: true,
+	},
+	// Deprecated Oct 2022
+	config.KV{
+		Key:           Cycle,
+		Value:         "",
+		HiddenIfEmpty: true,
 	},
 }
 
@@ -107,6 +117,16 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 	default:
 		return cfg, fmt.Errorf("unknown '%s' value", speed)
 	}
+
+	switch idleSpeed := env.Get(EnvIdleSpeed, kvs.GetWithDefault(IdleSpeed, DefaultKVS)); idleSpeed {
+	case "", "throttled": // Empty is the default mode;
+		cfg.IdleMode = 0
+	case "full":
+		cfg.IdleMode = 1
+	default:
+		return cfg, fmt.Errorf("unknown value: '%s'", idleSpeed)
+	}
+
 	return
 }
 

internal/event/name.go

@@ -36,6 +36,7 @@ const (
 	ObjectAccessedGetRetention
 	ObjectAccessedGetLegalHold
 	ObjectAccessedHead
+	ObjectAccessedAttributes
 	ObjectCreatedCompleteMultipartUpload
 	ObjectCreatedCopy
 	ObjectCreatedPost
@@ -84,7 +85,7 @@ func (name Name) Expand() []Name {
 	case ObjectAccessedAll:
 		return []Name{
 			ObjectAccessedGet, ObjectAccessedHead,
-			ObjectAccessedGetRetention, ObjectAccessedGetLegalHold,
+			ObjectAccessedGetRetention, ObjectAccessedGetLegalHold, ObjectAccessedAttributes,
 		}
 	case ObjectCreatedAll:
 		return []Name{
@@ -162,6 +163,8 @@ func (name Name) String() string {
 		return "s3:ObjectAccessed:GetLegalHold"
 	case ObjectAccessedHead:
 		return "s3:ObjectAccessed:Head"
+	case ObjectAccessedAttributes:
+		return "s3:ObjectAccessed:Attributes"
 	case ObjectCreatedAll:
 		return "s3:ObjectCreated:*"
 	case ObjectCreatedCompleteMultipartUpload:
@@ -278,6 +281,8 @@ func ParseName(s string) (Name, error) {
 		return ObjectAccessedGetLegalHold, nil
 	case "s3:ObjectAccessed:Head":
 		return ObjectAccessedHead, nil
+	case "s3:ObjectAccessed:Attributes":
+		return ObjectAccessedAttributes, nil
 	case "s3:ObjectCreated:*":
 		return ObjectCreatedAll, nil
 	case "s3:ObjectCreated:CompleteMultipartUpload":

internal/event/name_test.go

@@ -31,7 +31,7 @@ func TestNameExpand(t *testing.T) {
 	}{
 		{BucketCreated, []Name{BucketCreated}},
 		{BucketRemoved, []Name{BucketRemoved}},
-		{ObjectAccessedAll, []Name{ObjectAccessedGet, ObjectAccessedHead, ObjectAccessedGetRetention, ObjectAccessedGetLegalHold}},
+		{ObjectAccessedAll, []Name{ObjectAccessedGet, ObjectAccessedHead, ObjectAccessedGetRetention, ObjectAccessedGetLegalHold, ObjectAccessedAttributes}},
 		{ObjectCreatedAll, []Name{
 			ObjectCreatedCompleteMultipartUpload, ObjectCreatedCopy, ObjectCreatedPost, ObjectCreatedPut,
 			ObjectCreatedPutRetention, ObjectCreatedPutLegalHold, ObjectCreatedPutTagging, ObjectCreatedDeleteTagging,

internal/event/rulesmap_test.go

@@ -154,12 +154,12 @@ func TestRulesMapMatch(t *testing.T) {
 
 func TestNewRulesMap(t *testing.T) {
 	rulesMapCase1 := make(RulesMap)
-	rulesMapCase1.add([]Name{ObjectAccessedGet, ObjectAccessedHead, ObjectAccessedGetRetention, ObjectAccessedGetLegalHold},
+	rulesMapCase1.add([]Name{ObjectAccessedGet, ObjectAccessedHead, ObjectAccessedGetRetention, ObjectAccessedGetLegalHold, ObjectAccessedAttributes},
 		"*", TargetID{"1", "webhook"})
 
 	rulesMapCase2 := make(RulesMap)
 	rulesMapCase2.add([]Name{
-		ObjectAccessedGet, ObjectAccessedHead,
+		ObjectAccessedGet, ObjectAccessedHead, ObjectAccessedAttributes,
 		ObjectCreatedPut, ObjectAccessedGetRetention, ObjectAccessedGetLegalHold,
 	}, "*", TargetID{"1", "webhook"})
 

internal/hash/checksum.go

@@ -240,6 +240,49 @@ func ReadCheckSums(b []byte, part int) map[string]string {
 	return res
 }
 
+// ReadPartCheckSums will read all part checksums from b and return them.
+func ReadPartCheckSums(b []byte) (res []map[string]string) {
+	for len(b) > 0 {
+		t, n := binary.Uvarint(b)
+		if n <= 0 {
+			break
+		}
+		b = b[n:]
+
+		typ := ChecksumType(t)
+		length := typ.RawByteLen()
+		if length == 0 || len(b) < length {
+			break
+		}
+		// Skip main checksum
+		b = b[length:]
+		if !typ.Is(ChecksumIncludesMultipart) {
+			continue
+		}
+		parts, n := binary.Uvarint(b)
+		if n <= 0 {
+			break
+		}
+		if len(res) == 0 {
+			res = make([]map[string]string, parts)
+		}
+		b = b[n:]
+		for part := 0; part < int(parts); part++ {
+			if len(b) < length {
+				break
+			}
+			// Read part checksum
+			cs := base64.StdEncoding.EncodeToString(b[:length])
+			b = b[length:]
+			if res[part] == nil {
+				res[part] = make(map[string]string, 1)
+			}
+			res[part][typ.String()] = cs
+		}
+	}
+	return res
+}
+
 // NewChecksumWithType is similar to NewChecksumString but expects input algo of ChecksumType.
 func NewChecksumWithType(alg ChecksumType, value string) *Checksum {
 	if !alg.IsSet() {

internal/http/headers.go

@@ -55,6 +55,12 @@ const (
 	IfMatch           = "If-Match"
 	IfNoneMatch       = "If-None-Match"
 
+	// Request tags used in GetObjectAttributes
+	Checksum     = "Checksum"
+	StorageClass = "StorageClass"
+	ObjectSize   = "ObjectSize"
+	ObjectParts  = "ObjectParts"
+
 	// S3 storage class
 	AmzStorageClass = "x-amz-storage-class"
 
@@ -123,6 +129,11 @@ const (
 	AmzSecurityToken        = "X-Amz-Security-Token"
 	AmzDecodedContentLength = "X-Amz-Decoded-Content-Length"
 	AmzTrailer              = "X-Amz-Trailer"
+	AmzMaxParts             = "X-Amz-Max-Parts"
+	AmzPartNumberMarker     = "X-Amz-Part-Number-Marker"
+
+	// Constants used for GetObjectAttributes and GetObjectVersionAttributes
+	AmzObjectAttributes = "X-Amz-Object-Attributes"
 
 	AmzMetaUnencryptedContentLength = "X-Amz-Meta-X-Amz-Unencrypted-Content-Length"
 	AmzMetaUnencryptedContentMD5    = "X-Amz-Meta-X-Amz-Unencrypted-Content-Md5"